





import subprocess
import sys

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


class     A     (SCREAMING_SNAKE_CASE       ):
			'''simple docstring'''




			@require_torch
			def   a_			(     self  :       Tuple  )      ->      Union[str, Any]:







					"""simple docstring"""






					A__     =  """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

					A__     =  """
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\")
        """

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

					# should succeed
					A__     =  self.get_env()
					# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
					A__     =  """1"""
					A__     =  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   a_			(     self  :       Union[str, Any]  )      ->      List[str]:







					"""simple docstring"""






					A__     =  """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

					A__     =  """
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\")
        """

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

					# should succeed
					A__     =  self.get_env()
					A__     =  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   a_			(     self  :       Dict  )      ->      Tuple:







					"""simple docstring"""






					A__     =  """
from transformers import BertConfig, BertModel, BertTokenizer
        """

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

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

					# should succeed
					A__     =  self.get_env()
					A__     =  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
					A__     =  [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__     =  """1"""
					A__     =  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   a_			(     self  :       Any  )      ->      str:







					"""simple docstring"""






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

					A__     =  """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")
socket.socket = offline_socket
        """
					A__     =  self.get_env()
					A__     =  """1"""
					A__     =  [sys.executable, """-c""", """\n""".join([load, mock, run]  )]
					A__     =  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   a_			(     self  :       Optional[Any]  )      ->      Optional[int]:







					"""simple docstring"""






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

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










def   __lowerCamelCase   (				__a  :str       )						->							bool:






		"""simple docstring"""

		A__     =  0
		for ch in input_str:
				A__     =  ord(__a       )
				A__     =  pow(2       ,	__a       )

				# If we already turned on bit for current character's unicode
				if bitmap >> ch_unicode & 1 == 1:
						return False
				bitmap |= ch_bit_index_on
		return True


if __name__ == "__main__":
					import doctest

					doctest.testmod()



