

import subprocess
import sys

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



class 					a						(       __A				):



    """simple docstring"""



    @require_torch
    def 				UpperCamelCase	(     self: Tuple       ):






         """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(UpperCamelCase       )
         BertModel.from_pretrained(UpperCamelCase       )
         BertTokenizer.from_pretrained(UpperCamelCase       )
         pipeline(task="""fill-mask"""    ,						model=UpperCamelCase       )

         # 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(UpperCamelCase    ,						env=UpperCamelCase    ,						check=UpperCamelCase    ,						capture_output=UpperCamelCase       )
         self.assertEqual(result.returncode    ,						0    ,						result.stderr       )
         self.assertIn("""success"""    ,						result.stdout.decode()       )

    @require_torch
    def 				UpperCamelCase	(     self: 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(UpperCamelCase       )
         BertModel.from_pretrained(UpperCamelCase       )
         BertTokenizer.from_pretrained(UpperCamelCase       )
         pipeline(task="""fill-mask"""    ,						model=UpperCamelCase       )

         # 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(UpperCamelCase    ,						env=UpperCamelCase    ,						check=UpperCamelCase    ,						capture_output=UpperCamelCase       )
         self.assertEqual(result.returncode    ,						0    ,						result.stderr       )
         self.assertIn("""success"""    ,						result.stdout.decode()       )

    @require_torch
    def 				UpperCamelCase	(     self: Any       ):






         """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(UpperCamelCase    ,						env=UpperCamelCase    ,						check=UpperCamelCase    ,						capture_output=UpperCamelCase       )
         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(UpperCamelCase    ,						env=UpperCamelCase    ,						check=UpperCamelCase    ,						capture_output=UpperCamelCase       )
         self.assertEqual(result.returncode    ,						0    ,						result.stderr       )
         self.assertIn("""success"""    ,						result.stdout.decode()       )

    @require_torch
    def 				UpperCamelCase	(     self: Tuple       ):






         """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(UpperCamelCase    ,						env=UpperCamelCase    ,						check=UpperCamelCase    ,						capture_output=UpperCamelCase       )
         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       ):






         """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(UpperCamelCase    ,						env=UpperCamelCase    ,						check=UpperCamelCase    ,						capture_output=UpperCamelCase       )
         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(UpperCamelCase    ,						env=UpperCamelCase    ,						check=UpperCamelCase    ,						capture_output=UpperCamelCase       )
         self.assertEqual(result.returncode    ,						0    ,						result.stderr       )
         self.assertIn("""success"""    ,						result.stdout.decode()       )









"""simple docstring"""



from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


SCREAMING_SNAKE_CASE_					:	Any							  =						{
    'configuration_jukebox': [
        'JUKEBOX_PRETRAINED_CONFIG_ARCHIVE_MAP',
        'JukeboxConfig',
        'JukeboxPriorConfig',
        'JukeboxVQVAEConfig',
    ],
    'tokenization_jukebox': ['JukeboxTokenizer'],
}

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 SCREAMING_SNAKE_CASE_					:	Union[str, Any]							  =						[
     'JUKEBOX_PRETRAINED_MODEL_ARCHIVE_LIST',
     'JukeboxModel',
     'JukeboxPreTrainedModel',
     'JukeboxVQVAE',
     'JukeboxPrior',
 ]

if TYPE_CHECKING:
 from .configuration_jukebox import (
     JUKEBOX_PRETRAINED_CONFIG_ARCHIVE_MAP,
     JukeboxConfig,
     JukeboxPriorConfig,
     JukeboxVQVAEConfig,
 )
 from .tokenization_jukebox import JukeboxTokenizer

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_jukebox import (
      JUKEBOX_PRETRAINED_MODEL_ARCHIVE_LIST,
      JukeboxModel,
      JukeboxPreTrainedModel,
      JukeboxPrior,
      JukeboxVQVAE,
  )

else:
 import sys

 SCREAMING_SNAKE_CASE_					:	Optional[int]							  =						_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)






