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


__UpperCamelCase									=			get_tests_dir('''fixtures''')




class  UpperCamelCase					(				unittest.TestCase   ):







       def 		a_						( self)      ->      int:
            # A mock response for an HTTP head request to emulate server down
            snake_case_					       =					mock.Mock()
            snake_case_					       =					500
            snake_case_					       =					{}
            snake_case_					       =					HTTPError
            snake_case_					       =					{}

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







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







       @classmethod
       def 		a_						( cls)      ->      Any:
            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 		a_						( self)      ->      Any:
            snake_case_					       =					WavaVecaFeatureExtractor.from_pretrained(lowerCAmelCase__)
            feature_extractor.push_to_hub('test-feature-extractor',							use_auth_token=self._token)

            snake_case_					       =					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)

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







       def 		a_						( self)      ->      Dict:
            snake_case_					       =					WavaVecaFeatureExtractor.from_pretrained(lowerCAmelCase__)
            feature_extractor.push_to_hub('valid_org/test-feature-extractor',							use_auth_token=self._token)

            snake_case_					       =					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)

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







       def 		a_						( self)      ->      List[Any]:
            CustomFeatureExtractor.register_for_auto_class()
            snake_case_					       =					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'},							)

            snake_case_					       =					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')






"""simple docstring"""
import unittest
from queue import Empty
from threading import Thread

from transformers import AutoTokenizer, TextIteratorStreamer, TextStreamer, is_torch_available
from transformers.testing_utils import CaptureStdout, require_torch, torch_device

from ..test_modeling_common import ids_tensor


if is_torch_available():
  import torch

  from transformers import AutoModelForCausalLM
@require_torch
class 							_UpperCAmelCase							( unittest.TestCase      ):





     '''simple docstring'''



     def  __lowerCAmelCase	(					self		)							->					List[str]:
         _UpperCAmelCase							:  Any											=    AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''		)
         _UpperCAmelCase							:  Dict											=    AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''		).to(A		)
         _UpperCAmelCase							:  List[str]											=    -1

         _UpperCAmelCase							:  List[str]											=    ids_tensor((1, 5)       ,	vocab_size=model.config.vocab_size		).to(A		)
         _UpperCAmelCase							:  List[str]											=    model.generate(A       ,	max_new_tokens=1_0       ,	do_sample=A		)
         _UpperCAmelCase							:  List[Any]											=    tokenizer.decode(greedy_ids[0]		)

         with CaptureStdout() as cs:
             _UpperCAmelCase							:  str											=    TextStreamer(A		)
             model.generate(A       ,	max_new_tokens=1_0       ,	do_sample=A       ,	streamer=A		)
         # The greedy text should be printed to stdout, except for the final "\n" in the streamer
         _UpperCAmelCase							:  List[str]											=    cs.out[:-1]

         self.assertEqual(A       ,	A		)





     def  __lowerCAmelCase	(					self		)							->					Dict:
         _UpperCAmelCase							:  List[str]											=    AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''		)
         _UpperCAmelCase							:  List[Any]											=    AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''		).to(A		)
         _UpperCAmelCase							:  List[Any]											=    -1

         _UpperCAmelCase							:  Union[str, Any]											=    ids_tensor((1, 5)       ,	vocab_size=model.config.vocab_size		).to(A		)
         _UpperCAmelCase							:  List[Any]											=    model.generate(A       ,	max_new_tokens=1_0       ,	do_sample=A		)
         _UpperCAmelCase							:  str											=    tokenizer.decode(greedy_ids[0]		)

         _UpperCAmelCase							:  Union[str, Any]											=    TextIteratorStreamer(A		)
         _UpperCAmelCase							:  Any											=    {'''input_ids''': input_ids, '''max_new_tokens''': 1_0, '''do_sample''': False, '''streamer''': streamer}
         _UpperCAmelCase							:  Any											=    Thread(target=model.generate       ,	kwargs=A		)
         thread.start()
         _UpperCAmelCase							:  Any											=    ''''''
         for new_text in streamer:
             streamer_text += new_text

         self.assertEqual(A       ,	A		)





     def  __lowerCAmelCase	(					self		)							->					str:
         _UpperCAmelCase							:  Any											=    AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''		)
         _UpperCAmelCase							:  str											=    AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''		).to(A		)
         _UpperCAmelCase							:  Any											=    -1

         _UpperCAmelCase							:  Dict											=    ids_tensor((1, 5)       ,	vocab_size=model.config.vocab_size		).to(A		)
         _UpperCAmelCase							:  Dict											=    model.generate(A       ,	max_new_tokens=1_0       ,	do_sample=A		)
         _UpperCAmelCase							:  Dict											=    greedy_ids[:, input_ids.shape[1] :]
         _UpperCAmelCase							:  List[str]											=    tokenizer.decode(new_greedy_ids[0]		)

         with CaptureStdout() as cs:
             _UpperCAmelCase							:  Any											=    TextStreamer(A       ,	skip_prompt=A		)
             model.generate(A       ,	max_new_tokens=1_0       ,	do_sample=A       ,	streamer=A		)
         # The greedy text should be printed to stdout, except for the final "\n" in the streamer
         _UpperCAmelCase							:  Union[str, Any]											=    cs.out[:-1]

         self.assertEqual(A       ,	A		)





     def  __lowerCAmelCase	(					self		)							->					Optional[int]:
         # Tests that we can pass `decode_kwargs` to the streamer to control how the tokens are decoded. Must be tested
         # with actual models -- the dummy models' tokenizers are not aligned with their models, and
         # `skip_special_tokens=True` has no effect on them
         _UpperCAmelCase							:  int											=    AutoTokenizer.from_pretrained('''distilgpt2'''		)
         _UpperCAmelCase							:  Union[str, Any]											=    AutoModelForCausalLM.from_pretrained('''distilgpt2'''		).to(A		)
         _UpperCAmelCase							:  Tuple											=    -1

         _UpperCAmelCase							:  int											=    torch.ones((1, 5)       ,	device=A		).long() * model.config.bos_token_id
         with CaptureStdout() as cs:
             _UpperCAmelCase							:  Optional[Any]											=    TextStreamer(A       ,	skip_special_tokens=A		)
             model.generate(A       ,	max_new_tokens=1       ,	do_sample=A       ,	streamer=A		)

         # The prompt contains a special token, so the streamer should not print it. As such, the output text, when
         # re-tokenized, must only contain one token
         _UpperCAmelCase							:  Tuple											=    cs.out[:-1]  # Remove the final "\n"
         _UpperCAmelCase							:  int											=    tokenizer(A       ,	return_tensors='''pt'''		)
         self.assertEqual(streamer_text_tokenized.input_ids.shape       ,	(1, 1)		)





     def  __lowerCAmelCase	(					self		)							->					Union[str, Any]:
         _UpperCAmelCase							:  Any											=    AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''		)
         _UpperCAmelCase							:  Any											=    AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''		).to(A		)
         _UpperCAmelCase							:  Dict											=    -1

         _UpperCAmelCase							:  str											=    ids_tensor((1, 5)       ,	vocab_size=model.config.vocab_size		).to(A		)
         _UpperCAmelCase							:  List[Any]											=    TextIteratorStreamer(A       ,	timeout=0.001		)
         _UpperCAmelCase							:  Union[str, Any]											=    {'''input_ids''': input_ids, '''max_new_tokens''': 1_0, '''do_sample''': False, '''streamer''': streamer}
         _UpperCAmelCase							:  Optional[Any]											=    Thread(target=model.generate       ,	kwargs=A		)
         thread.start()

         # The streamer will timeout after 0.001 seconds, so an exception will be raised
         with self.assertRaises(A		):
             _UpperCAmelCase							:  Optional[Any]											=    ''''''
             for new_text in streamer:
                 streamer_text += new_text


