





import json
import os
import re
import shutil
import tempfile
import unittest
from typing import Tuple

from transformers import AddedToken, BatchEncoding, ByTaTokenizer
from transformers.utils import cached_property, is_tf_available, is_torch_available

from ...test_tokenization_common import TokenizerTesterMixin


if is_torch_available():
      snake_case								=							"""pt"""
elif is_tf_available():
      snake_case								=							"""tf"""
else:
      snake_case								=							"""jax"""





class 					A_					(   _a       ,						unittest.TestCase						):







       """simple docstring"""





       SCREAMING_SNAKE_CASE_   :			Dict  			=	ByTaTokenizer
       SCREAMING_SNAKE_CASE_   :			List[str]  			=	False







       def  __UpperCAmelCase			(     self						:						Any	)    ->       Tuple:
         super().setUp()
         _lowercase						    =			ByTaTokenizer()
         tokenizer.save_pretrained(self.tmpdirname	)







       @cached_property
       def  __UpperCAmelCase			(     self						:						int	)    ->       Union[str, Any]:
         return ByTaTokenizer.from_pretrained('google/byt5-small'	)







       def  __UpperCAmelCase			(     self						:						Optional[int]    ,**__A						:						int	)    ->       Any:
         return self.tokenizer_class.from_pretrained(self.tmpdirname    ,**A_	)







       def  __UpperCAmelCase			(     self						:						Any    ,__A						:						List[Any]    ,__A						:						List[str]=False    ,__A						:						List[Any]=20    ,__A						:						List[Any]=5	)    ->       List[Any]:
         _lowercase						    =			[]
         for i in range(len(A_	)	):
           try:
             _lowercase						    =			tokenizer.decode([i]    ,clean_up_tokenization_spaces=A_	)
           except UnicodeDecodeError:
             pass
           toks.append((i, tok)	)

         _lowercase						    =			list(filter(lambda __A	: re.match(r'^[ a-zA-Z]+$'    ,t[1]	)    ,A_	)	)
         _lowercase						    =			list(filter(lambda __A	: [t[0]] == tokenizer.encode(t[1]    ,add_special_tokens=A_	)    ,A_	)	)
         if max_length is not None and len(A_	) > max_length:
           _lowercase						    =			toks[:max_length]
         if min_length is not None and len(A_	) < min_length and len(A_	) > 0:
           while len(A_	) < min_length:
             _lowercase						    =			toks + toks
        # toks_str = [t[1] for t in toks]
         _lowercase						    =			[t[0] for t in toks]

         # Ensure consistency
         _lowercase						    =			tokenizer.decode(A_    ,clean_up_tokenization_spaces=A_	)
         if " " not in output_txt and len(A_	) > 1:
           _lowercase						    =			(
               tokenizer.decode([toks_ids[0]]    ,clean_up_tokenization_spaces=A_	)
               + ' '
               + tokenizer.decode(toks_ids[1:]    ,clean_up_tokenization_spaces=A_	)
           )
         if with_prefix_space:
           _lowercase						    =			' ' + output_txt
         _lowercase						    =			tokenizer.encode(A_    ,add_special_tokens=A_	)
         return output_txt, output_ids







       def  __UpperCAmelCase			(     self						:						List[Any]	)    ->       List[Any]:
         _lowercase						    =			self.ta_base_tokenizer
         _lowercase						    =			tokenizer(['hi</s>', 'I went to the gym</s>', '</s>']	)
         _lowercase						    =			tokenizer(['hi', 'I went to the gym', '']	)
         self.assertListEqual(batch_with_eos_added['input_ids']    ,batch_without_eos_added['input_ids']	)







       def  __UpperCAmelCase			(     self						:						Dict	)    ->       int:
         _lowercase						    =			self.ta_base_tokenizer
         _lowercase						    =			'Unicode €.'
         _lowercase						    =			tokenizer(A_	)
         _lowercase						    =			[88, 113, 108, 102, 114, 103, 104, 35, 229, 133, 175, 49, 1]
         self.assertEqual(encoded['input_ids']    ,A_	)

         # decoding
         _lowercase						    =			tokenizer.decode(A_	)
         self.assertEqual(A_    ,'Unicode €.</s>'	)

         _lowercase						    =			tokenizer('e è é ê ë'	)
         _lowercase						    =			[104, 35, 198, 171, 35, 198, 172, 35, 198, 173, 35, 198, 174, 1]
         self.assertEqual(encoded['input_ids']    ,A_	)
         # decoding
         _lowercase						    =			tokenizer.decode(A_	)
         self.assertEqual(A_    ,'e è é ê ë</s>'	)

         # encode/decode, but with `encode` instead of `__call__`
         self.assertEqual(tokenizer.decode(tokenizer.encode('e è é ê ë'	)	)    ,'e è é ê ë</s>'	)







       def  __UpperCAmelCase			(     self						:						Any	)    ->       str:
         _lowercase						    =			self.ta_base_tokenizer
         _lowercase						    =			['A long paragraph for summarization.', 'Another paragraph for summarization.']
         # fmt: off
         _lowercase						    =			[68, 35, 111, 114, 113, 106, 35, 115, 100, 117, 100, 106, 117, 100, 115, 107, 35, 105, 114, 117, 35, 118, 120, 112, 112, 100, 117, 108, 125, 100, 119, 108, 114, 113, 49, 1, 0]
         # fmt: on
         _lowercase						    =			tokenizer(A_    ,padding=A_    ,return_tensors=A_	)
         self.assertIsInstance(A_    ,A_	)

         if FRAMEWORK != "jax":
           _lowercase						    =			list(batch.input_ids.numpy()[0]	)
         else:
           _lowercase						    =			list(batch.input_ids.tolist()[0]	)

         self.assertListEqual(A_    ,A_	)

         self.assertEqual((2, 37)    ,batch.input_ids.shape	)
         self.assertEqual((2, 37)    ,batch.attention_mask.shape	)







       def  __UpperCAmelCase			(     self						:						Dict	)    ->       str:
         _lowercase						    =			self.ta_base_tokenizer
         _lowercase						    =			['A long paragraph for summarization.', 'Another paragraph for summarization.']
         _lowercase						    =			tokenizer(A_    ,padding=A_    ,return_tensors=A_	)
         # check if input_ids are returned and no decoder_input_ids
         self.assertIn('input_ids'    ,A_	)
         self.assertIn('attention_mask'    ,A_	)
         self.assertNotIn('decoder_input_ids'    ,A_	)
         self.assertNotIn('decoder_attention_mask'    ,A_	)







       def  __UpperCAmelCase			(     self						:						Dict	)    ->       Dict:
         _lowercase						    =			self.ta_base_tokenizer
         _lowercase						    =			[
             'Summary of the text.',
             'Another summary.',
         ]
         _lowercase						    =			tokenizer(
             text_target=A_    ,max_length=32    ,padding='max_length'    ,truncation=A_    ,return_tensors=A_	)
         self.assertEqual(32    ,targets['input_ids'].shape[1]	)







       def  __UpperCAmelCase			(     self						:						Union[str, Any]	)    ->       int:
         _lowercase						    =			self.ta_base_tokenizer
         _lowercase						    =			['A long paragraph for summarization. </s>']
         _lowercase						    =			['Summary of the text. </s>']
         # fmt: off
         _lowercase						    =			[68, 35, 111, 114, 113, 106, 35, 115, 100, 117, 100, 106, 117, 100, 115, 107, 35, 105, 114, 117, 35, 118, 120, 112, 112, 100, 117, 108, 125, 100, 119, 108, 114, 113, 49, 35, 1]
         _lowercase						    =			[86, 120, 112, 112, 100, 117, 124, 35, 114, 105, 35, 119, 107, 104, 35, 119, 104, 123, 119, 49, 35, 1]
         # fmt: on

         _lowercase						    =			tokenizer(A_    ,text_target=A_	)

         self.assertEqual(A_    ,batch['input_ids'][0]	)
         self.assertEqual(A_    ,batch['labels'][0]	)







       def  __UpperCAmelCase			(     self						:						int	)    ->       Optional[int]:
         _lowercase						    =			self.get_tokenizers()
         for tokenizer in tokenizers:
           with self.subTest(F"""{tokenizer.__class__.__name__}"""	):
             self.assertNotEqual(tokenizer.model_max_length    ,42	)

        # Now let's start the test
         _lowercase						    =			self.get_tokenizers()
         for tokenizer in tokenizers:
           with self.subTest(F"""{tokenizer.__class__.__name__}"""	):
             # Isolate this from the other tests because we save additional tokens/etc
             _lowercase						    =			tempfile.mkdtemp()

             _lowercase						    =			' He is very happy, UNwant\u00E9d,running'
             _lowercase						    =			tokenizer.encode(A_    ,add_special_tokens=A_	)
             tokenizer.save_pretrained(A_	)

             _lowercase						    =			tokenizer.__class__.from_pretrained(A_	)
             _lowercase						    =			after_tokenizer.encode(A_    ,add_special_tokens=A_	)
             self.assertListEqual(A_    ,A_	)

             shutil.rmtree(A_	)

         _lowercase						    =			self.get_tokenizers(model_max_length=42	)
         for tokenizer in tokenizers:
           with self.subTest(F"""{tokenizer.__class__.__name__}"""	):
             # Isolate this from the other tests because we save additional tokens/etc
             _lowercase						    =			tempfile.mkdtemp()

             _lowercase						    =			' He is very happy, UNwant\u00E9d,running'
             tokenizer.add_tokens(['bim', 'bambam']	)
             _lowercase						    =			tokenizer.additional_special_tokens
             additional_special_tokens.append('new_additional_special_token'	)
             tokenizer.add_special_tokens({'additional_special_tokens': additional_special_tokens}	)
             _lowercase						    =			tokenizer.encode(A_    ,add_special_tokens=A_	)
             tokenizer.save_pretrained(A_	)

             _lowercase						    =			tokenizer.__class__.from_pretrained(A_	)
             _lowercase						    =			after_tokenizer.encode(A_    ,add_special_tokens=A_	)
             self.assertListEqual(A_    ,A_	)
             self.assertIn('new_additional_special_token'    ,after_tokenizer.additional_special_tokens	)
             self.assertEqual(after_tokenizer.model_max_length    ,42	)

             _lowercase						    =			tokenizer.__class__.from_pretrained(A_    ,model_max_length=43	)
             self.assertEqual(tokenizer.model_max_length    ,43	)

             shutil.rmtree(A_	)







       def  __UpperCAmelCase			(     self						:						Tuple	)    ->       Tuple:
         _lowercase						    =			[]
         if self.test_slow_tokenizer:
           tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())	)

         if self.test_rust_tokenizer:
           tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())	)

         for tokenizer_class, tokenizer_utils in tokenizer_list:
           with tempfile.TemporaryDirectory() as tmp_dir:
             tokenizer_utils.save_pretrained(A_	)

             with open(os.path.join(A_    ,'special_tokens_map.json'	)    ,encoding='utf-8'	) as json_file:
               _lowercase						    =			json.load(A_	)

             with open(os.path.join(A_    ,'tokenizer_config.json'	)    ,encoding='utf-8'	) as json_file:
               _lowercase						    =			json.load(A_	)

             _lowercase						    =			[F"""<extra_id_{i}>""" for i in range(125	)]

             _lowercase						    =			added_tokens_extra_ids + [
                 'an_additional_special_token'
             ]
             _lowercase						    =			added_tokens_extra_ids + [
                 'an_additional_special_token'
             ]

             with open(os.path.join(A_    ,'special_tokens_map.json'	)    ,'w'    ,encoding='utf-8'	) as outfile:
               json.dump(A_    ,A_	)
             with open(os.path.join(A_    ,'tokenizer_config.json'	)    ,'w'    ,encoding='utf-8'	) as outfile:
               json.dump(A_    ,A_	)

             # the following checks allow us to verify that our test works as expected, i.e. that the tokenizer takes
             # into account the new value of additional_special_tokens given in the "tokenizer_config.json" and
             # "special_tokens_map.json" files
             _lowercase						    =			tokenizer_class.from_pretrained(
                 A_    ,)
             self.assertIn(
                 'an_additional_special_token'    ,tokenizer_without_change_in_init.additional_special_tokens	)
             # self.assertIn("an_additional_special_token",tokenizer_without_change_in_init.get_vocab()) # ByT5Tokenization no vocab
             self.assertEqual(
                 ['an_additional_special_token']    ,tokenizer_without_change_in_init.convert_ids_to_tokens(
                     tokenizer_without_change_in_init.convert_tokens_to_ids(['an_additional_special_token']	)	)    ,)

             # Now we test that we can change the value of additional_special_tokens in the from_pretrained
             _lowercase						    =			added_tokens_extra_ids + [AddedToken('a_new_additional_special_token'    ,lstrip=A_	)]
             _lowercase						    =			tokenizer_class.from_pretrained(
                 A_    ,additional_special_tokens=A_    ,)

             self.assertIn('a_new_additional_special_token'    ,tokenizer.additional_special_tokens	)
             self.assertEqual(
                 ['a_new_additional_special_token']    ,tokenizer.convert_ids_to_tokens(
                     tokenizer.convert_tokens_to_ids(['a_new_additional_special_token']	)	)    ,)







       def  __UpperCAmelCase			(     self						:						str	)    ->       str:
         _lowercase						    =			[]
         if self.test_slow_tokenizer:
           tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())	)

         if self.test_rust_tokenizer:
           tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())	)

         for tokenizer_class, tokenizer_utils in tokenizer_list:
           with tempfile.TemporaryDirectory() as tmp_dir:
             tokenizer_utils.save_pretrained(A_	)

             _lowercase						    =			tokenizer_class.from_pretrained(A_	)

             self.assertTrue(tokenizer.decode([255]	) == ''	)







       def  __UpperCAmelCase			(     self						:						Union[str, Any]	)    ->       Optional[int]:
         pass







       def  __UpperCAmelCase			(     self						:						Dict	)    ->       Dict:
         pass







       def  __UpperCAmelCase			(     self						:						Any	)    ->       str:
         pass







       def  __UpperCAmelCase			(     self						:						Tuple	)    ->       str:
         pass







       def  __UpperCAmelCase			(     self						:						Optional[Any]	)    ->       int:
         _lowercase						    =			self.get_tokenizers(fast=A_    ,do_lower_case=A_	)
         for tokenizer in tokenizers:
           with self.subTest(F"""{tokenizer.__class__.__name__}"""	):
             _lowercase						    =			['t', 'h', 'i', 's', ' ', 'i', 's', ' ', 'a', ' ', 't', 'e', 'x', 't', '</s>']
             _lowercase						    =			tokenizer.convert_tokens_to_string(A_	)

             self.assertIsInstance(A_    ,A_	)







       def  __UpperCAmelCase			(     self						:						Dict	)    ->       Union[str, Any]:
         _lowercase						    =			self.get_tokenizers()
         for tokenizer in tokenizers:
           with self.subTest(F"""{tokenizer.__class__.__name__}"""	):
             _lowercase						    =			[
                 'bos_token',
                 'eos_token',
                 'unk_token',
                 'sep_token',
                 'pad_token',
                 'cls_token',
                 'mask_token',
             ]

             _lowercase						    =			0
             _lowercase						    =			tokenizer.convert_ids_to_tokens(
                 A_    ,skip_special_tokens=A_	)

             for attr in attributes_list:
               setattr(A_    ,attr + '_id'    ,A_	)
               self.assertEqual(getattr(A_    ,A_	)    ,A_	)
               self.assertEqual(getattr(A_    ,attr + '_id'	)    ,A_	)

               setattr(A_    ,attr + '_id'    ,A_	)
               self.assertEqual(getattr(A_    ,A_	)    ,A_	)
               self.assertEqual(getattr(A_    ,attr + '_id'	)    ,A_	)

             setattr(A_    ,'additional_special_tokens_ids'    ,[]	)
             self.assertListEqual(getattr(A_    ,'additional_special_tokens'	)    ,[]	)
             self.assertListEqual(getattr(A_    ,'additional_special_tokens_ids'	)    ,[]	)

             setattr(A_    ,'additional_special_tokens_ids'    ,[token_id_to_test_setters]	)
             self.assertListEqual(getattr(A_    ,'additional_special_tokens'	)    ,[token_to_test_setters]	)
             self.assertListEqual(getattr(A_    ,'additional_special_tokens_ids'	)    ,[token_id_to_test_setters]	)



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


__snake_case   							=						get_tests_dir('''fixtures''')



class 	__lowerCamelCase			(unittest.TestCase						):







       def 		snake_case_	(			self: int					):






            '''simple docstring'''




            __UpperCamelCase =			mock.Mock()
            __UpperCamelCase =			500
            __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 		snake_case_	(			self: Dict					):






            '''simple docstring'''




            __UpperCamelCase =			WavaVecaFeatureExtractor.from_pretrained(
                'https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'					)



@is_staging_test
class 	__lowerCamelCase			(unittest.TestCase						):







       @classmethod
       def 		snake_case_	(			cls: Tuple					):






            '''simple docstring'''




            __UpperCamelCase =			TOKEN
            HfFolder.save_token(A_					)







       @classmethod
       def 		snake_case_	(			cls: Tuple					):






            '''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 		snake_case_	(			self: Tuple					):






            '''simple docstring'''




            __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 		snake_case_	(			self: List[str]					):






            '''simple docstring'''




            __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 		snake_case_	(			self: int					):






            '''simple docstring'''




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


