




import subprocess
import sys

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




class   a__			(       UpperCamelCase_  ):





						@require_torch
						def       __UpperCamelCase     (							self      :	Tuple)						->       Optional[int]:




											"""simple docstring"""







											_lowerCAmelCase:Optional[Any]									=  '''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

											_lowerCAmelCase:List[Any]									=  '''
mname = "hf-internal-testing/tiny-random-bert"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task="fill-mask", model=mname)
print("success")
        '''

											_lowerCAmelCase:Dict									=  '''
import socket
def offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")
socket.socket = offline_socket
        '''

											# Force fetching the files so that we can use the cache
											_lowerCAmelCase:Optional[int]									=  '''hf-internal-testing/tiny-random-bert'''
											BertConfig.from_pretrained(a__)
											BertModel.from_pretrained(a__)
											BertTokenizer.from_pretrained(a__)
											pipeline(task='''fill-mask'''     ,model=a__)

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

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





						@require_torch
						def       __UpperCamelCase     (							self      :	List[str])						->       Dict:




											"""simple docstring"""







											_lowerCAmelCase:int									=  '''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

											_lowerCAmelCase:Dict									=  '''
mname = "hf-internal-testing/tiny-random-bert"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task="fill-mask", model=mname)
print("success")
        '''

											_lowerCAmelCase:Tuple									=  '''
import socket
def offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")
socket.socket = offline_socket
        '''

											# Force fetching the files so that we can use the cache
											_lowerCAmelCase:Union[str, Any]									=  '''hf-internal-testing/tiny-random-bert'''
											BertConfig.from_pretrained(a__)
											BertModel.from_pretrained(a__)
											BertTokenizer.from_pretrained(a__)
											pipeline(task='''fill-mask'''     ,model=a__)

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

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





						@require_torch
						def       __UpperCamelCase     (							self      :	str)						->       Optional[int]:




											"""simple docstring"""







											_lowerCAmelCase:List[Any]									=  '''
from transformers import BertConfig, BertModel, BertTokenizer
        '''

											_lowerCAmelCase:Dict									=  '''
mname = "hf-internal-testing/tiny-random-bert-sharded"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print("success")
        '''

											_lowerCAmelCase:Optional[int]									=  '''
import socket
def offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")
socket.socket = offline_socket
        '''

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

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

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





						@require_torch
						def       __UpperCamelCase     (							self      :	Any)						->       Optional[Any]:




											"""simple docstring"""







											_lowerCAmelCase:Optional[Any]									=  '''
from transformers import pipeline
        '''
											_lowerCAmelCase:Tuple									=  '''
mname = "hf-internal-testing/tiny-random-bert"
pipe = pipeline(model=mname)
        '''

											_lowerCAmelCase:List[str]									=  '''
import socket
def offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")
socket.socket = offline_socket
        '''
											_lowerCAmelCase:Union[str, Any]									=  self.get_env()
											_lowerCAmelCase:Dict									=  '''1'''
											_lowerCAmelCase:Optional[int]									=  [sys.executable, '''-c''', '''\n'''.join([load, mock, run])]
											_lowerCAmelCase:Tuple									=  subprocess.run(a__     ,env=a__     ,check=a__     ,capture_output=a__)
											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       __UpperCamelCase     (							self      :	int)						->       Optional[Any]:




											"""simple docstring"""







											_lowerCAmelCase:int									=  '''
from transformers import AutoModel
        '''
											_lowerCAmelCase:int									=  '''
mname = "hf-internal-testing/test_dynamic_model"
AutoModel.from_pretrained(mname, trust_remote_code=True)
print("success")
        '''

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

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












"""simple docstring"""






import baseaa





def 							UpperCAmelCase		(		snake_case			:					str    ):
					return baseaa.aaaencode(string.encode('''utf-8'''    )    )





def 							UpperCAmelCase		(		snake_case			:					bytes    ):
					return baseaa.aaadecode(snake_case    ).decode('''utf-8'''    )


if __name__ == "__main__":
				import doctest

				doctest.testmod()






