






'''simple docstring'''



import os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


SCREAMING_SNAKE_CASE						=	'\\n    Text data.\n    Second line of data.'

SCREAMING_SNAKE_CASE						=	'file'


@pytest.fixture(scope='''session'''			)
def 							lowercase_				(    __A :       Optional[int]			)       ->      Dict:



 """simple docstring"""




 lowercase					:   Union[str, Any]				  =tmp_path_factory.mktemp('''data'''			) / (FILE_PATH + '''.zstd''')
 lowercase					:   Dict				  =bytes(__A					,      '''utf-8'''			)
 with zstd.open(__A					,      '''wb'''			) as f:
  f.write(__A			)
 return path


@pytest.fixture
def 							lowercase_				(    __A :       Any			)       ->      int:



 """simple docstring"""




 with open(os.path.join(tmpfs.local_root_dir					,      __A			)					,      '''w'''			) as f:
  f.write(__A			)
 return FILE_PATH


@pytest.mark.parametrize('''compression_format'''					,      ['''gzip''', '''xz''', '''zstd''']			)
def 							lowercase_				(    __A :       int					,      __A :       int					,      __A :       Any					,      __A :       List[Any]					,      __A :       Dict					,      __A :       Tuple			)       ->      Dict:



 """simple docstring"""




 lowercase					:   Optional[Any]				  ={'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_path}
 lowercase					:   List[Any]				  =input_paths[compression_format]
 lowercase					:   Optional[Any]				  =tmp_path / '''cache'''
 lowercase					:   List[Any]				  =DownloadConfig(cache_dir=__A					,      extract_compressed_file=__A			)
 lowercase					:   List[str]				  =cached_path(__A					,      download_config=__A			)
 with open(__A			) as f:
  lowercase					:   Any				  =f.read()
 with open(__A			) as f:
  lowercase					:   int				  =f.read()
 assert extracted_file_content == expected_file_content


@pytest.mark.parametrize('''default_extracted'''					,      [True, False]			)
@pytest.mark.parametrize('''default_cache_dir'''					,      [True, False]			)
def 							lowercase_				(    __A :       Union[str, Any]					,      __A :       List[Any]					,      __A :       List[Any]					,      __A :       Optional[Any]					,      __A :       Any			)       ->      Optional[Any]:



 """simple docstring"""




 lowercase					:   Dict				  ='''custom_cache'''
 lowercase					:   int				  ='''custom_extracted_dir'''
 lowercase					:   Optional[Any]				  =tmp_path / '''custom_extracted_path'''
 if default_extracted:
  lowercase					:   Tuple				  =('''downloads''' if default_cache_dir else custom_cache_dir, '''extracted''')
 else:
  monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_DIR'''					,      __A			)
  monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_PATH'''					,      str(__A			)			)
  lowercase					:   List[Any]				  =custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

 lowercase					:   str				  =xz_file
 lowercase					:   str				  =(
     DownloadConfig(extract_compressed_file=__A			)
     if default_cache_dir
     else DownloadConfig(cache_dir=tmp_path / custom_cache_dir					,      extract_compressed_file=__A			)
 )
 lowercase					:   Tuple				  =cached_path(__A					,      download_config=__A			)
 assert Path(__A			).parent.parts[-2:] == expected


def 							lowercase_				(    __A :       int			)       ->      Any:



 """simple docstring"""




 lowercase					:   Tuple				  =str(Path(__A			).resolve()			)
 assert cached_path(__A			) == text_file
 # relative path
 lowercase					:   Optional[Any]				  =str(Path(__A			).resolve().relative_to(Path(os.getcwd()			)			)			)
 assert cached_path(__A			) == text_file


def 							lowercase_				(    __A :       Union[str, Any]			)       ->      Optional[Any]:



 """simple docstring"""




 lowercase					:   Optional[int]				  =str(tmp_path.resolve() / '''__missing_file__.txt'''			)
 with pytest.raises(__A			):
  cached_path(__A			)
 # relative path
 lowercase					:   List[str]				  ='''./__missing_file__.txt'''
 with pytest.raises(__A			):
  cached_path(__A			)


def 							lowercase_				(    __A :       Optional[Any]			)       ->      Tuple:



 """simple docstring"""




 lowercase					:   Union[str, Any]				  =get_from_cache(F'tmp://{tmpfs_file}'			)
 with open(__A			) as f:
  lowercase					:   Optional[int]				  =f.read()
 assert output_file_content == FILE_CONTENT


@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,      __A			)
def 							lowercase_				(    )       ->      Dict:



 """simple docstring"""




 with pytest.raises(__A			):
  cached_path('''https://huggingface.co'''			)


@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,      __A			)
def 							lowercase_				(    __A :       Tuple			)       ->      Optional[Any]:



 """simple docstring"""




 lowercase					:   Optional[int]				  =tmp_path_factory.mktemp('''data'''			) / '''file.html'''
 with pytest.raises(__A			):
  http_get('''https://huggingface.co'''					,      temp_file=__A			)
 with pytest.raises(__A			):
  http_head('''https://huggingface.co'''			)


@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,      __A			)
def 							lowercase_				(    __A :       Optional[int]			)       ->      Optional[Any]:



 """simple docstring"""




 lowercase					:   str				  =tmp_path_factory.mktemp('''data'''			) / '''file.html'''
 with pytest.raises(__A			):
  ftp_get('''ftp://huggingface.co'''					,      temp_file=__A			)
 with pytest.raises(__A			):
  ftp_head('''ftp://huggingface.co'''			)





@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,      __A			)
def 							lowercase_				(    __A :       List[str]			)       ->      Any:



 """simple docstring"""




 lowercase					:   Tuple				  =tmp_path_factory.mktemp('''data'''			) / '''file.html'''
 with pytest.raises(__A			):
  fsspec_get('''s3://huggingface.co'''					,      temp_file=__A			)
 with pytest.raises(__A			):
  fsspec_head('''s3://huggingface.co'''			)











'''simple docstring'''



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 				A__      ( self			:		Tuple		)  ->  Optional[int]:


       '''simple docstring'''

       lowercase					:   Any				  ='''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

       lowercase					:   Optional[int]				  ='''
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")
        '''

       lowercase					:   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
       lowercase					:   Tuple				  ='''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
       lowercase					:   List[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					:   Optional[Any]				  ='''1'''
       lowercase					:   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			:		str		)  ->  List[str]:


       '''simple docstring'''

       lowercase					:   str				  ='''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

       lowercase					:   Optional[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")
        '''

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

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


       '''simple docstring'''

       lowercase					:   Optional[Any]				  ='''
from transformers import BertConfig, BertModel, BertTokenizer
        '''

       lowercase					:   List[Any]				  ='''
mname = "hf-internal-testing/tiny-random-bert-sharded"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print("success")
        '''

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

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

       # next emulate no network
       lowercase					:   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
       lowercase					:   Any				  ='''1'''
       lowercase					:   Optional[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			:		Tuple		)  ->  Optional[int]:


       '''simple docstring'''

       lowercase					:   Optional[int]				  ='''
from transformers import pipeline
        '''
       lowercase					:   List[Any]				  ='''
mname = "hf-internal-testing/tiny-random-bert"
pipe = pipeline(model=mname)
        '''

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


       '''simple docstring'''

       lowercase					:   List[str]				  ='''
from transformers import AutoModel
        '''
       lowercase					:   Dict				  ='''
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
       lowercase					:   Dict				  =[sys.executable, '''-c''', '''\n'''.join([load, run]		)]

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



