

import subprocess
import sys

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






class 	UpperCAmelCase__		(	A__	):






  """simple docstring"""


  @require_torch
  def 				lowercase_		(							self						:		Optional[Any] )		->							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
       SCREAMING_SNAKE_CASE__   =   '''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

       SCREAMING_SNAKE_CASE__   =   '''
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")
        '''

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

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

       SCREAMING_SNAKE_CASE__   =   '''
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")
        '''

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

       # should succeed
       SCREAMING_SNAKE_CASE__   =   self.get_env()
       SCREAMING_SNAKE_CASE__   =   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 				lowercase_		(							self						:		Optional[Any] )		->							Dict:
       # 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
       SCREAMING_SNAKE_CASE__   =   '''
from transformers import BertConfig, BertModel, BertTokenizer
        '''

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

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

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

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

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





def     UpperCAmelCase_       (						_A    ):





     '''simple docstring'''




     if not isinstance(_A				,		_A    ):
          raise ValueError('''multiplicative_persistence() only accepts integral values'''    )
     if num < 0:
          raise ValueError('''multiplicative_persistence() does not accept negative values'''    )

     SCREAMING_SNAKE_CASE__   =   0
     SCREAMING_SNAKE_CASE__   =   str(_A    )

     while len(_A    ) != 1:
          SCREAMING_SNAKE_CASE__   =   [int(_A    ) for i in num_string]

          SCREAMING_SNAKE_CASE__   =   1
          for i in range(0				,		len(_A    )    ):
               total *= numbers[i]

          SCREAMING_SNAKE_CASE__   =   str(_A    )

          steps += 1
     return steps
def     UpperCAmelCase_       (						_A    ):





     '''simple docstring'''




     if not isinstance(_A				,		_A    ):
          raise ValueError('''additive_persistence() only accepts integral values'''    )
     if num < 0:
          raise ValueError('''additive_persistence() does not accept negative values'''    )

     SCREAMING_SNAKE_CASE__   =   0
     SCREAMING_SNAKE_CASE__   =   str(_A    )

     while len(_A    ) != 1:
          SCREAMING_SNAKE_CASE__   =   [int(_A    ) for i in num_string]

          SCREAMING_SNAKE_CASE__   =   0
          for i in range(0				,		len(_A    )    ):
               total += numbers[i]

          SCREAMING_SNAKE_CASE__   =   str(_A    )

          steps += 1
     return steps


if __name__ == "__main__":
       import doctest

       doctest.testmod()


