


import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch





class   __a      (   _lowerCamelCase				):
 @require_torch
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Dict:




        '''simple docstring'''
        lowercase__:       Optional[Any]        =       '\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

        lowercase__:       Dict        =       '\nmname = \"hf-internal-testing/tiny-random-bert\"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task=\"fill-mask\", model=mname)\nprint(\"success\")\n        '

        lowercase__:       Any        =       '\nimport socket\ndef offline_socket(*args, **kwargs): raise RuntimeError(\"Offline mode is enabled, we shouldn\'t access internet\")\nsocket.socket = offline_socket\n        '

        # Force fetching the files so that we can use the cache
        lowercase__:       int        =       'hf-internal-testing/tiny-random-bert'
        BertConfig.from_pretrained(lowerCAmelCase__			)
        BertModel.from_pretrained(lowerCAmelCase__			)
        BertTokenizer.from_pretrained(lowerCAmelCase__			)
        pipeline(task='fill-mask'    ,			model=lowerCAmelCase__			)

        # baseline - just load from_pretrained with normal network
        lowercase__:       Dict        =       [sys.executable, '-c', '\n'.join([load, run, mock]			)]

        # should succeed
        lowercase__:       Dict        =       self.get_env()
        # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
        lowercase__:       Optional[int]        =       '1'
        lowercase__:       Tuple        =       subprocess.run(lowerCAmelCase__    ,			env=lowerCAmelCase__    ,			check=lowerCAmelCase__    ,			capture_output=lowerCAmelCase__			)
        self.assertEqual(result.returncode    ,			0    ,			result.stderr			)
        self.assertIn('success'    ,			result.stdout.decode()			)
 @require_torch
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    int:




        '''simple docstring'''
        lowercase__:       List[str]        =       '\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

        lowercase__:       Optional[int]        =       '\nmname = \"hf-internal-testing/tiny-random-bert\"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task=\"fill-mask\", model=mname)\nprint(\"success\")\n        '

        lowercase__:       Any        =       '\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error(\"Faking flaky internet\")\nsocket.socket = offline_socket\n        '

        # Force fetching the files so that we can use the cache
        lowercase__:       Any        =       'hf-internal-testing/tiny-random-bert'
        BertConfig.from_pretrained(lowerCAmelCase__			)
        BertModel.from_pretrained(lowerCAmelCase__			)
        BertTokenizer.from_pretrained(lowerCAmelCase__			)
        pipeline(task='fill-mask'    ,			model=lowerCAmelCase__			)

        # baseline - just load from_pretrained with normal network
        lowercase__:       Union[str, Any]        =       [sys.executable, '-c', '\n'.join([load, run, mock]			)]

        # should succeed
        lowercase__:       List[str]        =       self.get_env()
        lowercase__:       Dict        =       subprocess.run(lowerCAmelCase__    ,			env=lowerCAmelCase__    ,			check=lowerCAmelCase__    ,			capture_output=lowerCAmelCase__			)
        self.assertEqual(result.returncode    ,			0    ,			result.stderr			)
        self.assertIn('success'    ,			result.stdout.decode()			)
 @require_torch
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    List[str]:




        '''simple docstring'''
        lowercase__:       Tuple        =       '\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

        lowercase__:       Tuple        =       '\nmname = \"hf-internal-testing/tiny-random-bert-sharded\"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint(\"success\")\n        '

        lowercase__:       Dict        =       '\nimport socket\ndef offline_socket(*args, **kwargs): raise ValueError(\"Offline mode is enabled\")\nsocket.socket = offline_socket\n        '

        # baseline - just load from_pretrained with normal network
        lowercase__:       Tuple        =       [sys.executable, '-c', '\n'.join([load, run]			)]

        # should succeed
        lowercase__:       Any        =       self.get_env()
        lowercase__:       Union[str, Any]        =       subprocess.run(lowerCAmelCase__    ,			env=lowerCAmelCase__    ,			check=lowerCAmelCase__    ,			capture_output=lowerCAmelCase__			)
        self.assertEqual(result.returncode    ,			0    ,			result.stderr			)
        self.assertIn('success'    ,			result.stdout.decode()			)

        # next emulate no network
        lowercase__:       List[Any]        =       [sys.executable, '-c', '\n'.join([load, mock, run]			)]

        # Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
        # env["TRANSFORMERS_OFFLINE"] = "0"
        # result = subprocess.run(cmd, env=env, check=False, capture_output=True)
        # self.assertEqual(result.returncode, 1, result.stderr)

        # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
        lowercase__:       List[str]        =       '1'
        lowercase__:       List[str]        =       subprocess.run(lowerCAmelCase__    ,			env=lowerCAmelCase__    ,			check=lowerCAmelCase__    ,			capture_output=lowerCAmelCase__			)
        self.assertEqual(result.returncode    ,			0    ,			result.stderr			)
        self.assertIn('success'    ,			result.stdout.decode()			)
 @require_torch
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Tuple:




        '''simple docstring'''
        lowercase__:       Optional[int]        =       '\nfrom transformers import pipeline\n        '
        lowercase__:       Optional[Any]        =       '\nmname = \"hf-internal-testing/tiny-random-bert\"\npipe = pipeline(model=mname)\n        '

        lowercase__:       List[Any]        =       '\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")\nsocket.socket = offline_socket\n        '
        lowercase__:       Dict        =       self.get_env()
        lowercase__:       Union[str, Any]        =       '1'
        lowercase__:       List[str]        =       [sys.executable, '-c', '\n'.join([load, mock, run]			)]
        lowercase__:       Union[str, Any]        =       subprocess.run(lowerCAmelCase__    ,			env=lowerCAmelCase__    ,			check=lowerCAmelCase__    ,			capture_output=lowerCAmelCase__			)
        self.assertEqual(result.returncode    ,			1    ,			result.stderr			)
        self.assertIn(
            'You cannot infer task automatically within `pipeline` when using offline mode'    ,			result.stderr.decode().replace('\n'    ,			''			)    ,			)
 @require_torch
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Optional[Any]:




        '''simple docstring'''
        lowercase__:       Dict        =       '\nfrom transformers import AutoModel\n        '
        lowercase__:       Union[str, Any]        =       '\nmname = \"hf-internal-testing/test_dynamic_model\"\nAutoModel.from_pretrained(mname, trust_remote_code=True)\nprint(\"success\")\n        '

        # baseline - just load from_pretrained with normal network
        lowercase__:       Any        =       [sys.executable, '-c', '\n'.join([load, run]			)]

        # should succeed
        lowercase__:       List[str]        =       self.get_env()
        lowercase__:       Optional[int]        =       subprocess.run(lowerCAmelCase__    ,			env=lowerCAmelCase__    ,			check=lowerCAmelCase__    ,			capture_output=lowerCAmelCase__			)
        self.assertEqual(result.returncode    ,			0    ,			result.stderr			)
        self.assertIn('success'    ,			result.stdout.decode()			)

        # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
        lowercase__:       Any        =       '1'
        lowercase__:       Optional[int]        =       subprocess.run(lowerCAmelCase__    ,			env=lowerCAmelCase__    ,			check=lowerCAmelCase__    ,			capture_output=lowerCAmelCase__			)
        self.assertEqual(result.returncode    ,			0    ,			result.stderr			)
        self.assertIn('success'    ,			result.stdout.decode()			)





from ...configuration_utils import PretrainedConfig
from ...utils import logging


__lowerCAmelCase			=							logging.get_logger(__name__)

__lowerCAmelCase			=							{
    '''google/pegasus-large''': '''https://huggingface.co/google/pegasus-large/resolve/main/config.json''',
    # See all PEGASUS models at https://huggingface.co/models?filter=pegasus
}





class   __a      (   __UpperCamelCase				):
 __lowercase			:							Any       					=			'pegasus'
 __lowercase			:							Union[str, Any]       					=			['past_key_values']
 __lowercase			:							Any       					=			{'num_attention_heads': 'encoder_attention_heads', 'hidden_size': 'd_model'}
 def __init__(       self    ,			lowerCAmelCase__=50_265    ,			lowerCAmelCase__=1_024    ,			lowerCAmelCase__=12    ,			lowerCAmelCase__=4_096    ,			lowerCAmelCase__=16    ,			lowerCAmelCase__=12    ,			lowerCAmelCase__=4_096    ,			lowerCAmelCase__=16    ,			lowerCAmelCase__=0.0    ,			lowerCAmelCase__=0.0    ,			lowerCAmelCase__=True    ,			lowerCAmelCase__=True    ,			lowerCAmelCase__="gelu"    ,			lowerCAmelCase__=1_024    ,			lowerCAmelCase__=0.1    ,			lowerCAmelCase__=0.0    ,			lowerCAmelCase__=0.0    ,			lowerCAmelCase__=0.0_2    ,			lowerCAmelCase__=0    ,			lowerCAmelCase__=False    ,			lowerCAmelCase__=0    ,			lowerCAmelCase__=1    ,			lowerCAmelCase__=1    ,			**lowerCAmelCase__    ,			) ->    Union[str, Any]:




        '''simple docstring'''
        lowercase__:       int        =       vocab_size
        lowercase__:       Optional[int]        =       max_position_embeddings
        lowercase__:       List[str]        =       d_model
        lowercase__:       Optional[Any]        =       encoder_ffn_dim
        lowercase__:       Optional[Any]        =       encoder_layers
        lowercase__:       Union[str, Any]        =       encoder_attention_heads
        lowercase__:       Optional[int]        =       decoder_ffn_dim
        lowercase__:       Tuple        =       decoder_layers
        lowercase__:       Union[str, Any]        =       decoder_attention_heads
        lowercase__:       Dict        =       dropout
        lowercase__:       List[str]        =       attention_dropout
        lowercase__:       List[str]        =       activation_dropout
        lowercase__:       Optional[int]        =       activation_function
        lowercase__:       Dict        =       init_std
        lowercase__:       Optional[Any]        =       encoder_layerdrop
        lowercase__:       List[str]        =       decoder_layerdrop
        lowercase__:       Union[str, Any]        =       use_cache
        lowercase__:       Any        =       encoder_layers
        lowercase__:       List[str]        =       scale_embedding  # scale factor will be sqrt(d_model) if True
        super().__init__(
            pad_token_id=lowerCAmelCase__    ,			eos_token_id=lowerCAmelCase__    ,			is_encoder_decoder=lowerCAmelCase__    ,			decoder_start_token_id=lowerCAmelCase__    ,			forced_eos_token_id=lowerCAmelCase__    ,			**lowerCAmelCase__    ,			)
 @property
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    int:




        '''simple docstring'''
        return self.encoder_attention_heads
 @property
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    int:




        '''simple docstring'''
        return self.d_model


