




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


_SCREAMING_SNAKE_CASE			=						get_tests_dir("fixtures")
class   __UpperCAmelCase				(     unittest.TestCase				):



					'''simple docstring'''





					def     __snake_case						(	self     :						Optional[int])    ->							List[str]:
								# A mock response for an HTTP head request to emulate server down
								A_									= mock.Mock()
								A_									= 500
								A_									= {}
								A_									= HTTPError
								A_									= {}

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



					def     __snake_case						(	self     :						Optional[Any])    ->							int:
								# This test is for deprecated behavior and can be removed in v5
								A_									= 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     __snake_case						(	cls     :						Dict)    ->							int:
								A_									= TOKEN
								HfFolder.save_token(_lowercase)



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

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

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

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



					def     __snake_case						(	self     :						Optional[int])    ->							int:
								A_									= WavaVecaFeatureExtractor.from_pretrained(_lowercase)
								feature_extractor.push_to_hub('valid_org/test-feature-extractor'      ,       use_auth_token=self._token)

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

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

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



					def     __snake_case						(	self     :						List[str])    ->							Union[str, Any]:
								CustomFeatureExtractor.register_for_auto_class()
								A_									= CustomFeatureExtractor.from_pretrained(_lowercase)

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

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

from datasets import load_dataset

from transformers.pipelines import pipeline
from transformers.testing_utils import is_pipeline_test, nested_simplify, require_torch, slow
@is_pipeline_test
@require_torch
class   __UpperCAmelCase				(     unittest.TestCase				):



					'''simple docstring'''





					@require_torch
					def     __snake_case						(	self     :						str)    ->							Optional[int]:
								A_									= pipeline(
								    task='zero-shot-audio-classification'      ,       model='hf-internal-testing/tiny-clap-htsat-unfused')
								A_									= load_dataset('ashraq/esc50')
								A_									= dataset['train']['audio'][-1]['array']
								A_									= audio_classifier(_lowercase      ,       candidate_labels=['Sound of a dog', 'Sound of vaccum cleaner'])
								self.assertEqual(
								    nested_simplify(_lowercase)      ,       [{'score': 0.5_01, 'label': 'Sound of a dog'}, {'score': 0.4_99, 'label': 'Sound of vaccum cleaner'}]      ,       )



					@unittest.skip('No models are available in TF')
					def     __snake_case						(	self     :						Tuple)    ->							str:
								pass



					@slow
					@require_torch
					def     __snake_case						(	self     :						Optional[Any])    ->							Tuple:
								A_									= pipeline(
								    task='zero-shot-audio-classification'      ,       model='laion/clap-htsat-unfused'      ,       )
								# This is an audio of a dog
								A_									= load_dataset('ashraq/esc50')
								A_									= dataset['train']['audio'][-1]['array']
								A_									= audio_classifier(_lowercase      ,       candidate_labels=['Sound of a dog', 'Sound of vaccum cleaner'])

								self.assertEqual(
								    nested_simplify(_lowercase)      ,       [
								        {'score': 0.9_99, 'label': 'Sound of a dog'},
								        {'score': 0.0_01, 'label': 'Sound of vaccum cleaner'},
								    ]      ,       )

								A_									= audio_classifier([audio] * 5      ,       candidate_labels=['Sound of a dog', 'Sound of vaccum cleaner'])
								self.assertEqual(
								    nested_simplify(_lowercase)      ,       [
								        [
								            {'score': 0.9_99, 'label': 'Sound of a dog'},
								            {'score': 0.0_01, 'label': 'Sound of vaccum cleaner'},
								        ],
								    ]
								    * 5      ,       )
								A_									= audio_classifier(
								    [audio] * 5      ,       candidate_labels=['Sound of a dog', 'Sound of vaccum cleaner']      ,       batch_size=5)
								self.assertEqual(
								    nested_simplify(_lowercase)      ,       [
								        [
								            {'score': 0.9_99, 'label': 'Sound of a dog'},
								            {'score': 0.0_01, 'label': 'Sound of vaccum cleaner'},
								        ],
								    ]
								    * 5      ,       )



					@unittest.skip('No models are available in TF')
					def     __snake_case						(	self     :						List[str])    ->							str:
								pass




