


import subprocess
import sys

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





class   __a			(    _lowerCAmelCase      ):
 @require_torch
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				str:




        '''simple docstring'''
        lowercase__:   str  			=   "\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        "

        lowercase__:   Union[str, Any]  			=   "\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        "

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

        # should succeed
        lowercase__:   Tuple  			=   self.get_env()
        # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
        lowercase__:   Tuple  			=   "1"
        lowercase__:   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 )					->				Dict:




        '''simple docstring'''
        lowercase__:   Dict  			=   "\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        "

        lowercase__:   Union[str, Any]  			=   "\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        "

        lowercase__:   Optional[int]  			=   "\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
        lowercase__:   str  			=   "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
        lowercase__:   Any  			=   [sys.executable, "-c", "\n".join([load, run, mock] )]

        # should succeed
        lowercase__:   str  			=   self.get_env()
        lowercase__:   List[str]  			=   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 )					->				Union[str, Any]:




        '''simple docstring'''
        lowercase__:   Tuple  			=   "\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        "

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

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

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

        # next emulate no network
        lowercase__:   Dict  			=   [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
        lowercase__:   int  			=   "1"
        lowercase__:   Dict  			=   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 )					->				int:




        '''simple docstring'''
        lowercase__:   List[str]  			=   "\nfrom transformers import pipeline\n        "
        lowercase__:   Tuple  			=   "\nmname = \"hf-internal-testing/tiny-random-bert\"\npipe = pipeline(model=mname)\n        "

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




        '''simple docstring'''
        lowercase__:   Optional[int]  			=   "\nfrom transformers import AutoModel\n        "
        lowercase__:   Tuple  			=   "\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
        lowercase__:   Dict  			=   [sys.executable, "-c", "\n".join([load, run] )]

        # should succeed
        lowercase__:   Optional[Any]  			=   self.get_env()
        lowercase__:   int  			=   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
        lowercase__:   str  			=   "1"
        lowercase__:   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() )









'''simple docstring'''


import pytest

from datasets import inspect_metric, list_metrics, load_metric
@pytest.fixture
def 							UpperCAmelCase_  (						__lowerCamelCase					:					List[str] ):
   monkeypatch.setattr("datasets.utils.deprecation_utils._emitted_deprecation_warnings"	,set() )
@pytest.fixture
def 							UpperCAmelCase_  (						__lowerCamelCase					:					Any ):
   class       a_   :





        def __init__( self		:					int	,      lowercase		:					int						):





           """simple docstring"""




           lowercase_			:Optional[Any]    			= metric_id







   class       a_   :
        __A      =			[MetricMock(_lowerCAmelCase							) for metric_id in ["accuracy", "mse", "precision", "codeparrot/apps_metric"]]





        def 							lowercase__      ( self		:					Union[str, Any]						):





           """simple docstring"""




           return self._metrics







   monkeypatch.setattr("datasets.inspect.huggingface_hub"	,HfhMock() )


@pytest.mark.parametrize(
    "func, args"	,[(load_metric, ("metrics/mse",)), (list_metrics, ()), (inspect_metric, ("metrics/mse", "tmp_path"))] )
def 							UpperCAmelCase_  (						__lowerCamelCase					:					Union[str, Any]	,__lowerCamelCase					:					int	,__lowerCamelCase					:					Optional[int]	,__lowerCamelCase					:					Union[str, Any]	,__lowerCamelCase					:					Tuple ):
   if "tmp_path" in args:
      lowercase_			:Union[str, Any]    			= tuple(arg if arg != "tmp_path" else tmp_path for arg in args )
   with pytest.warns(__lowerCamelCase	,match="https://huggingface.co/docs/evaluate" ):
      func(*__lowerCamelCase )


