





from transformers import BertTokenizer, EncoderDecoderModel, SeqaSeqTrainer, SeqaSeqTrainingArguments
from transformers.testing_utils import TestCasePlus, require_torch, slow
from transformers.utils import is_datasets_available


if is_datasets_available():
  import datasets







class   __lowerCAmelCase			(					_UpperCamelCase  ):





   @slow
   @require_torch
   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      a__	:						Optional[Any]      		=  EncoderDecoderModel.from_encoder_decoder_pretrained("prajjwal1/bert-tiny"					,			"prajjwal1/bert-tiny"      )
      a__	:						List[Any]      		=  BertTokenizer.from_pretrained("bert-base-uncased"      )

      a__	:						List[Any]      		=  bertabert.config.encoder.vocab_size
      a__	:						Optional[int]      		=  tokenizer.sep_token_id
      a__	:						Optional[Any]      		=  tokenizer.cls_token_id
      a__	:						Any      		=  128

      a__	:						Optional[int]      		=  datasets.load_dataset("cnn_dailymail"					,			"3.0.0"					,			split="train[:1%]"      )
      a__	:						Optional[int]      		=  datasets.load_dataset("cnn_dailymail"					,			"3.0.0"					,			split="validation[:1%]"      )

      a__	:						Tuple      		=  train_dataset.select(range(32      )      )
      a__	:						Optional[Any]      		=  val_dataset.select(range(16      )      )

      a__	:						int      		=  4

      def _map_to_encoder_decoder_inputs(snake_case      ):
         # Tokenizer will automatically set [BOS] <text> [EOS]
         a__	:						List[Any]      		=  tokenizer(batch["article"]					,			padding="max_length"					,			truncation=snake_case					,			max_length=512      )
         a__	:						int      		=  tokenizer(batch["highlights"]					,			padding="max_length"					,			truncation=snake_case					,			max_length=128      )
         a__	:						str      		=  inputs.input_ids
         a__	:						Any      		=  inputs.attention_mask

         a__	:						Tuple      		=  outputs.input_ids
         a__	:						List[str]      		=  outputs.input_ids.copy()
         a__	:						Dict      		=  [
             [-100 if token == tokenizer.pad_token_id else token for token in labels] for labels in batch["labels"]
         ]
         a__	:						int      		=  outputs.attention_mask

         assert all(len(snake_case      ) == 512 for x in inputs.input_ids      )
         assert all(len(snake_case      ) == 128 for x in outputs.input_ids      )

         return batch

      def _compute_metrics(snake_case      ):
         a__	:						Any      		=  pred.label_ids
         a__	:						int      		=  pred.predictions

         # all unnecessary tokens are removed
         a__	:						Tuple      		=  tokenizer.batch_decode(snake_case					,			skip_special_tokens=snake_case      )
         a__	:						Union[str, Any]      		=  tokenizer.batch_decode(snake_case					,			skip_special_tokens=snake_case      )

         a__	:						Tuple      		=  sum([int(pred_str[i] == label_str[i]      ) for i in range(len(snake_case      )      )]      ) / len(snake_case      )

         return {"accuracy": accuracy}

      # map train dataset
      a__	:						Dict      		=  train_dataset.map(
          _map_to_encoder_decoder_inputs					,			batched=snake_case					,			batch_size=snake_case					,			remove_columns=["article", "highlights"]					,			)
      train_dataset.set_format(
          type="torch"					,			columns=["input_ids", "attention_mask", "decoder_input_ids", "decoder_attention_mask", "labels"]					,			)

      # same for validation dataset
      a__	:						Tuple      		=  val_dataset.map(
          _map_to_encoder_decoder_inputs					,			batched=snake_case					,			batch_size=snake_case					,			remove_columns=["article", "highlights"]					,			)
      val_dataset.set_format(
          type="torch"					,			columns=["input_ids", "attention_mask", "decoder_input_ids", "decoder_attention_mask", "labels"]					,			)

      a__	:						Dict      		=  self.get_auto_remove_tmp_dir()

      a__	:						Optional[int]      		=  SeqaSeqTrainingArguments(
          output_dir=snake_case					,			per_device_train_batch_size=snake_case					,			per_device_eval_batch_size=snake_case					,			predict_with_generate=snake_case					,			evaluation_strategy="steps"					,			do_train=snake_case					,			do_eval=snake_case					,			warmup_steps=0					,			eval_steps=2					,			logging_steps=2					,			)

      # instantiate trainer
      a__	:						List[str]      		=  SeqaSeqTrainer(
          model=snake_case					,			args=snake_case					,			compute_metrics=_compute_metrics					,			train_dataset=snake_case					,			eval_dataset=snake_case					,			tokenizer=snake_case					,			)

      # start training
      trainer.train()








import subprocess
import sys

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







class   __lowerCAmelCase			(					_UpperCamelCase  ):





   @require_torch
   def 	_snake_case ( self      )				->  str:
      """simple docstring"""





      a__	:						Tuple      		=  "\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        "

      a__	:						str      		=  "\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__	:						Optional[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
      a__	:						Tuple      		=  "hf-internal-testing/tiny-random-bert"
      BertConfig.from_pretrained(snake_case      )
      BertModel.from_pretrained(snake_case      )
      BertTokenizer.from_pretrained(snake_case      )
      pipeline(task="fill-mask"					,			model=snake_case      )

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

      # should succeed
      a__	:						Dict      		=  self.get_env()
      # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
      a__	:						Tuple      		=  "1"
      a__	:						List[Any]      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )





   @require_torch
   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      a__	:						Tuple      		=  "\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        "

      a__	:						Optional[Any]      		=  "\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__	:						Tuple      		=  "\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__	:						List[Any]      		=  "hf-internal-testing/tiny-random-bert"
      BertConfig.from_pretrained(snake_case      )
      BertModel.from_pretrained(snake_case      )
      BertTokenizer.from_pretrained(snake_case      )
      pipeline(task="fill-mask"					,			model=snake_case      )

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

      # should succeed
      a__	:						Any      		=  self.get_env()
      a__	:						int      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )





   @require_torch
   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      a__	:						List[Any]      		=  "\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        "

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

      a__	:						int      		=  "\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__	:						List[Any]      		=  [sys.executable, "-c", "\n".join([load, run]      )]

      # should succeed
      a__	:						str      		=  self.get_env()
      a__	:						List[str]      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )

      # next emulate no network
      a__	:						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
      a__	:						Union[str, Any]      		=  "1"
      a__	:						Tuple      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )





   @require_torch
   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





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

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





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

      # should succeed
      a__	:						Optional[Any]      		=  self.get_env()
      a__	:						Union[str, Any]      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=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__	:						Dict      		=  "1"
      a__	:						Union[str, Any]      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )

