


import unittest

from transformers import GPTSwaTokenizer
from transformers.testing_utils import get_tests_dir, require_sentencepiece, require_tokenizers, slow

from ...test_tokenization_common import TokenizerTesterMixin


UpperCAmelCase_				:			List[Any]   =	get_tests_dir('fixtures/test_sentencepiece_with_bytefallback.model')
@require_sentencepiece
@require_tokenizers
class 			SCREAMING_SNAKE_CASE__  (    __UpperCamelCase						,   unittest.TestCase      ):
      snake_case__    :					List[str]					       =						GPTSwaTokenizer
      snake_case__    :					Dict					       =						False
      snake_case__    :					Optional[Any]					       =						True
      snake_case__    :					List[Any]					       =						False
      def 	SCREAMING_SNAKE_CASE							(	self    :	Tuple					)		->      str:
             super().setUp()

             # We have a SentencePiece fixture for testing
             a_ :		List[Any]							= GPTSwaTokenizer(SCREAMING_SNAKE_CASE__					,							eos_token='<unk>'					,							bos_token='<unk>'					,							pad_token='<unk>'					)

             tokenizer.save_pretrained(self.tmpdirname					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[str]					,							SCREAMING_SNAKE_CASE__    :	Any					)		->      Union[str, Any]:
             a_ :		int							= """This is a test"""
             a_ :		Optional[Any]							= """This is a test"""
             return input_text, output_text
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[Any]					)		->      List[str]:
             a_ :		Optional[Any]							= """<s>"""
             a_ :		int							= 1

             self.assertEqual(self.get_tokenizer()._convert_token_to_id(SCREAMING_SNAKE_CASE__					)					,							SCREAMING_SNAKE_CASE__					)
             self.assertEqual(self.get_tokenizer()._convert_id_to_token(SCREAMING_SNAKE_CASE__					)					,							SCREAMING_SNAKE_CASE__					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      int:
             a_ :		Tuple							= list(self.get_tokenizer().get_vocab().keys()					)

             self.assertEqual(vocab_keys[0]					,							'<unk>'					)
             self.assertEqual(vocab_keys[1]					,							'<s>'					)
             self.assertEqual(vocab_keys[-1]					,							'j'					)
             self.assertEqual(len(SCREAMING_SNAKE_CASE__					)					,							2_0_0_0					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Tuple					)		->      Any:
             self.assertEqual(self.get_tokenizer().vocab_size					,							2_0_0_0					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[str]					)		->      List[Any]:
             a_ :		int							= GPTSwaTokenizer(SCREAMING_SNAKE_CASE__					)

             a_ :		Union[str, Any]							= tokenizer.tokenize('This is a test'					)
             self.assertListEqual(SCREAMING_SNAKE_CASE__					,							['▁This', '▁is', '▁a', '▁t', 'est']					)

             self.assertListEqual(tokenizer.convert_tokens_to_ids(SCREAMING_SNAKE_CASE__					)					,							[4_6_5, 2_8_7, 2_6_5, 6_3_1, 8_4_2]					)

             a_ :		Optional[Any]							= tokenizer.tokenize('I was born in 92000, and this is falsé.'					)
             # fmt: off
             self.assertListEqual(
                 SCREAMING_SNAKE_CASE__					,							['▁I', '▁was', '▁bor', 'n', '▁in', '▁', '<0x39>', '2', '0', '0', '0', ',', '▁and', '▁this', '▁is', '▁f', 'al', 's', '<0xC3>', '<0xA9>', '.']					,							)
             # fmt: on

             a_ :		Optional[int]							= tokenizer.convert_tokens_to_ids(SCREAMING_SNAKE_CASE__					)
             self.assertListEqual(
                 SCREAMING_SNAKE_CASE__					,							[2_6_2, 2_7_2, 1_5_2_5, 2_8_6, 2_7_1, 2_6_8, 6_0, 9_1_6, 6_3_3, 6_3_3, 6_3_3, 2_5_9, 2_6_6, 3_0_1, 2_8_7, 3_8_4, 3_6_7, 2_6_3, 1_9_8, 1_7_2, 2_6_0]					,							)

             a_ :		Optional[int]							= tokenizer.convert_ids_to_tokens(SCREAMING_SNAKE_CASE__					)
             # fmt: off
             self.assertListEqual(
                 SCREAMING_SNAKE_CASE__					,							['▁I', '▁was', '▁bor', 'n', '▁in', '▁', '<0x39>', '2', '0', '0', '0', ',', '▁and', '▁this', '▁is', '▁f', 'al', 's', '<0xC3>', '<0xA9>', '.']					)
             # fmt: on
      def 	SCREAMING_SNAKE_CASE							(	self    :	Dict					)		->      Dict:
             a_ :		str							= GPTSwaTokenizer(SCREAMING_SNAKE_CASE__					)
             a_ :		List[Any]							= ["""This is a test""", """I was born in 92000, and this is falsé."""]
             a_ :		int							= [
                 [4_6_5, 2_8_7, 2_6_5, 6_3_1, 8_4_2],
                 [2_6_2, 2_7_2, 1_5_2_5, 2_8_6, 2_7_1, 2_6_8, 6_0, 9_1_6, 6_3_3, 6_3_3, 6_3_3, 2_5_9, 2_6_6, 3_0_1, 2_8_7, 3_8_4, 3_6_7, 2_6_3, 1_9_8, 1_7_2, 2_6_0],
             ]

             # Test that encode_fast returns the same as tokenize + convert_tokens_to_ids
             for text, expected_ids in zip(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					):
                    self.assertListEqual(tokenizer.encode_fast(SCREAMING_SNAKE_CASE__					)					,							SCREAMING_SNAKE_CASE__					)

             # Test that decode_fast returns the input text
             for text, token_ids in zip(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					):
                    self.assertEqual(tokenizer.decode_fast(SCREAMING_SNAKE_CASE__					)					,							SCREAMING_SNAKE_CASE__					)

      @slow
      def 	SCREAMING_SNAKE_CASE							(	self    :	Any					)		->      Optional[int]:
             a_ :		Any							= [
                 """<|python|>def fibonacci(n)\n    if n < 0:\n        print('Incorrect input')""",
                 """Hey there, how are you doing this fine day?""",
                 """This is a text with a trailing spaces followed by a dot     .""",
                 """Häj sväjs lillebrör! =)""",
                 """Det är inget fel på Mr. Cool""",
             ]

             # fmt: off
             a_ :		Tuple							= {"""input_ids""": [[6_3_4_2_3, 5, 6_8_1_1, 1_4_9_5_4, 2_8_2, 8_1_6, 3_8_2_1, 6_3_4_6_6, 6_3_4_2_5, 6_3_4_6_2, 1_8, 6_3_9_7_8, 6_7_8, 3_0_1, 1_3_2_0, 6_3_4_2_3, 6_3_4_5_5, 6_3_4_5_8, 1_8, 6_3_9_8_2, 4_2_4_6, 3_9_4_0, 1_9_0_1, 4_7_7_8_9, 5_5_4_7, 1_8_9_9_4], [1_9_6_3_0, 1_1_0_0, 6_3_4_4_6, 1_3_4_2, 6_3_3, 5_4_4, 4_4_8_8, 5_9_3, 5_1_0_2, 2_4_1_6, 6_3_4_9_5, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [1_6_5_2, 4_2_8, 2_6_8, 1_9_3_6, 5_1_5, 2_6_8, 5_8_5_9_3, 2_2_4_1_3, 9_1_0_6, 5_4_6, 2_6_8, 3_3_2_1_3, 6_3_9_7_9, 6_9_8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [5_5_1_3_0, 6_3_4_5_0, 9_2_4, 6_3_4_4_9, 2_2_4_9, 4_0_6_2, 1_5_5_8, 3_1_8, 6_3_5_0_4, 2_1_4_9_8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [5_0_9, 3_7_7, 2_8_2_7, 2_5_5_9, 3_3_2, 6_5_7_5, 6_3_4_4_3, 2_6_8_0_1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]], """token_type_ids""": [[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]], """attention_mask""": [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]]}
             # fmt: on
             self.tokenizer_integration_test_util(
                 expected_encoding=SCREAMING_SNAKE_CASE__					,							model_name='AI-Sweden/gpt-sw3-126m'					,							sequences=SCREAMING_SNAKE_CASE__					,							)






import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch
class 			SCREAMING_SNAKE_CASE__  (    lowercase__      ):
      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	Optional[Any]					)		->      List[Any]:
             # this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
             # `transformers` is loaded, and it's too late for inside pytest - so we are changing it
             # while running an external program

             # python one-liner segments

             # this must be loaded before socket.socket is monkey-patched
             a_ :		str							=  '\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

             a_ :		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_ :		Dict							=  '\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_ :		Dict							=  '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_ :		Union[str, Any]							=  [sys.executable, '-c', '\n'.join([load, run, mock]					)]

             # should succeed
             a_ :		List[str]							=  self.get_env()
             # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
             a_ :		Any							=  '1'
             a_ :		Optional[Any]							=  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 	SCREAMING_SNAKE_CASE							(	self    :	str					)		->      List[Any]:
             # python one-liner segments
             # this must be loaded before socket.socket is monkey-patched
             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_ :		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
             a_ :		Tuple							=  '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_ :		Optional[Any]							=  [sys.executable, '-c', '\n'.join([load, run, mock]					)]

             # should succeed
             a_ :		Dict							=  self.get_env()
             a_ :		Dict							=  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 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      Union[str, Any]:
             # this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
             # `transformers` is loaded, and it's too late for inside pytest - so we are changing it
             # while running an external program

             # python one-liner segments

             # this must be loaded before socket.socket is monkey-patched
             a_ :		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_ :		Any							=  '\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_ :		Any							=  [sys.executable, '-c', '\n'.join([load, run]					)]

             # should succeed
             a_ :		Tuple							=  self.get_env()
             a_ :		int							=  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_ :		str							=  [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_ :		Dict							=  '1'
             a_ :		int							=  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 	SCREAMING_SNAKE_CASE							(	self    :	Optional[int]					)		->      Union[str, Any]:
             a_ :		Union[str, Any]							=  '\nfrom transformers import pipeline\n        '
             a_ :		Optional[Any]							=  '\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

             a_ :		Any							=  '\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
             a_ :		Union[str, Any]							=  self.get_env()
             a_ :		Optional[Any]							=  '1'
             a_ :		int							=  [sys.executable, '-c', '\n'.join([load, mock, run]					)]
             a_ :		Union[str, Any]							=  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 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      Dict:
             a_ :		Optional[int]							=  '\nfrom transformers import AutoModel\n        '
             a_ :		Dict							=  '\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_ :		Tuple							=  [sys.executable, '-c', '\n'.join([load, run]					)]

             # should succeed
             a_ :		Optional[Any]							=  self.get_env()
             a_ :		Dict							=  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_ :		Optional[int]							=  '1'
             a_ :		Optional[Any]							=  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()					)


