
"""simple docstring"""




import subprocess
import sys

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







class     lowercase(_lowercase       ):



     @require_torch
     def 			lowercase__  (							self   )		->     Dict:







         """simple docstring"""



         a__		=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

         a__		=			'\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        '

         a__		=			'\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
         a__		=			'hf-internal-testing/tiny-random-bert'
         BertConfig.from_pretrained(__SCREAMING_SNAKE_CASE   )
         BertModel.from_pretrained(__SCREAMING_SNAKE_CASE   )
         BertTokenizer.from_pretrained(__SCREAMING_SNAKE_CASE   )
         pipeline(task='fill-mask'     ,				model=__SCREAMING_SNAKE_CASE   )

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

         # should succeed
         a__		=			self.get_env()
         # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
         a__		=			'1'
         a__		=			subprocess.run(__SCREAMING_SNAKE_CASE     ,				env=__SCREAMING_SNAKE_CASE     ,				check=__SCREAMING_SNAKE_CASE     ,				capture_output=__SCREAMING_SNAKE_CASE   )
         self.assertEqual(result.returncode     ,				0     ,				result.stderr   )
         self.assertIn('success'     ,				result.stdout.decode()   )



     @require_torch
     def 			lowercase__  (							self   )		->     Any:







         """simple docstring"""



         a__		=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

         a__		=			'\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        '

         a__		=			'\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
         a__		=			'hf-internal-testing/tiny-random-bert'
         BertConfig.from_pretrained(__SCREAMING_SNAKE_CASE   )
         BertModel.from_pretrained(__SCREAMING_SNAKE_CASE   )
         BertTokenizer.from_pretrained(__SCREAMING_SNAKE_CASE   )
         pipeline(task='fill-mask'     ,				model=__SCREAMING_SNAKE_CASE   )

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

         # should succeed
         a__		=			self.get_env()
         a__		=			subprocess.run(__SCREAMING_SNAKE_CASE     ,				env=__SCREAMING_SNAKE_CASE     ,				check=__SCREAMING_SNAKE_CASE     ,				capture_output=__SCREAMING_SNAKE_CASE   )
         self.assertEqual(result.returncode     ,				0     ,				result.stderr   )
         self.assertIn('success'     ,				result.stdout.decode()   )



     @require_torch
     def 			lowercase__  (							self   )		->     Union[str, Any]:







         """simple docstring"""



         a__		=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

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

         a__		=			'\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
         a__		=			[sys.executable, '-c', '\n'.join([load, run]   )]

         # should succeed
         a__		=			self.get_env()
         a__		=			subprocess.run(__SCREAMING_SNAKE_CASE     ,				env=__SCREAMING_SNAKE_CASE     ,				check=__SCREAMING_SNAKE_CASE     ,				capture_output=__SCREAMING_SNAKE_CASE   )
         self.assertEqual(result.returncode     ,				0     ,				result.stderr   )
         self.assertIn('success'     ,				result.stdout.decode()   )

         # next emulate no network
         a__		=			[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
         a__		=			'1'
         a__		=			subprocess.run(__SCREAMING_SNAKE_CASE     ,				env=__SCREAMING_SNAKE_CASE     ,				check=__SCREAMING_SNAKE_CASE     ,				capture_output=__SCREAMING_SNAKE_CASE   )
         self.assertEqual(result.returncode     ,				0     ,				result.stderr   )
         self.assertIn('success'     ,				result.stdout.decode()   )



     @require_torch
     def 			lowercase__  (							self   )		->     Any:







         """simple docstring"""



         a__		=			'\nfrom transformers import pipeline\n        '
         a__		=			'\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

         a__		=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
         a__		=			self.get_env()
         a__		=			'1'
         a__		=			[sys.executable, '-c', '\n'.join([load, mock, run]   )]
         a__		=			subprocess.run(__SCREAMING_SNAKE_CASE     ,				env=__SCREAMING_SNAKE_CASE     ,				check=__SCREAMING_SNAKE_CASE     ,				capture_output=__SCREAMING_SNAKE_CASE   )
         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 			lowercase__  (							self   )		->     List[str]:







         """simple docstring"""



         a__		=			'\nfrom transformers import AutoModel\n        '
         a__		=			'\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
         a__		=			[sys.executable, '-c', '\n'.join([load, run]   )]

         # should succeed
         a__		=			self.get_env()
         a__		=			subprocess.run(__SCREAMING_SNAKE_CASE     ,				env=__SCREAMING_SNAKE_CASE     ,				check=__SCREAMING_SNAKE_CASE     ,				capture_output=__SCREAMING_SNAKE_CASE   )
         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
         a__		=			'1'
         a__		=			subprocess.run(__SCREAMING_SNAKE_CASE     ,				env=__SCREAMING_SNAKE_CASE     ,				check=__SCREAMING_SNAKE_CASE     ,				capture_output=__SCREAMING_SNAKE_CASE   )
         self.assertEqual(result.returncode     ,				0     ,				result.stderr   )
         self.assertIn('success'     ,				result.stdout.decode()   )




"""simple docstring"""




import torch
import torch.nn as nn
from transformers.modeling_utils import ModuleUtilsMixin
from transformers.models.ta.modeling_ta import TaBlock, TaConfig, TaLayerNorm

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin







class     lowercase(_lowercase    ,						_lowercase    ,						_lowercase       ):



     @register_to_config
     def __init__(							self     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE = False     ,				)		->     Union[str, Any]:







         """simple docstring"""



         super().__init__()

         a__		=			nn.Embedding(__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )

         a__		=			nn.Embedding(__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )
         a__		=			False

         a__		=			nn.Dropout(p=__SCREAMING_SNAKE_CASE   )

         a__		=			TaConfig(
             vocab_size=__SCREAMING_SNAKE_CASE     ,				d_model=__SCREAMING_SNAKE_CASE     ,				num_heads=__SCREAMING_SNAKE_CASE     ,				d_kv=__SCREAMING_SNAKE_CASE     ,				d_ff=__SCREAMING_SNAKE_CASE     ,				dropout_rate=__SCREAMING_SNAKE_CASE     ,				feed_forward_proj=__SCREAMING_SNAKE_CASE     ,				is_decoder=__SCREAMING_SNAKE_CASE     ,				is_encoder_decoder=__SCREAMING_SNAKE_CASE     ,				)

         a__		=			nn.ModuleList()
         for lyr_num in range(__SCREAMING_SNAKE_CASE   ):
             a__		=			TaBlock(__SCREAMING_SNAKE_CASE   )
             self.encoders.append(__SCREAMING_SNAKE_CASE   )

         a__		=			TaLayerNorm(__SCREAMING_SNAKE_CASE   )
         a__		=			nn.Dropout(p=__SCREAMING_SNAKE_CASE   )




     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )		->     List[Any]:







         """simple docstring"""



         a__		=			self.token_embedder(__SCREAMING_SNAKE_CASE   )

         a__		=			encoder_input_tokens.shape[1]
         a__		=			torch.arange(__SCREAMING_SNAKE_CASE     ,				device=encoder_input_tokens.device   )
         x += self.position_encoding(__SCREAMING_SNAKE_CASE   )

         a__		=			self.dropout_pre(__SCREAMING_SNAKE_CASE   )

         # inverted the attention mask
         a__		=			encoder_input_tokens.size()
         a__		=			self.get_extended_attention_mask(__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )

         for lyr in self.encoders:
             a__		=			lyr(__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )[0]
         a__		=			self.layer_norm(__SCREAMING_SNAKE_CASE   )

         return self.dropout_post(__SCREAMING_SNAKE_CASE   ), encoder_inputs_mask


