



"""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 :				List[str]					    = get_tests_dir("fixtures")
class       _a			(       unittest.TestCase):
			"""simple docstring"""



			def 			lowercase__    (		self  :     List[str] )->Tuple:
							# A mock response for an HTTP head request to emulate server down
							_UpperCAmelCase      		=     mock.Mock()
							_UpperCAmelCase      		=     5_0_0
							_UpperCAmelCase      		=     {}
							_UpperCAmelCase      		=     HTTPError
							_UpperCAmelCase      		=     {}

							# Download this model to make sure it's in the cache.
							_UpperCAmelCase      		=     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=_a ) as mock_head:
											_UpperCAmelCase      		=     WavaVecaFeatureExtractor.from_pretrained('''hf-internal-testing/tiny-random-wav2vec2''' )
											# This check we did call the fake head request
											mock_head.assert_called()



			def 			lowercase__    (		self  :     Any )->Tuple:
							# This test is for deprecated behavior and can be removed in v5
							_UpperCAmelCase      		=     WavaVecaFeatureExtractor.from_pretrained(
							    '''https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json''' )



@is_staging_test
class       _a			(       unittest.TestCase):
			"""simple docstring"""



			@classmethod
			def 			lowercase__    (		cls  :     Optional[int] )->Optional[Any]:
							_UpperCAmelCase      		=     TOKEN
							HfFolder.save_token(_a )



			@classmethod
			def 			lowercase__    (		cls  :     Tuple )->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 			lowercase__    (		self  :     Optional[Any] )->str:
							_UpperCAmelCase      		=     WavaVecaFeatureExtractor.from_pretrained(_a )
							feature_extractor.push_to_hub('''test-feature-extractor'''					, use_auth_token=self._token )

							_UpperCAmelCase      		=     WavaVecaFeatureExtractor.from_pretrained(F'{USER}/test-feature-extractor' )
							for k, v in feature_extractor.__dict__.items():
											self.assertEqual(_a					, getattr(_a					, _a ) )

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

							_UpperCAmelCase      		=     WavaVecaFeatureExtractor.from_pretrained(F'{USER}/test-feature-extractor' )
							for k, v in feature_extractor.__dict__.items():
											self.assertEqual(_a					, getattr(_a					, _a ) )



			def 			lowercase__    (		self  :     int )->Dict:
							_UpperCAmelCase      		=     WavaVecaFeatureExtractor.from_pretrained(_a )
							feature_extractor.push_to_hub('''valid_org/test-feature-extractor'''					, use_auth_token=self._token )

							_UpperCAmelCase      		=     WavaVecaFeatureExtractor.from_pretrained('''valid_org/test-feature-extractor''' )
							for k, v in feature_extractor.__dict__.items():
											self.assertEqual(_a					, getattr(_a					, _a ) )

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

							_UpperCAmelCase      		=     WavaVecaFeatureExtractor.from_pretrained('''valid_org/test-feature-extractor-org''' )
							for k, v in feature_extractor.__dict__.items():
											self.assertEqual(_a					, getattr(_a					, _a ) )



			def 			lowercase__    (		self  :     int )->Any:
							CustomFeatureExtractor.register_for_auto_class()
							_UpperCAmelCase      		=     CustomFeatureExtractor.from_pretrained(_a )

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

							_UpperCAmelCase      		=     AutoFeatureExtractor.from_pretrained(
							    F'{USER}/test-dynamic-feature-extractor'					, trust_remote_code=_a )
							# 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''' )






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

import transformers
from transformers import (
    CONFIG_MAPPING,
    IMAGE_PROCESSOR_MAPPING,
    AutoConfig,
    AutoImageProcessor,
    CLIPConfig,
    CLIPImageProcessor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER


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

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_image_processing import CustomImageProcessor  # noqa E402

class 							UpperCAmelCase_						(     unittest.TestCase  ):
							"""simple docstring"""







							def 							__lowercase							(     self			)       ->       List[Any]:
												_a      :       int   =						0


							def 							__lowercase							(     self			)       ->       List[str]:
												_a      :       Dict   =						AutoImageProcessor.from_pretrained('''openai/clip-vit-base-patch32'''			)
												self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Union[str, Any]:
												with tempfile.TemporaryDirectory() as tmpdirname:
																	_a      :       Tuple   =						Path(_a			) / '''preprocessor_config.json'''
																	_a      :       Optional[Any]   =						Path(_a			) / '''config.json'''
																	json.dump(
																	    {'''image_processor_type''': '''CLIPImageProcessor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)
																	json.dump({'''model_type''': '''clip'''}			,				open(_a			,				'''w'''			)			)

																	_a      :       List[str]   =						AutoImageProcessor.from_pretrained(_a			)
																	self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Optional[Any]:
												# Ensure we can load the image processor from the feature extractor config
												with tempfile.TemporaryDirectory() as tmpdirname:
																	_a      :       Optional[int]   =						Path(_a			) / '''preprocessor_config.json'''
																	_a      :       Any   =						Path(_a			) / '''config.json'''
																	json.dump(
																	    {'''feature_extractor_type''': '''CLIPFeatureExtractor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)
																	json.dump({'''model_type''': '''clip'''}			,				open(_a			,				'''w'''			)			)

																	_a      :       Optional[Any]   =						AutoImageProcessor.from_pretrained(_a			)
																	self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Optional[Any]:
												with tempfile.TemporaryDirectory() as tmpdirname:
																	_a      :       Dict   =						CLIPConfig()

																	# Create a dummy config file with image_proceesor_type
																	_a      :       Tuple   =						Path(_a			) / '''preprocessor_config.json'''
																	_a      :       List[str]   =						Path(_a			) / '''config.json'''
																	json.dump(
																	    {'''image_processor_type''': '''CLIPImageProcessor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)
																	json.dump({'''model_type''': '''clip'''}			,				open(_a			,				'''w'''			)			)

																	# remove image_processor_type to make sure config.json alone is enough to load image processor locally
																	_a      :       Tuple   =						AutoImageProcessor.from_pretrained(_a			).to_dict()

																	config_dict.pop('''image_processor_type'''			)
																	_a      :       Tuple   =						CLIPImageProcessor(**_a			)

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

																	_a      :       List[str]   =						AutoImageProcessor.from_pretrained(_a			)

																	# make sure private variable is not incorrectly saved
																	_a      :       Optional[int]   =						json.loads(config.to_json_string()			)
																	self.assertTrue('''_processor_class''' not in dict_as_saved			)

												self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Dict:
												with tempfile.TemporaryDirectory() as tmpdirname:
																	_a      :       Optional[int]   =						Path(_a			) / '''preprocessor_config.json'''
																	json.dump(
																	    {'''image_processor_type''': '''CLIPImageProcessor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)

																	_a      :       List[str]   =						AutoImageProcessor.from_pretrained(_a			)
																	self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Any:
												with self.assertRaisesRegex(
												    _a			,				'''clip-base is not a local folder and is not a valid model identifier'''			):
																	_a      :       Dict   =						AutoImageProcessor.from_pretrained('''clip-base'''			)


							def 							__lowercase							(     self			)       ->       List[Any]:
												with self.assertRaisesRegex(
												    _a			,				R'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'''			):
																	_a      :       List[str]   =						AutoImageProcessor.from_pretrained(_a			,				revision='''aaaaaa'''			)


							def 							__lowercase							(     self			)       ->       Dict:
												with self.assertRaisesRegex(
												    _a			,				'''hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json.'''			,				):
																	_a      :       Optional[int]   =						AutoImageProcessor.from_pretrained('''hf-internal-testing/config-no-model'''			)


							def 							__lowercase							(     self			)       ->       Union[str, Any]:
												# If remote code is not set, we will time out when asking whether to load the model.
												with self.assertRaises(_a			):
																	_a      :       str   =						AutoImageProcessor.from_pretrained('''hf-internal-testing/test_dynamic_image_processor'''			)
												# If remote code is disabled, we can't load this config.
												with self.assertRaises(_a			):
																	_a      :       Optional[Any]   =						AutoImageProcessor.from_pretrained(
																	    '''hf-internal-testing/test_dynamic_image_processor'''			,				trust_remote_code=_a			)

												_a      :       Union[str, Any]   =						AutoImageProcessor.from_pretrained(
												    '''hf-internal-testing/test_dynamic_image_processor'''			,				trust_remote_code=_a			)
												self.assertEqual(image_processor.__class__.__name__			,				'''NewImageProcessor'''			)

												# Test image processor can be reloaded.
												with tempfile.TemporaryDirectory() as tmp_dir:
																	image_processor.save_pretrained(_a			)
																	_a      :       Optional[Any]   =						AutoImageProcessor.from_pretrained(_a			,				trust_remote_code=_a			)
												self.assertEqual(reloaded_image_processor.__class__.__name__			,				'''NewImageProcessor'''			)


							def 							__lowercase							(     self			)       ->       Dict:
												try:
																	AutoConfig.register('''custom'''			,				_a			)
																	AutoImageProcessor.register(_a			,				_a			)
																	# Trying to register something existing in the Transformers library will raise an error
																	with self.assertRaises(_a			):
																						AutoImageProcessor.register(_a			,				_a			)

																	with tempfile.TemporaryDirectory() as tmpdirname:
																						_a      :       int   =						Path(_a			) / '''preprocessor_config.json'''
																						_a      :       int   =						Path(_a			) / '''config.json'''
																						json.dump(
																						    {'''feature_extractor_type''': '''CLIPFeatureExtractor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)
																						json.dump({'''model_type''': '''clip'''}			,				open(_a			,				'''w'''			)			)

																						_a      :       int   =						CustomImageProcessor.from_pretrained(_a			)

																	# Now that the config is registered, it can be used as any other config with the auto-API
																	with tempfile.TemporaryDirectory() as tmp_dir:
																						image_processor.save_pretrained(_a			)
																						_a      :       Optional[Any]   =						AutoImageProcessor.from_pretrained(_a			)
																						self.assertIsInstance(_a			,				_a			)

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




							def 							__lowercase							(     self			)       ->       Union[str, Any]:

												class 							UpperCAmelCase_						(     __lowercase  ):
																			"""simple docstring"""







																			UpperCAmelCase__			:		Tuple             =       True

												try:
																	AutoConfig.register('''custom'''			,				_a			)
																	AutoImageProcessor.register(_a			,				_a			)
																	# If remote code is not set, the default is to use local
																	_a      :       str   =						AutoImageProcessor.from_pretrained('''hf-internal-testing/test_dynamic_image_processor'''			)
																	self.assertEqual(image_processor.__class__.__name__			,				'''NewImageProcessor'''			)
																	self.assertTrue(image_processor.is_local			)

																	# If remote code is disabled, we load the local one.
																	_a      :       int   =						AutoImageProcessor.from_pretrained(
																	    '''hf-internal-testing/test_dynamic_image_processor'''			,				trust_remote_code=_a			)
																	self.assertEqual(image_processor.__class__.__name__			,				'''NewImageProcessor'''			)
																	self.assertTrue(image_processor.is_local			)

																	# If remote is enabled, we load from the Hub
																	_a      :       Dict   =						AutoImageProcessor.from_pretrained(
																	    '''hf-internal-testing/test_dynamic_image_processor'''			,				trust_remote_code=_a			)
																	self.assertEqual(image_processor.__class__.__name__			,				'''NewImageProcessor'''			)
																	self.assertTrue(not hasattr(_a			,				'''is_local'''			)			)

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



