


'''simple docstring'''





import subprocess
import sys

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




class 				UpperCAmelCase			(			_snake_case						):


 @require_torch
 def        __SCREAMING_SNAKE_CASE					(      self							:	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
    UpperCAmelCase__	:List[str]      =      '''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

    UpperCAmelCase__	:List[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 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__	: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__	:Tuple      =      [sys.executable, '''-c''', '''\n'''.join([load, run, mock] )]

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

    UpperCAmelCase__	:Union[str, 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 socket.error("Faking flaky internet")
socket.socket = offline_socket
        '''

    # Force fetching the files so that we can use the cache
    UpperCAmelCase__	:Union[str, 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__	:List[str]      =      [sys.executable, '''-c''', '''\n'''.join([load, run, mock] )]

    # should succeed
    UpperCAmelCase__	:Union[str, Any]      =      self.get_env()
    UpperCAmelCase__	:Tuple      =      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        __SCREAMING_SNAKE_CASE					(      self							:	List[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
    UpperCAmelCase__	:Union[str, Any]      =      '''
from transformers import BertConfig, BertModel, BertTokenizer
        '''

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

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

    # should succeed
    UpperCAmelCase__	:str      =      self.get_env()
    UpperCAmelCase__	:List[Any]      =      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__	:Tuple      =      [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__	:int      =      '''1'''
    UpperCAmelCase__	:List[Any]      =      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        __SCREAMING_SNAKE_CASE					(      self							:	Any ):
    UpperCAmelCase__	:Dict      =      '''
from transformers import pipeline
        '''
    UpperCAmelCase__	:List[str]      =      '''
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__	:List[Any]      =      self.get_env()
    UpperCAmelCase__	:str      =      '''1'''
    UpperCAmelCase__	:Optional[int]      =      [sys.executable, '''-c''', '''\n'''.join([load, mock, run] )]
    UpperCAmelCase__	:int      =      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        __SCREAMING_SNAKE_CASE					(      self							:	int ):
    UpperCAmelCase__	:str      =      '''
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__	:Union[str, Any]      =      [sys.executable, '''-c''', '''\n'''.join([load, run] )]

    # should succeed
    UpperCAmelCase__	:Optional[int]      =      self.get_env()
    UpperCAmelCase__	:Dict      =      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__	:Dict      =      '''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'''





from ..utils import DummyObject, requires_backends




class 				UpperCAmelCase			(			metaclass=_snake_case						):
 UpperCAmelCase				=					["note_seq"]


 def __init__(      self							:	List[str]			,       *__lowerCamelCase							:	int			,       **__lowerCamelCase							:	Optional[int] ):
    requires_backends(self			,       ['''note_seq'''] )


 @classmethod
 def        __SCREAMING_SNAKE_CASE					(      cls							:	Optional[Any]			,       *__lowerCamelCase							:	Dict			,       **__lowerCamelCase							:	Optional[Any] ):
    requires_backends(cls			,       ['''note_seq'''] )





 @classmethod
 def        __SCREAMING_SNAKE_CASE					(      cls							:	str			,       *__lowerCamelCase							:	Any			,       **__lowerCamelCase							:	Optional[Any] ):
    requires_backends(cls			,       ['''note_seq'''] )



