


'''simple docstring'''



import subprocess
import sys

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


class 		_snake_case					(     a__	):




					@require_torch
					def 				snake_case__						(      self):
								# 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
								UpperCAmelCase__  :					List[Any]        =     """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

								UpperCAmelCase__  :					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\")
        """

								UpperCAmelCase__  :					List[str]        =     """
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
								UpperCAmelCase__  :					Optional[int]        =     """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
								UpperCAmelCase__  :					List[Any]        =     [sys.executable, """-c""", """\n""".join([load, run, mock])]

								# should succeed
								UpperCAmelCase__  :					Optional[int]        =     self.get_env()
								# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
								UpperCAmelCase__  :					int        =     """1"""
								UpperCAmelCase__  :					Optional[int]        =     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 				snake_case__						(      self):
								# python one-liner segments
								# this must be loaded before socket.socket is monkey-patched
								UpperCAmelCase__  :					Tuple        =     """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

								UpperCAmelCase__  :					str        =     """
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\")
        """

								UpperCAmelCase__  :					Optional[Any]        =     """
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
								UpperCAmelCase__  :					List[Any]        =     """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
								UpperCAmelCase__  :					Union[str, Any]        =     [sys.executable, """-c""", """\n""".join([load, run, mock])]

								# should succeed
								UpperCAmelCase__  :					Any        =     self.get_env()
								UpperCAmelCase__  :					Optional[int]        =     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 				snake_case__						(      self):
								# 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
								UpperCAmelCase__  :					Optional[int]        =     """
from transformers import BertConfig, BertModel, BertTokenizer
        """

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

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

								# should succeed
								UpperCAmelCase__  :					Union[str, Any]        =     self.get_env()
								UpperCAmelCase__  :					int        =     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
								UpperCAmelCase__  :					List[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
								UpperCAmelCase__  :					Tuple        =     """1"""
								UpperCAmelCase__  :					str        =     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 				snake_case__						(      self):
								UpperCAmelCase__  :					int        =     """
from transformers import pipeline
        """
								UpperCAmelCase__  :					List[Any]        =     """
mname = \"hf-internal-testing/tiny-random-bert\"
pipe = pipeline(model=mname)
        """

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

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



'''simple docstring'''



import warnings
from functools import wraps
from typing import Callable







def     _UpperCamelCase      (		UpperCamelCase__					):

			@wraps(UpperCamelCase__					)
			def _inner_fn(*UpperCamelCase__ ,      **UpperCamelCase__					):
						warnings.warn(
						    (f'''\'{fn.__name__}\' is experimental and might be subject to breaking changes in the future.''') ,      UpperCamelCase__ ,      )
						return fn(*UpperCamelCase__ ,      **UpperCamelCase__					)

			return _inner_fn