



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


_lowerCAmelCase    : Dict							=							get_tests_dir("fixtures")






class      UpperCAmelCase_  (    unittest.TestCase						):



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

           # Download this model to make sure it's in the cache.
           _UpperCAmelCase      :     Optional[Any]        =   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:
                _UpperCAmelCase      :     Optional[int]        =   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					:     List[str]						):
           # This test is for deprecated behavior and can be removed in v5
           _UpperCAmelCase      :     Optional[int]        =   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 						snake_case_ (					cls					:     Optional[int]						):
           _UpperCAmelCase      :     Any        =   TOKEN
           HfFolder.save_token(UpperCamelCase_						)



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

           _UpperCAmelCase      :     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						)

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



      def 						snake_case_ (					self					:     Optional[Any]						):
           _UpperCAmelCase      :     Tuple        =   WavaVecaFeatureExtractor.from_pretrained(UpperCamelCase_						)
           feature_extractor.push_to_hub("valid_org/test-feature-extractor"					,     use_auth_token=self._token						)

           _UpperCAmelCase      :     int        =   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						)

           _UpperCAmelCase      :     List[str]        =   WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor-org"						)
           for k, v in feature_extractor.__dict__.items():
                self.assertEqual(UpperCamelCase_					,     getattr(UpperCamelCase_					,     UpperCamelCase_						)						)



      def 						snake_case_ (					self					:     Dict						):
           CustomFeatureExtractor.register_for_auto_class()
           _UpperCAmelCase      :     List[str]        =   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"}					,     )

           _UpperCAmelCase      :     int        =   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 io
import json
import unittest

from parameterized import parameterized

from transformers import FSMTForConditionalGeneration, FSMTTokenizer
from transformers.testing_utils import get_tests_dir, require_torch, slow, torch_device
from utils import calculate_bleu


_lowerCAmelCase    : Union[str, Any]							=							get_tests_dir() + "/test_data/fsmt/fsmt_val_data.json"
with io.open(filename, "r", encoding="utf-8") as f:
 _lowerCAmelCase    : Tuple							=							json.load(f)






@require_torch
class      UpperCAmelCase_  (    unittest.TestCase						):



      def 						snake_case_ (					self					:     str					,     A					:     Union[str, Any]						):
           return FSMTTokenizer.from_pretrained(A						)



      def 						snake_case_ (					self					:     Union[str, Any]					,     A					:     Union[str, Any]						):
           _UpperCAmelCase      :     List[Any]        =   FSMTForConditionalGeneration.from_pretrained(A						).to(A						)
           if torch_device == "cuda":
                model.half()
           return model



      @parameterized.expand(
          [
              ["en-ru", 26.0],
              ["ru-en", 22.0],
              ["en-de", 22.0],
              ["de-en", 29.0],
          ]						)
      @slow
      def 						snake_case_ (					self					:     Any					,     A					:     Dict					,     A					:     List[str]						):
           # note: this test is not testing the best performance since it only evals a small batch
           # but it should be enough to detect a regression in the output quality
           _UpperCAmelCase      :     Any        =   f'facebook/wmt19-{pair}'
           _UpperCAmelCase      :     Dict        =   self.get_tokenizer(A						)
           _UpperCAmelCase      :     Optional[int]        =   self.get_model(A						)

           _UpperCAmelCase      :     int        =   bleu_data[pair]["src"]
           _UpperCAmelCase      :     Optional[int]        =   bleu_data[pair]["tgt"]

           _UpperCAmelCase      :     List[str]        =   tokenizer(A					,     return_tensors="pt"					,     truncation=A					,     padding="longest"						).to(A						)
           _UpperCAmelCase      :     List[str]        =   model.generate(
               input_ids=batch.input_ids					,     num_beams=8					,     )
           _UpperCAmelCase      :     Any        =   tokenizer.batch_decode(
               A					,     skip_special_tokens=A					,     clean_up_tokenization_spaces=A						)
           _UpperCAmelCase      :     Any        =   calculate_bleu(A					,     A						)
           print(A						)
           self.assertGreaterEqual(scores["bleu"]					,     A						)
