



import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch
class 			_lowerCAmelCase   ( __A			):






			"""simple docstring"""





			@require_torch
			def 		snake_case__			(				self	:       Tuple   )						->					Optional[int]:
									"""simple docstring"""







									__lowercase					=				'''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

									__lowercase					=				'''
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")
        '''

									__lowercase					=				'''
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
									__lowercase					=				'''hf-internal-testing/tiny-random-bert'''
									BertConfig.from_pretrained(lowercase   )
									BertModel.from_pretrained(lowercase   )
									BertTokenizer.from_pretrained(lowercase   )
									pipeline(task="""fill-mask"""			,     model=lowercase   )

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

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





			@require_torch
			def 		snake_case__			(				self	:       str   )						->					List[str]:
									"""simple docstring"""







									__lowercase					=				'''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

									__lowercase					=				'''
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")
        '''

									__lowercase					=				'''
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
									__lowercase					=				'''hf-internal-testing/tiny-random-bert'''
									BertConfig.from_pretrained(lowercase   )
									BertModel.from_pretrained(lowercase   )
									BertTokenizer.from_pretrained(lowercase   )
									pipeline(task="""fill-mask"""			,     model=lowercase   )

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

									# should succeed
									__lowercase					=				self.get_env()
									__lowercase					=				subprocess.run(lowercase			,     env=lowercase			,     check=lowercase			,     capture_output=lowercase   )
									self.assertEqual(result.returncode			,     0			,     result.stderr   )
									self.assertIn("""success"""			,     result.stdout.decode()   )





			@require_torch
			def 		snake_case__			(				self	:       Any   )						->					Optional[Any]:
									"""simple docstring"""







									__lowercase					=				'''
from transformers import BertConfig, BertModel, BertTokenizer
        '''

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

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

									# should succeed
									__lowercase					=				self.get_env()
									__lowercase					=				subprocess.run(lowercase			,     env=lowercase			,     check=lowercase			,     capture_output=lowercase   )
									self.assertEqual(result.returncode			,     0			,     result.stderr   )
									self.assertIn("""success"""			,     result.stdout.decode()   )

									# next emulate no network
									__lowercase					=				[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
									__lowercase					=				'''1'''
									__lowercase					=				subprocess.run(lowercase			,     env=lowercase			,     check=lowercase			,     capture_output=lowercase   )
									self.assertEqual(result.returncode			,     0			,     result.stderr   )
									self.assertIn("""success"""			,     result.stdout.decode()   )





			@require_torch
			def 		snake_case__			(				self	:       Tuple   )						->					Optional[int]:
									"""simple docstring"""







									__lowercase					=				'''
from transformers import pipeline
        '''
									__lowercase					=				'''
mname = "hf-internal-testing/tiny-random-bert"
pipe = pipeline(model=mname)
        '''

									__lowercase					=				'''
import socket
def offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")
socket.socket = offline_socket
        '''
									__lowercase					=				self.get_env()
									__lowercase					=				'''1'''
									__lowercase					=				[sys.executable, '''-c''', '''\n'''.join([load, mock, run]   )]
									__lowercase					=				subprocess.run(lowercase			,     env=lowercase			,     check=lowercase			,     capture_output=lowercase   )
									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	:       int   )						->					Optional[int]:
									"""simple docstring"""







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

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








from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tf_available, is_torch_available


UpperCamelCase__ =						{
    "configuration_ctrl": ["CTRL_PRETRAINED_CONFIG_ARCHIVE_MAP", "CTRLConfig"],
    "tokenization_ctrl": ["CTRLTokenizer"],
}

try:
			if not is_torch_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			UpperCamelCase__ =						[
			    "CTRL_PRETRAINED_MODEL_ARCHIVE_LIST",
			    "CTRLForSequenceClassification",
			    "CTRLLMHeadModel",
			    "CTRLModel",
			    "CTRLPreTrainedModel",
			]

try:
			if not is_tf_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			UpperCamelCase__ =						[
			    "TF_CTRL_PRETRAINED_MODEL_ARCHIVE_LIST",
			    "TFCTRLForSequenceClassification",
			    "TFCTRLLMHeadModel",
			    "TFCTRLModel",
			    "TFCTRLPreTrainedModel",
			]


if TYPE_CHECKING:
			from .configuration_ctrl import CTRL_PRETRAINED_CONFIG_ARCHIVE_MAP, CTRLConfig
			from .tokenization_ctrl import CTRLTokenizer

			try:
						if not is_torch_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_ctrl import (
						    CTRL_PRETRAINED_MODEL_ARCHIVE_LIST,
						    CTRLForSequenceClassification,
						    CTRLLMHeadModel,
						    CTRLModel,
						    CTRLPreTrainedModel,
						)

			try:
						if not is_tf_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_tf_ctrl import (
						    TF_CTRL_PRETRAINED_MODEL_ARCHIVE_LIST,
						    TFCTRLForSequenceClassification,
						    TFCTRLLMHeadModel,
						    TFCTRLModel,
						    TFCTRLPreTrainedModel,
						)

else:
			import sys

			UpperCamelCase__ =						_LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)





