'''simple docstring'''

import subprocess
import sys

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







class 				_lowercase     (				UpperCAmelCase__			):


 '''simple docstring'''






 @require_torch
 def a					(      self   :	Tuple							)  ->    Union[str, Any]:
  # 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
  __lowerCAmelCase											=						"""
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

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

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

  # should succeed
  __lowerCAmelCase											=						self.get_env()
  # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
  __lowerCAmelCase											=						"""1"""
  __lowerCAmelCase											=						subprocess.run(SCREAMING_SNAKE_CASE__      ,       env=SCREAMING_SNAKE_CASE__      ,       check=SCREAMING_SNAKE_CASE__      ,       capture_output=SCREAMING_SNAKE_CASE__							)
  self.assertEqual(result.returncode      ,       0      ,       result.stderr							)
  self.assertIn("""success"""      ,       result.stdout.decode()							)
 @require_torch
 def a					(      self   :	List[str]							)  ->    int:
  # python one-liner segments
  # this must be loaded before socket.socket is monkey-patched
  __lowerCAmelCase											=						"""
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

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

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

  # should succeed
  __lowerCAmelCase											=						self.get_env()
  __lowerCAmelCase											=						subprocess.run(SCREAMING_SNAKE_CASE__      ,       env=SCREAMING_SNAKE_CASE__      ,       check=SCREAMING_SNAKE_CASE__      ,       capture_output=SCREAMING_SNAKE_CASE__							)
  self.assertEqual(result.returncode      ,       0      ,       result.stderr							)
  self.assertIn("""success"""      ,       result.stdout.decode()							)
 @require_torch
 def a					(      self   :	Any							)  ->    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
  __lowerCAmelCase											=						"""
from transformers import BertConfig, BertModel, BertTokenizer
        """

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

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

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

  # next emulate no network
  __lowerCAmelCase											=						[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											=						"""1"""
  __lowerCAmelCase											=						subprocess.run(SCREAMING_SNAKE_CASE__      ,       env=SCREAMING_SNAKE_CASE__      ,       check=SCREAMING_SNAKE_CASE__      ,       capture_output=SCREAMING_SNAKE_CASE__							)
  self.assertEqual(result.returncode      ,       0      ,       result.stderr							)
  self.assertIn("""success"""      ,       result.stdout.decode()							)
 @require_torch
 def a					(      self   :	Dict							)  ->    int:
  __lowerCAmelCase											=						"""
from transformers import pipeline
        """
  __lowerCAmelCase											=						"""
mname = \"hf-internal-testing/tiny-random-bert\"
pipe = pipeline(model=mname)
        """

  __lowerCAmelCase											=						"""
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")
socket.socket = offline_socket
        """
  __lowerCAmelCase											=						self.get_env()
  __lowerCAmelCase											=						"""1"""
  __lowerCAmelCase											=						[sys.executable, """-c""", """\n""".join([load, mock, run]							)]
  __lowerCAmelCase											=						subprocess.run(SCREAMING_SNAKE_CASE__      ,       env=SCREAMING_SNAKE_CASE__      ,       check=SCREAMING_SNAKE_CASE__      ,       capture_output=SCREAMING_SNAKE_CASE__							)
  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   :	str							)  ->    Dict:
  __lowerCAmelCase											=						"""
from transformers import AutoModel
        """
  __lowerCAmelCase											=						"""
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											=						[sys.executable, """-c""", """\n""".join([load, run]							)]

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







'''simple docstring'''

from abc import ABC, abstractmethod
from argparse import ArgumentParser







class 				_lowercase     (				UpperCAmelCase__			):


 '''simple docstring'''






 @staticmethod
 @abstractmethod
 def a					(      SCREAMING_SNAKE_CASE__   :	ArgumentParser							)  ->    Tuple:
  raise NotImplementedError()





 @abstractmethod
 def a					(      self   :	int							)  ->    Union[str, Any]:
  raise NotImplementedError()






