

'''simple docstring'''
import json
from typing import TYPE_CHECKING, List, Optional, Tuple

from tokenizers import pre_tokenizers

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging


if TYPE_CHECKING:
     from transformers.pipelines.conversational import Conversation


_UpperCAmelCase  :		List[Any]  =		logging.get_logger(__name__)

_UpperCAmelCase  :		int  =		{'''vocab_file''': '''vocab.json''', '''merges_file''': '''merges.txt''', '''tokenizer_file''': '''tokenizer.json'''}

_UpperCAmelCase  :		List[str]  =		{
    '''tokenizer_file''': {
        '''EleutherAI/gpt-neox-20b''': '''https://huggingface.co/EleutherAI/gpt-neox-20b/resolve/main/tokenizer.json''',
    },
}

_UpperCAmelCase  :		int  =		{
    '''gpt-neox-20b''': 20_48,
}

class 						__magic_name__   (							__SCREAMING_SNAKE_CASE ):
 UpperCamelCase__     =       VOCAB_FILES_NAMES
 UpperCamelCase__     =       PRETRAINED_VOCAB_FILES_MAP
 UpperCamelCase__     =       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 UpperCamelCase__     =       ['input_ids', 'attention_mask']







 def __init__(       self  , snake_case_=None  , snake_case_=None  , snake_case_=None  , snake_case_="<|endoftext|>"  , snake_case_="<|endoftext|>"  , snake_case_="<|endoftext|>"  , snake_case_=False  , **snake_case_  , ):
   super().__init__(
       snake_case_  , snake_case_  , tokenizer_file=snake_case_  , unk_token=snake_case_  , bos_token=snake_case_  , eos_token=snake_case_  , add_prefix_space=snake_case_  , **snake_case_  , )

   lowercase									=json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()   )
   if pre_tok_state.get('''add_prefix_space'''  , snake_case_   ) != add_prefix_space:
     lowercase									=getattr(snake_case_  , pre_tok_state.pop('''type'''   )   )
     lowercase									=add_prefix_space
     lowercase									=pre_tok_class(**snake_case_   )

   lowercase									=add_prefix_space







 def 						_A(       self  , snake_case_  , snake_case_ = None   ):
   lowercase									=self._tokenizer.model.save(snake_case_  , name=snake_case_   )
   return tuple(snake_case_   )







 def 						_A(       self  , snake_case_   ):
   lowercase									=[]
   for is_user, text in conversation.iter_texts():
     input_ids.extend(self.encode(snake_case_  , add_special_tokens=snake_case_   ) + [self.eos_token_id]   )

   if len(snake_case_   ) > self.model_max_length:
     lowercase									=input_ids[-self.model_max_length :]
   return input_ids






import subprocess
import sys

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





class 			lowerCamelCase_				( _SCREAMING_SNAKE_CASE   ):


 '''simple docstring'''







 @require_torch
 def  lowerCAmelCase_    (				self    :						int		):
       # 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
       SCREAMING_SNAKE_CASE_      		=     '\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

       SCREAMING_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        '

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

       # should succeed
       SCREAMING_SNAKE_CASE_      		=     self.get_env()
       # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
       SCREAMING_SNAKE_CASE_      		=     '1'
       SCREAMING_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    :						Tuple		):
       # python one-liner segments
       # this must be loaded before socket.socket is monkey-patched
       SCREAMING_SNAKE_CASE_      		=     '\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

       SCREAMING_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        '

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

       # should succeed
       SCREAMING_SNAKE_CASE_      		=     self.get_env()
       SCREAMING_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    :						List[str]		):
       # 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
       SCREAMING_SNAKE_CASE_      		=     '\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

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

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

       # should succeed
       SCREAMING_SNAKE_CASE_      		=     self.get_env()
       SCREAMING_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
       SCREAMING_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
       SCREAMING_SNAKE_CASE_      		=     '1'
       SCREAMING_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    :						Any		):
       SCREAMING_SNAKE_CASE_      		=     '\nfrom transformers import pipeline\n        '
       SCREAMING_SNAKE_CASE_      		=     '\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

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

       # should succeed
       SCREAMING_SNAKE_CASE_      		=     self.get_env()
       SCREAMING_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
       SCREAMING_SNAKE_CASE_      		=     '1'
       SCREAMING_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()		)