


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

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
)
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
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


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


class     __UpperCAmelCase							(       unittest.TestCase							):


     """simple docstring"""


     def 	snake_case_  (	self    ):
        __a					=  0


     def 	snake_case_  (	self    ):
        __a					=  AutoFeatureExtractor.from_pretrained("""facebook/wav2vec2-base-960h"""    )
        self.assertIsInstance(__A  ,       __A    )


     def 	snake_case_  (	self    ):
        __a					=  AutoFeatureExtractor.from_pretrained(__A    )
        self.assertIsInstance(__A  ,       __A    )


     def 	snake_case_  (	self    ):
        with tempfile.TemporaryDirectory() as tmpdirname:
           __a					=  WavaVecaConfig()

           # remove feature_extractor_type to make sure config.json alone is enough to load feature processor locally
           __a					=  AutoFeatureExtractor.from_pretrained(__A    ).to_dict()

           config_dict.pop("""feature_extractor_type"""    )
           __a					=  WavaVecaFeatureExtractor(**__A    )

           # save in new folder
           model_config.save_pretrained(__A    )
           config.save_pretrained(__A    )

           __a					=  AutoFeatureExtractor.from_pretrained(__A    )

           # make sure private variable is not incorrectly saved
           __a					=  json.loads(config.to_json_string()    )
           self.assertTrue("""_processor_class""" not in dict_as_saved    )

        self.assertIsInstance(__A  ,       __A    )


     def 	snake_case_  (	self    ):
        __a					=  AutoFeatureExtractor.from_pretrained(__A    )
        self.assertIsInstance(__A  ,       __A    )


     def 	snake_case_  (	self    ):
        with self.assertRaisesRegex(
            __A  ,       """bert-base is not a local folder and is not a valid model identifier"""    ):
           __a					=  AutoFeatureExtractor.from_pretrained("""bert-base"""    )


     def 	snake_case_  (	self    ):
        with self.assertRaisesRegex(
            __A  ,       R"""aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"""    ):
           __a					=  AutoFeatureExtractor.from_pretrained(__A  ,       revision="""aaaaaa"""    )


     def 	snake_case_  (	self    ):
        with self.assertRaisesRegex(
            __A  ,       """hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json."""  ,       ):
           __a					=  AutoFeatureExtractor.from_pretrained("""hf-internal-testing/config-no-model"""    )


     def 	snake_case_  (	self    ):
        # If remote code is not set, we will time out when asking whether to load the model.
        with self.assertRaises(__A    ):
           __a					=  AutoFeatureExtractor.from_pretrained(
               """hf-internal-testing/test_dynamic_feature_extractor"""    )
        # If remote code is disabled, we can't load this config.
        with self.assertRaises(__A    ):
           __a					=  AutoFeatureExtractor.from_pretrained(
               """hf-internal-testing/test_dynamic_feature_extractor"""  ,       trust_remote_code=__A    )

        __a					=  AutoFeatureExtractor.from_pretrained(
            """hf-internal-testing/test_dynamic_feature_extractor"""  ,       trust_remote_code=__A    )
        self.assertEqual(feature_extractor.__class__.__name__  ,       """NewFeatureExtractor"""    )

        # Test feature extractor can be reloaded.
        with tempfile.TemporaryDirectory() as tmp_dir:
           feature_extractor.save_pretrained(__A    )
           __a					=  AutoFeatureExtractor.from_pretrained(__A  ,       trust_remote_code=__A    )
        self.assertEqual(reloaded_feature_extractor.__class__.__name__  ,       """NewFeatureExtractor"""    )


     def 	snake_case_  (	self    ):
        try:
           AutoConfig.register("""custom"""  ,       __A    )
           AutoFeatureExtractor.register(__A  ,       __A    )
           # Trying to register something existing in the Transformers library will raise an error
           with self.assertRaises(__A    ):
              AutoFeatureExtractor.register(__A  ,       __A    )

           # Now that the config is registered, it can be used as any other config with the auto-API
           __a					=  CustomFeatureExtractor.from_pretrained(__A    )
           with tempfile.TemporaryDirectory() as tmp_dir:
              feature_extractor.save_pretrained(__A    )
              __a					=  AutoFeatureExtractor.from_pretrained(__A    )
              self.assertIsInstance(__A  ,       __A    )

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


     def 	snake_case_  (	self    ):


        class     __UpperCAmelCase							(       __A							):


             """simple docstring"""
             _lowerCamelCase   =      True

        try:
           AutoConfig.register("""custom"""  ,       __A    )
           AutoFeatureExtractor.register(__A  ,       __A    )
           # If remote code is not set, the default is to use local
           __a					=  AutoFeatureExtractor.from_pretrained(
               """hf-internal-testing/test_dynamic_feature_extractor"""    )
           self.assertEqual(feature_extractor.__class__.__name__  ,       """NewFeatureExtractor"""    )
           self.assertTrue(feature_extractor.is_local    )

           # If remote code is disabled, we load the local one.
           __a					=  AutoFeatureExtractor.from_pretrained(
               """hf-internal-testing/test_dynamic_feature_extractor"""  ,       trust_remote_code=__A    )
           self.assertEqual(feature_extractor.__class__.__name__  ,       """NewFeatureExtractor"""    )
           self.assertTrue(feature_extractor.is_local    )

           # If remote is enabled, we load from the Hub
           __a					=  AutoFeatureExtractor.from_pretrained(
               """hf-internal-testing/test_dynamic_feature_extractor"""  ,       trust_remote_code=__A    )
           self.assertEqual(feature_extractor.__class__.__name__  ,       """NewFeatureExtractor"""    )
           self.assertTrue(not hasattr(__A  ,       """is_local"""    )    )

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










'''simple docstring'''




import json
import os
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from requests.exceptions import HTTPError

from transformers.utils import (
    CONFIG_NAME,
    FLAX_WEIGHTS_NAME,
    TF2_WEIGHTS_NAME,
    TRANSFORMERS_CACHE,
    WEIGHTS_NAME,
    cached_file,
    get_file_from_repo,
    has_file,
)


lowercase__       : Any         =       '''hf-internal-testing/tiny-random-bert'''
lowercase__       : Optional[Any]         =       os.path.join(TRANSFORMERS_CACHE, '''models--hf-internal-testing--tiny-random-bert''')
lowercase__       : List[Any]         =       '''9b8c223d42b2188cb49d29af482996f9d0f3e5a6'''


class 					SCREAMING_SNAKE_CASE       (unittest.TestCase				):


     def 				SCREAMING_SNAKE_CASE    (		self):

          '''simple docstring'''






          __A		:					List[Any]						=			cached_file(_UpperCAmelCase		,      _UpperCAmelCase)
          # Should have downloaded the file in here
          self.assertTrue(os.path.isdir(_UpperCAmelCase))
          # Cache should contain at least those three subfolders:
          for subfolder in ["blobs", "refs", "snapshots"]:
               self.assertTrue(os.path.isdir(os.path.join(_UpperCAmelCase		,      _UpperCAmelCase)))
          with open(os.path.join(_UpperCAmelCase		,      'refs'		,      'main')) as f:
               __A		:					Any						=			f.read()
          self.assertEqual(_UpperCAmelCase		,      os.path.join(_UpperCAmelCase		,      'snapshots'		,      _UpperCAmelCase		,      _UpperCAmelCase))
          self.assertTrue(os.path.isfile(_UpperCAmelCase))

          # File is cached at the same place the second time.
          __A		:					Tuple						=			cached_file(_UpperCAmelCase		,      _UpperCAmelCase)
          self.assertEqual(_UpperCAmelCase		,      _UpperCAmelCase)

          # Using a specific revision to test the full commit hash.
          __A		:					List[Any]						=			cached_file(_UpperCAmelCase		,      _UpperCAmelCase		,      revision='9b8c223')
          self.assertEqual(_UpperCAmelCase		,      os.path.join(_UpperCAmelCase		,      'snapshots'		,      _UpperCAmelCase		,      _UpperCAmelCase))


     def 				SCREAMING_SNAKE_CASE    (		self):

          '''simple docstring'''






          with self.assertRaisesRegex(_UpperCAmelCase		,      'is not a valid model identifier'):
               __A		:					Dict						=			cached_file('tiny-random-bert'		,      _UpperCAmelCase)

          with self.assertRaisesRegex(_UpperCAmelCase		,      'is not a valid git identifier'):
               __A		:					Optional[int]						=			cached_file(_UpperCAmelCase		,      _UpperCAmelCase		,      revision='aaaa')

          with self.assertRaisesRegex(_UpperCAmelCase		,      'does not appear to have a file named'):
               __A		:					int						=			cached_file(_UpperCAmelCase		,      'conf')


     def 				SCREAMING_SNAKE_CASE    (		self):

          '''simple docstring'''






          with self.assertRaisesRegex(_UpperCAmelCase		,      'does not appear to have a file named'):
               __A		:					Any						=			cached_file(_UpperCAmelCase		,      'conf')

          with open(os.path.join(_UpperCAmelCase		,      'refs'		,      'main')) as f:
               __A		:					Dict						=			f.read()
          self.assertTrue(os.path.isfile(os.path.join(_UpperCAmelCase		,      '.no_exist'		,      _UpperCAmelCase		,      'conf')))

          __A		:					List[Any]						=			cached_file(_UpperCAmelCase		,      'conf'		,      _raise_exceptions_for_missing_entries=_UpperCAmelCase)
          self.assertIsNone(_UpperCAmelCase)

          __A		:					str						=			cached_file(_UpperCAmelCase		,      'conf'		,      local_files_only=_UpperCAmelCase		,      _raise_exceptions_for_missing_entries=_UpperCAmelCase)
          self.assertIsNone(_UpperCAmelCase)

          __A		:					List[str]						=			mock.Mock()
          __A		:					Dict						=			500
          __A		:					List[str]						=			{}
          __A		:					List[Any]						=			HTTPError
          __A		:					Optional[Any]						=			{}

          # Under the mock environment we get a 500 error when trying to reach the tokenizer.
          with mock.patch('requests.Session.request'		,      return_value=_UpperCAmelCase) as mock_head:
               __A		:					Dict						=			cached_file(_UpperCAmelCase		,      'conf'		,      _raise_exceptions_for_connection_errors=_UpperCAmelCase)
               self.assertIsNone(_UpperCAmelCase)
               # This check we did call the fake head request
               mock_head.assert_called()


     def 				SCREAMING_SNAKE_CASE    (		self):

          '''simple docstring'''






          self.assertTrue(has_file('hf-internal-testing/tiny-bert-pt-only'		,      _UpperCAmelCase))
          self.assertFalse(has_file('hf-internal-testing/tiny-bert-pt-only'		,      _UpperCAmelCase))
          self.assertFalse(has_file('hf-internal-testing/tiny-bert-pt-only'		,      _UpperCAmelCase))


     def 				SCREAMING_SNAKE_CASE    (		self):

          '''simple docstring'''






          self.assertIsNone(get_file_from_repo('bert-base-cased'		,      'ahah.txt'))

          # The function raises if the repository does not exist.
          with self.assertRaisesRegex(_UpperCAmelCase		,      'is not a valid model identifier'):
               get_file_from_repo('bert-base-case'		,      _UpperCAmelCase)

          # The function raises if the revision does not exist.
          with self.assertRaisesRegex(_UpperCAmelCase		,      'is not a valid git identifier'):
               get_file_from_repo('bert-base-cased'		,      _UpperCAmelCase		,      revision='ahaha')

          __A		:					List[str]						=			get_file_from_repo('bert-base-cased'		,      _UpperCAmelCase)
          # The name is the cached name which is not very easy to test, so instead we load the content.
          __A		:					List[str]						=			json.loads(open(_UpperCAmelCase		,      'r').read())
          self.assertEqual(config['hidden_size']		,      768)


     def 				SCREAMING_SNAKE_CASE    (		self):

          '''simple docstring'''






          with tempfile.TemporaryDirectory() as tmp_dir:
               __A		:					Tuple						=			Path(_UpperCAmelCase) / 'a.txt'
               filename.touch()
               self.assertEqual(get_file_from_repo(_UpperCAmelCase		,      'a.txt')		,      str(_UpperCAmelCase))

               self.assertIsNone(get_file_from_repo(_UpperCAmelCase		,      'b.txt'))