




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


a_        =       get_tests_dir('fixtures')

class   UpperCAmelCase_    (			unittest.TestCase		):
		def 	_lowerCamelCase							(						self       )						->       Optional[int]:
					# A mock response for an HTTP head request to emulate server down
					__lowercase			:  Union[str, Any]								=       mock.Mock()
					__lowercase			:  List[Any]								=       5_00
					__lowercase			:  Union[str, Any]								=       {}
					__lowercase			:  int								=       HTTPError
					__lowercase			:  Any								=       {}

					# Download this model to make sure it's in the cache.
					__lowercase			:  Tuple								=       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=UpperCamelCase_       ) as mock_head:
								__lowercase			:  int								=       WavaVecaFeatureExtractor.from_pretrained('''hf-internal-testing/tiny-random-wav2vec2'''       )
								# This check we did call the fake head request
								mock_head.assert_called()
		def 	_lowerCamelCase							(						self       )						->       Any:
					# This test is for deprecated behavior and can be removed in v5
					__lowercase			:  Optional[Any]								=       WavaVecaFeatureExtractor.from_pretrained(
					    '''https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'''       )


@is_staging_test
class   UpperCAmelCase_    (			unittest.TestCase		):
		@classmethod
		def 	_lowerCamelCase							(						cls       )						->       Any:
					__lowercase			:  Union[str, Any]								=       TOKEN
					HfFolder.save_token(UpperCamelCase_       )
		@classmethod
		def 	_lowerCamelCase							(						cls       )						->       Optional[int]:
					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 	_lowerCamelCase							(						self       )						->       Tuple:
					__lowercase			:  Union[str, Any]								=       WavaVecaFeatureExtractor.from_pretrained(UpperCamelCase_       )
					feature_extractor.push_to_hub('''test-feature-extractor'''				,						use_auth_token=self._token       )

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

					# 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(
								    UpperCamelCase_				,						repo_id='''test-feature-extractor'''				,						push_to_hub=UpperCamelCase_				,						use_auth_token=self._token       )

					__lowercase			:  List[Any]								=       WavaVecaFeatureExtractor.from_pretrained(F"""{USER}/test-feature-extractor"""       )
					for k, v in feature_extractor.__dict__.items():
								self.assertEqual(UpperCamelCase_				,						getattr(UpperCamelCase_				,						UpperCamelCase_       )       )
		def 	_lowerCamelCase							(						self       )						->       Tuple:
					__lowercase			:  Dict								=       WavaVecaFeatureExtractor.from_pretrained(UpperCamelCase_       )
					feature_extractor.push_to_hub('''valid_org/test-feature-extractor'''				,						use_auth_token=self._token       )

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

					# 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(
								    UpperCamelCase_				,						repo_id='''valid_org/test-feature-extractor-org'''				,						push_to_hub=UpperCamelCase_				,						use_auth_token=self._token       )

					__lowercase			:  Dict								=       WavaVecaFeatureExtractor.from_pretrained('''valid_org/test-feature-extractor-org'''       )
					for k, v in feature_extractor.__dict__.items():
								self.assertEqual(UpperCamelCase_				,						getattr(UpperCamelCase_				,						UpperCamelCase_       )       )
		def 	_lowerCamelCase							(						self       )						->       str:
					CustomFeatureExtractor.register_for_auto_class()
					__lowercase			:  Dict								=       CustomFeatureExtractor.from_pretrained(UpperCamelCase_       )

					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'''}				,						)

					__lowercase			:  List[str]								=       AutoFeatureExtractor.from_pretrained(
					    F"""{USER}/test-dynamic-feature-extractor"""				,						trust_remote_code=UpperCamelCase_       )
					# 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 pytest
import requests

from datasets.utils.file_utils import http_head

from .utils import OfflineSimulationMode, RequestWouldHangIndefinitelyError, offline







@pytest.mark.integration
def       a_				(	):
							with offline(OfflineSimulationMode.CONNECTION_TIMES_OUT ):
														with pytest.raises(UpperCamelCase_ ):
																					requests.request("GET"						,  "https://huggingface.co" )
														with pytest.raises(requests.exceptions.ConnectTimeout ):
																					requests.request("GET"						,  "https://huggingface.co"						,  timeout=1.0 )







@pytest.mark.integration
def       a_				(	):
							with offline(OfflineSimulationMode.CONNECTION_FAILS ):
														with pytest.raises(requests.exceptions.ConnectionError ):
																					requests.request("GET"						,  "https://huggingface.co" )







def       a_				(	):
							with offline(OfflineSimulationMode.HF_DATASETS_OFFLINE_SET_TO_1 ):
														with pytest.raises(UpperCamelCase_ ):
																					http_head("https://huggingface.co" )





