




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    SCREAMING_SNAKE_CASE_				(       unittest.TestCase   ):



	"""simple docstring"""


	def 					lowerCamelCase__     (					self		:						Tuple	)	->       int:







						"""simple docstring"""







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

						# Download this model to make sure it's in the cache.
						__UpperCamelCase					:	str             =      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=lowerCAmelCase	) as mock_head:
											__UpperCamelCase					:	Optional[Any]             =      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	)	->       Union[str, Any]:







						"""simple docstring"""







						__UpperCamelCase					:	List[str]             =      WavaVecaFeatureExtractor.from_pretrained(
						    """https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json"""	)






@is_staging_test
class    SCREAMING_SNAKE_CASE_				(       unittest.TestCase   ):



	"""simple docstring"""


	@classmethod
	def 					lowerCamelCase__     (					cls		:						Union[str, Any]	)	->       Optional[Any]:







						"""simple docstring"""







						__UpperCamelCase					:	Optional[Any]             =      TOKEN
						HfFolder.save_token(lowerCAmelCase	)






	@classmethod
	def 					lowerCamelCase__     (					cls		:						Optional[Any]	)	->       Dict:







						"""simple docstring"""







						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		:						Optional[int]	)	->       Any:







						"""simple docstring"""







						__UpperCamelCase					:	Optional[int]             =      WavaVecaFeatureExtractor.from_pretrained(lowerCAmelCase	)
						feature_extractor.push_to_hub("""test-feature-extractor"""					,     use_auth_token=self._token	)

						__UpperCamelCase					:	Optional[int]             =      WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''	)
						for k, v in feature_extractor.__dict__.items():
											self.assertEqual(lowerCAmelCase					,     getattr(lowerCAmelCase					,     lowerCAmelCase	)	)

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

						__UpperCamelCase					:	Optional[int]             =      WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''	)
						for k, v in feature_extractor.__dict__.items():
											self.assertEqual(lowerCAmelCase					,     getattr(lowerCAmelCase					,     lowerCAmelCase	)	)






	def 					lowerCamelCase__     (					self		:						Any	)	->       List[str]:







						"""simple docstring"""







						__UpperCamelCase					:	Union[str, Any]             =      WavaVecaFeatureExtractor.from_pretrained(lowerCAmelCase	)
						feature_extractor.push_to_hub("""valid_org/test-feature-extractor"""					,     use_auth_token=self._token	)

						__UpperCamelCase					:	Optional[int]             =      WavaVecaFeatureExtractor.from_pretrained("""valid_org/test-feature-extractor"""	)
						for k, v in feature_extractor.__dict__.items():
											self.assertEqual(lowerCAmelCase					,     getattr(lowerCAmelCase					,     lowerCAmelCase	)	)

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

						__UpperCamelCase					:	Tuple             =      WavaVecaFeatureExtractor.from_pretrained("""valid_org/test-feature-extractor-org"""	)
						for k, v in feature_extractor.__dict__.items():
											self.assertEqual(lowerCAmelCase					,     getattr(lowerCAmelCase					,     lowerCAmelCase	)	)






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







						"""simple docstring"""







						CustomFeatureExtractor.register_for_auto_class()
						__UpperCamelCase					:	Dict             =      CustomFeatureExtractor.from_pretrained(lowerCAmelCase	)

						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					:	int             =      AutoFeatureExtractor.from_pretrained(
						    F'''{USER}/test-dynamic-feature-extractor'''					,     trust_remote_code=lowerCAmelCase	)
						# 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 re
from pathlib import Path
from unittest import TestCase

import pytest


@pytest.mark.integration
class    SCREAMING_SNAKE_CASE_				(       _UpperCamelCase   ):



	"""simple docstring"""


	def 					lowerCamelCase__     (					self		:						Any					,     lowerCAmelCase		:						str	)	->       List[Any]:







						"""simple docstring"""







						with open(lowerCAmelCase					,     encoding="""utf-8"""	) as input_file:
											__UpperCamelCase					:	Union[str, Any]             =      re.compile(R"""(?!.*\b(?:encoding|rb|w|wb|w+|wb+|ab|ab+)\b)(?<=\s)(open)\((.*)\)"""	)
											__UpperCamelCase					:	str             =      input_file.read()
											__UpperCamelCase					:	Optional[int]             =      regexp.search(lowerCAmelCase	)

						return match






	def 					lowerCamelCase__     (					self		:						Optional[Any]					,     lowerCAmelCase		:						str	)	->       Union[str, Any]:







						"""simple docstring"""







						with open(lowerCAmelCase					,     encoding="""utf-8"""	) as input_file:
											__UpperCamelCase					:	Dict             =      re.compile(R"""#[^\r\n]*print\(|\"[^\r\n]*print\(|\"\"\".*?print\(.*?\"\"\"|(print\()"""					,     re.DOTALL	)
											__UpperCamelCase					:	str             =      input_file.read()
											# use `re.finditer` to handle the case where the ignored groups would be matched first by `re.search`
											__UpperCamelCase					:	List[Any]             =      regexp.finditer(lowerCAmelCase	)

						__UpperCamelCase					:	int             =      [match for match in matches if match is not None and match.group(1	) is not None]
						return matches[0] if matches else None






	def 					lowerCamelCase__     (					self		:						List[Any]	)	->       List[str]:







						"""simple docstring"""







						__UpperCamelCase					:	Optional[int]             =      Path("""./datasets"""	)
						__UpperCamelCase					:	Optional[int]             =      list(dataset_paths.absolute().glob("""**/*.py"""	)	)

						for dataset in dataset_files:
											if self._no_encoding_on_file_open(str(lowerCAmelCase	)	):
																raise AssertionError(F'''open(...) must use utf-8 encoding in {dataset}'''	)






	def 					lowerCamelCase__     (					self		:						Optional[Any]	)	->       Any:







						"""simple docstring"""







						__UpperCamelCase					:	Optional[int]             =      Path("""./datasets"""	)
						__UpperCamelCase					:	Optional[int]             =      list(dataset_paths.absolute().glob("""**/*.py"""	)	)

						for dataset in dataset_files:
											if self._no_print_statements(str(lowerCAmelCase	)	):
																raise AssertionError(F'''print statement found in {dataset}. Use datasets.logger/logging instead.'''	)






