




"""simple docstring"""

import unittest
from pathlib import Path
from tempfile import TemporaryDirectory

from transformers import AutoConfig, TFGPTaLMHeadModel, is_keras_nlp_available, is_tf_available
from transformers.models.gpta.tokenization_gpta import GPTaTokenizer
from transformers.testing_utils import require_keras_nlp, require_tf, slow


if is_tf_available():
       import tensorflow as tf

if is_keras_nlp_available():
       from transformers.models.gpta import TFGPTaTokenizer


a					=			["gpt2"]
a					=			"gpt2"

if is_tf_available():





       class 				lowercase_       (     tf.Module							):



        '''simple docstring'''

        def __init__( self       :       Optional[Any]    ,							_UpperCAmelCase       :       Tuple		):
              super().__init__()
              _A					   =   tokenizer
              _A					   =   AutoConfig.from_pretrained(lowerCAmelCase_		)
              _A					   =   TFGPTaLMHeadModel.from_config(lowerCAmelCase_		)



        @tf.function(input_signature=(tf.TensorSpec((None,)    ,							tf.string    ,							name='text'		),)		)
        def 			lowerCAmelCase_							( self       :       List[str]    ,							_UpperCAmelCase       :       Any		):
              _A					   =   self.tokenizer(lowerCAmelCase_		)
              _A					   =   tokenized['input_ids'].to_tensor()

              _A					   =   tf.cast(input_ids_dense > 0    ,							tf.intaa		)
              # input_mask = tf.reshape(input_mask, [-1, MAX_SEQ_LEN])

              _A					   =   self.model(input_ids=lowerCAmelCase_    ,							attention_mask=lowerCAmelCase_		)['logits']

              return outputs





@require_tf
@require_keras_nlp
class 				lowercase_       (     unittest.TestCase							):



 '''simple docstring'''

 def 			lowerCAmelCase_							( self       :       Optional[Any]		):
       super().setUp()

       _A					   =   [GPTaTokenizer.from_pretrained(lowerCAmelCase_		) for checkpoint in (TOKENIZER_CHECKPOINTS)]
       _A					   =   [TFGPTaTokenizer.from_pretrained(lowerCAmelCase_		) for checkpoint in TOKENIZER_CHECKPOINTS]
       assert len(self.tokenizers		) == len(self.tf_tokenizers		)

       _A					   =   [
           'This is a straightforward English test sentence.',
           'This one has some weird characters\rto\nsee\r\nif  those\u00E9break things.',
           'Now we\'re going to add some Chinese: 一 二 三 一二三',
           'And some much more rare Chinese: 齉 堃 齉堃',
           'Je vais aussi écrire en français pour tester les accents',
           'Classical Irish also has some unusual characters, so in they go: Gaelaċ, ꝼ',
       ]
       _A					   =   list(zip(self.test_sentences    ,							self.test_sentences[::-1]		)		)

 def 			lowerCAmelCase_							( self       :       Optional[int]		):






       for tokenizer, tf_tokenizer in zip(self.tokenizers    ,							self.tf_tokenizers		):
             for test_inputs in self.test_sentences:
                   _A					   =   tokenizer([test_inputs]    ,							return_tensors='tf'		)
                   _A					   =   tf_tokenizer([test_inputs]		)

                   for key in python_outputs.keys():
                         # convert them to numpy to avoid messing with ragged tensors
                         _A					   =   python_outputs[key].numpy()
                         _A					   =   tf_outputs[key].numpy()

                         self.assertTrue(tf.reduce_all(python_outputs_values.shape == tf_outputs_values.shape		)		)
                         self.assertTrue(tf.reduce_all(tf.cast(lowerCAmelCase_    ,							tf.intaa		) == tf_outputs_values		)		)

 @slow
 def 			lowerCAmelCase_							( self       :       Tuple		):






       for tf_tokenizer in self.tf_tokenizers:
             _A					   =   tf.function(lowerCAmelCase_		)
             for test_inputs in self.test_sentences:
                   _A					   =   tf.constant(lowerCAmelCase_		)
                   _A					   =   compiled_tokenizer(lowerCAmelCase_		)
                   _A					   =   tf_tokenizer(lowerCAmelCase_		)

                   for key in eager_outputs.keys():
                         self.assertTrue(tf.reduce_all(eager_outputs[key] == compiled_outputs[key]		)		)

 @slow
 def 			lowerCAmelCase_							( self       :       Any		):






       for tf_tokenizer in self.tf_tokenizers:
             _A					   =   ModelToSave(tokenizer=lowerCAmelCase_		)
             _A					   =   tf.convert_to_tensor([self.test_sentences[0]]		)
             _A					   =   model.serving(lowerCAmelCase_		)  # Build model with some sample inputs
             with TemporaryDirectory() as tempdir:
                   _A					   =   Path(lowerCAmelCase_		) / 'saved.model'
                   tf.saved_model.save(lowerCAmelCase_    ,							lowerCAmelCase_    ,							signatures={'serving_default': model.serving}		)
                   _A					   =   tf.saved_model.load(lowerCAmelCase_		)
             _A					   =   loaded_model.signatures['serving_default'](lowerCAmelCase_		)['output_0']
             # We may see small differences because the loaded model is compiled, so we need an epsilon for the test
             self.assertTrue(tf.reduce_all(out == loaded_output		)		)

 @slow
 def 			lowerCAmelCase_							( self       :       List[Any]		):






       for tf_tokenizer in self.tf_tokenizers:
             _A					   =   tf.convert_to_tensor([self.test_sentences[0]]		)
             _A					   =   tf_tokenizer(lowerCAmelCase_		)  # Build model with some sample inputs

             _A					   =   tf_tokenizer.get_config()
             _A					   =   TFGPTaTokenizer.from_config(lowerCAmelCase_		)
             _A					   =   model_from_config(lowerCAmelCase_		)

             for key in from_config_output.keys():
                   self.assertTrue(tf.reduce_all(from_config_output[key] == out[key]		)		)



 @slow
 def 			lowerCAmelCase_							( self       :       Optional[Any]		):






       for tf_tokenizer in self.tf_tokenizers:
             # for the test to run
             _A					   =   123_123

             for max_length in [3, 5, 1_024]:
                   _A					   =   tf.convert_to_tensor([self.test_sentences[0]]		)
                   _A					   =   tf_tokenizer(lowerCAmelCase_    ,							max_length=lowerCAmelCase_		)

                   _A					   =   out['input_ids'].numpy().shape[1]

                   assert out_length == max_length



'''simple docstring'''







import subprocess
import sys

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


class 						__UpperCAmelCase      (					_lowerCamelCase						):


							@require_torch
							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






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

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

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

								# should succeed
								_snake_case   						=		self.get_env()
								# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
								_snake_case   						=		'1'
								_snake_case   						=		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  lowerCamelCase				(       self  ):





								"""simple docstring"""






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

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

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

								# should succeed
								_snake_case   						=		self.get_env()
								_snake_case   						=		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  lowerCamelCase				(       self  ):





								"""simple docstring"""






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

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

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

								# should succeed
								_snake_case   						=		self.get_env()
								_snake_case   						=		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
								_snake_case   						=		[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
								_snake_case   						=		'1'
								_snake_case   						=		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  lowerCamelCase				(       self  ):





								"""simple docstring"""






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

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





								"""simple docstring"""






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

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

