


'''simple docstring'''







import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch
class 						UpperCamelCase__       (			_UpperCAmelCase						):



     """simple docstring"""






     @require_torch
     def a						(			self ):

            '''simple docstring'''

            _lowerCAmelCase     :				Union[str, Any]       =							'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

            _lowerCAmelCase     :				int       =							'\nmname = \"hf-internal-testing/tiny-random-bert\"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task=\"fill-mask\", model=mname)\nprint(\"success\")\n        '

            _lowerCAmelCase     :				Any       =							'\nimport socket\ndef offline_socket(*args, **kwargs): raise RuntimeError(\"Offline mode is enabled, we shouldn\'t access internet\")\nsocket.socket = offline_socket\n        '

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

            # should succeed
            _lowerCAmelCase     :				int       =							self.get_env()
            # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
            _lowerCAmelCase     :				Optional[int]       =							'1'
            _lowerCAmelCase     :				Optional[int]       =							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 a						(			self ):

            '''simple docstring'''

            _lowerCAmelCase     :				List[str]       =							'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

            _lowerCAmelCase     :				Dict       =							'\nmname = \"hf-internal-testing/tiny-random-bert\"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task=\"fill-mask\", model=mname)\nprint(\"success\")\n        '

            _lowerCAmelCase     :				Tuple       =							'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error(\"Faking flaky internet\")\nsocket.socket = offline_socket\n        '

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

            # should succeed
            _lowerCAmelCase     :				Dict       =							self.get_env()
            _lowerCAmelCase     :				List[Any]       =							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 a						(			self ):

            '''simple docstring'''

            _lowerCAmelCase     :				Union[str, Any]       =							'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

            _lowerCAmelCase     :				List[Any]       =							'\nmname = \"hf-internal-testing/tiny-random-bert-sharded\"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint(\"success\")\n        '

            _lowerCAmelCase     :				Any       =							'\nimport socket\ndef offline_socket(*args, **kwargs): raise ValueError(\"Offline mode is enabled\")\nsocket.socket = offline_socket\n        '

            # baseline - just load from_pretrained with normal network
            _lowerCAmelCase     :				Union[str, Any]       =							[sys.executable, '-c', '\n'.join([load, run] )]

            # should succeed
            _lowerCAmelCase     :				Tuple       =							self.get_env()
            _lowerCAmelCase     :				Optional[int]       =							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
            _lowerCAmelCase     :				Any       =							[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     :				Optional[Any]       =							'1'
            _lowerCAmelCase     :				List[Any]       =							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 a						(			self ):

            '''simple docstring'''

            _lowerCAmelCase     :				Tuple       =							'\nfrom transformers import pipeline\n        '
            _lowerCAmelCase     :				str       =							'\nmname = \"hf-internal-testing/tiny-random-bert\"\npipe = pipeline(model=mname)\n        '

            _lowerCAmelCase     :				Any       =							'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")\nsocket.socket = offline_socket\n        '
            _lowerCAmelCase     :				Tuple       =							self.get_env()
            _lowerCAmelCase     :				int       =							'1'
            _lowerCAmelCase     :				Tuple       =							[sys.executable, '-c', '\n'.join([load, mock, run] )]
            _lowerCAmelCase     :				Optional[int]       =							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 a						(			self ):

            '''simple docstring'''

            _lowerCAmelCase     :				Union[str, Any]       =							'\nfrom transformers import AutoModel\n        '
            _lowerCAmelCase     :				List[Any]       =							'\nmname = \"hf-internal-testing/test_dynamic_model\"\nAutoModel.from_pretrained(mname, trust_remote_code=True)\nprint(\"success\")\n        '

            # baseline - just load from_pretrained with normal network
            _lowerCAmelCase     :				str       =							[sys.executable, '-c', '\n'.join([load, run] )]

            # should succeed
            _lowerCAmelCase     :				str       =							self.get_env()
            _lowerCAmelCase     :				Dict       =							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
            _lowerCAmelCase     :				Tuple       =							'1'
            _lowerCAmelCase     :				List[Any]       =							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"""



import importlib
import os
from dataclasses import dataclass
from enum import Enum
from typing import Any, Dict, Optional, Union

import torch

from ..utils import BaseOutput


a       :						int       		=		'''scheduler_config.json'''




class 		a_    (			_UpperCAmelCase						):
    a   :					List[Any]			     =					1
    a   :					Tuple			     =					2
    a   :					Dict			     =					3
    a   :					str			     =					4
    a   :					Optional[int]			     =					5
    a   :					Any			     =					6
    a   :					int			     =					7
    a   :					Any			     =					8
    a   :					List[Any]			     =					9
    a   :					Any			     =					10
    a   :					List[str]			     =					11
    a   :					Optional[int]			     =					12
    a   :					Any			     =					13
    a   :					Dict			     =					14







@dataclass
class 		a_    (			_UpperCAmelCase						):
    a   :					torch.FloatTensor







class 		a_    :
    a   :					Dict			     =					SCHEDULER_CONFIG_NAME
    a   :					List[str]			     =					[]
    a   :					Optional[int]			     =					True







    @classmethod
    def   _snake_case       (      cls      :							List[str]  ,       __UpperCamelCase      :							Dict[str, Any] = None  ,       __UpperCamelCase      :							Optional[str] = None  ,       __UpperCamelCase      :							List[str]=False  ,       **__UpperCamelCase      :							int  ,       )		->Dict:






        '''simple docstring'''






        _UpperCAmelCase     ,_UpperCAmelCase     ,_UpperCAmelCase        = cls.load_config(
            pretrained_model_name_or_path=__UpperCamelCase  ,       subfolder=__UpperCamelCase  ,       return_unused_kwargs=__UpperCamelCase  ,       return_commit_hash=__UpperCamelCase  ,       **__UpperCamelCase  ,       )
        return cls.from_config(__UpperCamelCase  ,       return_unused_kwargs=__UpperCamelCase  ,       **__UpperCamelCase  )







    def   _snake_case       (      self      :							Tuple  ,       __UpperCamelCase      :							Union[str, os.PathLike]  ,       __UpperCamelCase      :							bool = False  ,       **__UpperCamelCase      :							List[str]  )		->Any:






        '''simple docstring'''






        self.save_config(save_directory=__UpperCamelCase  ,       push_to_hub=__UpperCamelCase  ,       **__UpperCamelCase  )







    @property
    def   _snake_case       (      self      :							Optional[int]  )		->List[Any]:






        '''simple docstring'''






        return self._get_compatibles()







    @classmethod
    def   _snake_case       (      cls      :							List[str]  )		->Dict:






        '''simple docstring'''






        _UpperCAmelCase        = list(set([cls.__name__] + cls._compatibles  )  )
        _UpperCAmelCase        = importlib.import_module(__name__.split("""."""  )[0]  )
        _UpperCAmelCase        = [
            getattr(__UpperCamelCase  ,       __UpperCamelCase  ) for c in compatible_classes_str if hasattr(__UpperCamelCase  ,       __UpperCamelCase  )
        ]
        return compatible_classes