



"""simple docstring"""







import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


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

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


lowercase__		:    Optional[int]				   =     get_tests_dir("""fixtures""")







class 				UpperCamelCase__       (     unittest.TestCase  ):




      """simple docstring"""




      def 					SCREAMING_SNAKE_CASE__			(				self    :				Dict	):
        # A mock response for an HTTP head request to emulate server down
        lowerCAmelCase_       :    List[str]								=			mock.Mock()
        lowerCAmelCase_       :    int								=			5_0_0
        lowerCAmelCase_       :    Any								=			{}
        lowerCAmelCase_       :    Union[str, Any]								=			HTTPError
        lowerCAmelCase_       :    Dict								=			{}

        # Download this model to make sure it's in the cache.
        lowerCAmelCase_       :    Any								=			WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'	)
        # Under the mock environment we get a 500 error when trying to reach the model.
        with mock.patch('requests.Session.request'   ,      return_value=SCREAMING_SNAKE_CASE_	) as mock_head:
          lowerCAmelCase_       :    List[str]								=			WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'	)
          # This check we did call the fake head request
          mock_head.assert_called()





      def 					SCREAMING_SNAKE_CASE__			(				self    :				Optional[int]	):
        # This test is for deprecated behavior and can be removed in v5
        lowerCAmelCase_       :    Optional[int]								=			WavaVecaFeatureExtractor.from_pretrained(
            'https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'	)







@is_staging_test
class 				UpperCamelCase__       (     unittest.TestCase  ):




      """simple docstring"""




      @classmethod
      def 					SCREAMING_SNAKE_CASE__			(				cls    :				int	):
        lowerCAmelCase_       :    List[str]								=			TOKEN
        HfFolder.save_token(SCREAMING_SNAKE_CASE_	)



      @classmethod
      def 					SCREAMING_SNAKE_CASE__			(				cls    :				Tuple	):
        try:
          delete_repo(token=cls._token   ,      repo_id='test-feature-extractor'	)
        except HTTPError:
          pass

        try:
          delete_repo(token=cls._token   ,      repo_id='valid_org/test-feature-extractor-org'	)
        except HTTPError:
          pass

        try:
          delete_repo(token=cls._token   ,      repo_id='test-dynamic-feature-extractor'	)
        except HTTPError:
          pass



      def 					SCREAMING_SNAKE_CASE__			(				self    :				List[Any]	):
        lowerCAmelCase_       :    Optional[int]								=			WavaVecaFeatureExtractor.from_pretrained(SCREAMING_SNAKE_CASE_	)
        feature_extractor.push_to_hub('test-feature-extractor'   ,      use_auth_token=self._token	)

        lowerCAmelCase_       :    int								=			WavaVecaFeatureExtractor.from_pretrained(F"{USER}/test-feature-extractor"	)
        for k, v in feature_extractor.__dict__.items():
          self.assertEqual(SCREAMING_SNAKE_CASE_   ,      getattr(SCREAMING_SNAKE_CASE_   ,      SCREAMING_SNAKE_CASE_	)	)

        # Reset repo
        delete_repo(token=self._token   ,      repo_id='test-feature-extractor'	)

        # Push to hub via save_pretrained
        with tempfile.TemporaryDirectory() as tmp_dir:
          feature_extractor.save_pretrained(
              SCREAMING_SNAKE_CASE_   ,      repo_id='test-feature-extractor'   ,      push_to_hub=SCREAMING_SNAKE_CASE_   ,      use_auth_token=self._token	)

        lowerCAmelCase_       :    str								=			WavaVecaFeatureExtractor.from_pretrained(F"{USER}/test-feature-extractor"	)
        for k, v in feature_extractor.__dict__.items():
          self.assertEqual(SCREAMING_SNAKE_CASE_   ,      getattr(SCREAMING_SNAKE_CASE_   ,      SCREAMING_SNAKE_CASE_	)	)



      def 					SCREAMING_SNAKE_CASE__			(				self    :				Any	):
        lowerCAmelCase_       :    Union[str, Any]								=			WavaVecaFeatureExtractor.from_pretrained(SCREAMING_SNAKE_CASE_	)
        feature_extractor.push_to_hub('valid_org/test-feature-extractor'   ,      use_auth_token=self._token	)

        lowerCAmelCase_       :    List[str]								=			WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor'	)
        for k, v in feature_extractor.__dict__.items():
          self.assertEqual(SCREAMING_SNAKE_CASE_   ,      getattr(SCREAMING_SNAKE_CASE_   ,      SCREAMING_SNAKE_CASE_	)	)

        # Reset repo
        delete_repo(token=self._token   ,      repo_id='valid_org/test-feature-extractor'	)

        # Push to hub via save_pretrained
        with tempfile.TemporaryDirectory() as tmp_dir:
          feature_extractor.save_pretrained(
              SCREAMING_SNAKE_CASE_   ,      repo_id='valid_org/test-feature-extractor-org'   ,      push_to_hub=SCREAMING_SNAKE_CASE_   ,      use_auth_token=self._token	)

        lowerCAmelCase_       :    Optional[Any]								=			WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor-org'	)
        for k, v in feature_extractor.__dict__.items():
          self.assertEqual(SCREAMING_SNAKE_CASE_   ,      getattr(SCREAMING_SNAKE_CASE_   ,      SCREAMING_SNAKE_CASE_	)	)





      def 					SCREAMING_SNAKE_CASE__			(				self    :				Optional[Any]	):
        CustomFeatureExtractor.register_for_auto_class()
        lowerCAmelCase_       :    Union[str, Any]								=			CustomFeatureExtractor.from_pretrained(SCREAMING_SNAKE_CASE_	)

        feature_extractor.push_to_hub('test-dynamic-feature-extractor'   ,      use_auth_token=self._token	)

        # This has added the proper auto_map field to the config
        self.assertDictEqual(
            feature_extractor.auto_map   ,      {'AutoFeatureExtractor': 'custom_feature_extraction.CustomFeatureExtractor'}   ,      )

        lowerCAmelCase_       :    Optional[Any]								=			AutoFeatureExtractor.from_pretrained(
            F"{USER}/test-dynamic-feature-extractor"   ,      trust_remote_code=SCREAMING_SNAKE_CASE_	)
        # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
        self.assertEqual(new_feature_extractor.__class__.__name__   ,      'CustomFeatureExtractor'	)










"""simple docstring"""







import os

from huggingface_hub.constants import HUGGINGFACE_HUB_CACHE, hf_cache_home


lowercase__		:    int				   =     HUGGINGFACE_HUB_CACHE


lowercase__		:    Tuple				   =     """config.json"""
lowercase__		:    Union[str, Any]				   =     """diffusion_pytorch_model.bin"""
lowercase__		:    List[Any]				   =     """diffusion_flax_model.msgpack"""
lowercase__		:    List[str]				   =     """model.onnx"""
lowercase__		:    List[Any]				   =     """diffusion_pytorch_model.safetensors"""
lowercase__		:    Dict				   =     """weights.pb"""
lowercase__		:    List[Any]				   =     """https://huggingface.co"""
lowercase__		:    List[Any]				   =     default_cache_path
lowercase__		:    Tuple				   =     """diffusers_modules"""
lowercase__		:    Tuple				   =     os.getenv("""HF_MODULES_CACHE""", os.path.join(hf_cache_home, """modules"""))
lowercase__		:    List[str]				   =     ["""fp16""", """non-ema"""]
lowercase__		:    Optional[Any]				   =     """.self_attn"""





