



"""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,
)


snake_case   					=				'''\
    Text data.
    Second line of data.'''

snake_case   					=				'''file'''




@pytest.fixture(scope='''session'''	)
def 			snake_case		(		lowerCAmelCase_	)						->				Any:
  _snake_case   	=      tmp_path_factory.mktemp('''data'''	) / (FILE_PATH + '''.zstd''')
  _snake_case   	=      bytes(lowerCAmelCase_  ,						'''utf-8'''	)
  with zstd.open(lowerCAmelCase_  ,						'''wb'''	) as f:
    f.write(lowerCAmelCase_	)
  return path




@pytest.fixture
def 			snake_case		(		lowerCAmelCase_	)						->				Tuple:
  with open(os.path.join(tmpfs.local_root_dir  ,						lowerCAmelCase_	)  ,						'''w'''	) as f:
    f.write(lowerCAmelCase_	)
  return FILE_PATH




@pytest.mark.parametrize('''compression_format'''  ,						['''gzip''', '''xz''', '''zstd''']	)
def 			snake_case		(		lowerCAmelCase_  ,						lowerCAmelCase_  ,						lowerCAmelCase_  ,						lowerCAmelCase_  ,						lowerCAmelCase_  ,						lowerCAmelCase_	)						->				Tuple:
  _snake_case   	=      {'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_path}
  _snake_case   	=      input_paths[compression_format]
  _snake_case   	=      tmp_path / '''cache'''
  _snake_case   	=      DownloadConfig(cache_dir=lowerCAmelCase_  ,						extract_compressed_file=lowerCAmelCase_	)
  _snake_case   	=      cached_path(lowerCAmelCase_  ,						download_config=lowerCAmelCase_	)
  with open(lowerCAmelCase_	) as f:
    _snake_case   	=      f.read()
  with open(lowerCAmelCase_	) as f:
    _snake_case   	=      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 			snake_case		(		lowerCAmelCase_  ,						lowerCAmelCase_  ,						lowerCAmelCase_  ,						lowerCAmelCase_  ,						lowerCAmelCase_	)						->				Union[str, Any]:
  _snake_case   	=      '''custom_cache'''
  _snake_case   	=      '''custom_extracted_dir'''
  _snake_case   	=      tmp_path / '''custom_extracted_path'''
  if default_extracted:
    _snake_case   	=      ('''downloads''' if default_cache_dir else custom_cache_dir, '''extracted''')
  else:
    monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_DIR'''  ,						lowerCAmelCase_	)
    monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_PATH'''  ,						str(lowerCAmelCase_	)	)
    _snake_case   	=      custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

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




def 			snake_case		(		lowerCAmelCase_	)						->				List[str]:
  # absolute path
  _snake_case   	=      str(Path(lowerCAmelCase_	).resolve()	)
  assert cached_path(lowerCAmelCase_	) == text_file
  # relative path
  _snake_case   	=      str(Path(lowerCAmelCase_	).resolve().relative_to(Path(os.getcwd()	)	)	)
  assert cached_path(lowerCAmelCase_	) == text_file




def 			snake_case		(		lowerCAmelCase_	)						->				Dict:
  # absolute path
  _snake_case   	=      str(tmp_path.resolve() / '''__missing_file__.txt'''	)
  with pytest.raises(lowerCAmelCase_	):
    cached_path(lowerCAmelCase_	)
  # relative path
  _snake_case   	=      '''./__missing_file__.txt'''
  with pytest.raises(lowerCAmelCase_	):
    cached_path(lowerCAmelCase_	)




def 			snake_case		(		lowerCAmelCase_	)						->				Any:
  _snake_case   	=      get_from_cache(f"""tmp://{tmpfs_file}"""	)
  with open(lowerCAmelCase_	) as f:
    _snake_case   	=      f.read()
  assert output_file_content == FILE_CONTENT




@patch('''datasets.config.HF_DATASETS_OFFLINE'''  ,						lowerCAmelCase_	)
def 			snake_case		(		)						->				Optional[Any]:
  with pytest.raises(lowerCAmelCase_	):
    cached_path('''https://huggingface.co'''	)




@patch('''datasets.config.HF_DATASETS_OFFLINE'''  ,						lowerCAmelCase_	)
def 			snake_case		(		lowerCAmelCase_	)						->				Optional[int]:
  _snake_case   	=      tmp_path_factory.mktemp('''data'''	) / '''file.html'''
  with pytest.raises(lowerCAmelCase_	):
    http_get('''https://huggingface.co'''  ,						temp_file=lowerCAmelCase_	)
  with pytest.raises(lowerCAmelCase_	):
    http_head('''https://huggingface.co'''	)




@patch('''datasets.config.HF_DATASETS_OFFLINE'''  ,						lowerCAmelCase_	)
def 			snake_case		(		lowerCAmelCase_	)						->				int:
  _snake_case   	=      tmp_path_factory.mktemp('''data'''	) / '''file.html'''
  with pytest.raises(lowerCAmelCase_	):
    ftp_get('''ftp://huggingface.co'''  ,						temp_file=lowerCAmelCase_	)
  with pytest.raises(lowerCAmelCase_	):
    ftp_head('''ftp://huggingface.co'''	)







@patch('''datasets.config.HF_DATASETS_OFFLINE'''  ,						lowerCAmelCase_	)
def 			snake_case		(		lowerCAmelCase_	)						->				str:
  _snake_case   	=      tmp_path_factory.mktemp('''data'''	) / '''file.html'''
  with pytest.raises(lowerCAmelCase_	):
    fsspec_get('''s3://huggingface.co'''  ,						temp_file=lowerCAmelCase_	)
  with pytest.raises(lowerCAmelCase_	):
    fsspec_head('''s3://huggingface.co'''	)










"""simple docstring"""






import importlib
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
import transformers.models.auto
from transformers.models.auto.configuration_auto import CONFIG_MAPPING, AutoConfig
from transformers.models.bert.configuration_bert import BertConfig
from transformers.models.roberta.configuration_roberta import RobertaConfig
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, get_tests_dir


sys.path.append(str(Path(__file__).parent.parent.parent.parent / '''utils'''))

from test_module.custom_configuration import CustomConfig  # noqa E402


snake_case   					=				get_tests_dir('''fixtures/dummy-config.json''')



class     UpperCAmelCase    (	unittest.TestCase  ):



   def    __UpperCAmelCase    (       self    :	Any				):



     """simple docstring"""
     _snake_case   	=      0



   def    __UpperCAmelCase    (       self    :	Optional[Any]				):



     """simple docstring"""
     self.assertIsNotNone(transformers.models.auto.__spec__				)
     self.assertIsNotNone(importlib.util.find_spec('''transformers.models.auto'''				)				)



   def    __UpperCAmelCase    (       self    :	List[str]				):



     """simple docstring"""
     _snake_case   	=      AutoConfig.from_pretrained('''bert-base-uncased'''				)
     self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)



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



     """simple docstring"""
     _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase				)
     self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	int				):



     """simple docstring"""
     _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase				)
     self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)



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



     """simple docstring"""
     _snake_case   	=      AutoConfig.for_model('''roberta'''				)
     self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	Dict				):



     """simple docstring"""
     with tempfile.TemporaryDirectory() as tmp_dir:
       # This model name contains bert and roberta, but roberta ends up being picked.
       _snake_case   	=      os.path.join(__lowerCamelCase							,       '''fake-roberta'''				)
       os.makedirs(__lowerCamelCase							,       exist_ok=__lowerCamelCase				)
       with open(os.path.join(__lowerCamelCase							,       '''config.json'''				)							,       '''w'''				) as f:
         f.write(json.dumps({}				)				)
       _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase				)
       self.assertEqual(type(__lowerCamelCase				)							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	Optional[Any]				):



     """simple docstring"""
     try:
       AutoConfig.register('''custom'''							,       __lowerCamelCase				)
       # Wrong model type will raise an error
       with self.assertRaises(__lowerCamelCase				):
         AutoConfig.register('''model'''							,       __lowerCamelCase				)
       # Trying to register something existing in the Transformers library will raise an error
       with self.assertRaises(__lowerCamelCase				):
         AutoConfig.register('''bert'''							,       __lowerCamelCase				)

       # Now that the config is registered, it can be used as any other config with the auto-API
       _snake_case   	=      CustomConfig()
       with tempfile.TemporaryDirectory() as tmp_dir:
         config.save_pretrained(__lowerCamelCase				)
         _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase				)
         self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)

     finally:
       if "custom" in CONFIG_MAPPING._extra_content:
         del CONFIG_MAPPING._extra_content["custom"]



   def    __UpperCAmelCase    (       self    :	Optional[int]				):



     """simple docstring"""
     with self.assertRaisesRegex(
         __lowerCamelCase							,       '''bert-base is not a local folder and is not a valid model identifier'''				):
       _snake_case   	=      AutoConfig.from_pretrained('''bert-base'''				)



   def    __UpperCAmelCase    (       self    :	Dict				):



     """simple docstring"""
     with self.assertRaisesRegex(
         __lowerCamelCase							,       R'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'''				):
       _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase							,       revision='''aaaaaa'''				)



   def    __UpperCAmelCase    (       self    :	Dict				):



     """simple docstring"""
     with self.assertRaisesRegex(
         __lowerCamelCase							,       '''hf-internal-testing/no-config-test-repo does not appear to have a file named config.json.'''							,       ):
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/no-config-test-repo'''				)



   def    __UpperCAmelCase    (       self    :	str				):



     """simple docstring"""
     # If remote code is not set, we will time out when asking whether to load the model.
     with self.assertRaises(__lowerCamelCase				):
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''				)
     # If remote code is disabled, we can't load this config.
     with self.assertRaises(__lowerCamelCase				):
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''							,       trust_remote_code=__lowerCamelCase				)

     _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''							,       trust_remote_code=__lowerCamelCase				)
     self.assertEqual(config.__class__.__name__							,       '''NewModelConfig'''				)

     # Test config can be reloaded.
     with tempfile.TemporaryDirectory() as tmp_dir:
       config.save_pretrained(__lowerCamelCase				)
       _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase							,       trust_remote_code=__lowerCamelCase				)
     self.assertEqual(reloaded_config.__class__.__name__							,       '''NewModelConfig'''				)





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



     """simple docstring"""



     class     UpperCAmelCase    (	__SCREAMING_SNAKE_CASE  ):
        A__		:	List[Any]           =  '''new-model'''

     try:
       AutoConfig.register('''new-model'''							,       __lowerCamelCase				)
       # If remote code is not set, the default is to use local
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''				)
       self.assertEqual(config.__class__.__name__							,       '''NewModelConfigLocal'''				)

       # If remote code is disabled, we load the local one.
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''							,       trust_remote_code=__lowerCamelCase				)
       self.assertEqual(config.__class__.__name__							,       '''NewModelConfigLocal'''				)

       # If remote is enabled, we load from the Hub
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''							,       trust_remote_code=__lowerCamelCase				)
       self.assertEqual(config.__class__.__name__							,       '''NewModelConfig'''				)

     finally:
       if "new-model" in CONFIG_MAPPING._extra_content:
         del CONFIG_MAPPING._extra_content["new-model"]





