




import shutil
import tempfile
import unittest

from transformers import (
    SPIECE_UNDERLINE,
    AddedToken,
    BatchEncoding,
    NllbTokenizer,
    NllbTokenizerFast,
    is_torch_available,
)
from transformers.testing_utils import (
    get_tests_dir,
    nested_simplify,
    require_sentencepiece,
    require_tokenizers,
    require_torch,
)

from ...test_tokenization_common import TokenizerTesterMixin


_lowerCamelCase     :		List[str]								=       get_tests_dir("""fixtures/test_sentencepiece.model""")


if is_torch_available():
  from transformers.models.mam_aaa.modeling_mam_aaa import shift_tokens_right

_lowerCamelCase     :		Tuple								=       256047
_lowerCamelCase     :		Any								=       256145


@require_sentencepiece
@require_tokenizers
class     UpperCamelCase_   (			UpperCAmelCase__		, unittest.TestCase					):

    '''simple docstring'''






    UpperCAmelCase__							  =				NllbTokenizer
    UpperCAmelCase__							  =				NllbTokenizerFast
    UpperCAmelCase__							  =				True
    UpperCAmelCase__							  =				True
    UpperCAmelCase__							  =				{}


    def 	SCREAMING_SNAKE_CASE							(      self				:     int)						->int:





           '''simple docstring'''

           super().setUp()

           # We have a SentencePiece fixture for testing
           A__							=   NllbTokenizer(UpperCAmelCase__ ,			keep_accents=UpperCAmelCase__)
           tokenizer.save_pretrained(self.tmpdirname)


    def 	SCREAMING_SNAKE_CASE							(      self				:     Optional[Any])						->List[Any]:





           '''simple docstring'''

           A__							=   NllbTokenizer(UpperCAmelCase__ ,			keep_accents=UpperCAmelCase__)

           A__							=   tokenizer.tokenize('''This is a test''')
           self.assertListEqual(UpperCAmelCase__ ,			['''▁This''', '''▁is''', '''▁a''', '''▁t''', '''est'''])

           self.assertListEqual(
               tokenizer.convert_tokens_to_ids(UpperCAmelCase__) ,			[value + tokenizer.fairseq_offset for value in [285, 46, 10, 170, 382]] ,			)

           A__							=   tokenizer.tokenize('''I was born in 92000, and this is falsé.''')
           self.assertListEqual(
               UpperCAmelCase__ ,			[
                   SPIECE_UNDERLINE + '''I''',
                   SPIECE_UNDERLINE + '''was''',
                   SPIECE_UNDERLINE + '''b''',
                   '''or''',
                   '''n''',
                   SPIECE_UNDERLINE + '''in''',
                   SPIECE_UNDERLINE + '''''',
                   '''9''',
                   '''2''',
                   '''0''',
                   '''0''',
                   '''0''',
                   ''',''',
                   SPIECE_UNDERLINE + '''and''',
                   SPIECE_UNDERLINE + '''this''',
                   SPIECE_UNDERLINE + '''is''',
                   SPIECE_UNDERLINE + '''f''',
                   '''al''',
                   '''s''',
                   '''é''',
                   '''.''',
               ] ,			)
           A__							=   tokenizer.convert_tokens_to_ids(UpperCAmelCase__)
           self.assertListEqual(
               UpperCAmelCase__ ,			[
                   value + tokenizer.fairseq_offset
                   for value in [8, 21, 84, 55, 24, 19, 7, 2, 602, 347, 347, 347, 3, 12, 66, 46, 72, 80, 6, 2, 4]
               ] ,			)

           A__							=   tokenizer.convert_ids_to_tokens(UpperCAmelCase__)
           self.assertListEqual(
               UpperCAmelCase__ ,			[
                   SPIECE_UNDERLINE + '''I''',
                   SPIECE_UNDERLINE + '''was''',
                   SPIECE_UNDERLINE + '''b''',
                   '''or''',
                   '''n''',
                   SPIECE_UNDERLINE + '''in''',
                   SPIECE_UNDERLINE + '''''',
                   '''<unk>''',
                   '''2''',
                   '''0''',
                   '''0''',
                   '''0''',
                   ''',''',
                   SPIECE_UNDERLINE + '''and''',
                   SPIECE_UNDERLINE + '''this''',
                   SPIECE_UNDERLINE + '''is''',
                   SPIECE_UNDERLINE + '''f''',
                   '''al''',
                   '''s''',
                   '''<unk>''',
                   '''.''',
               ] ,			)


    def 	SCREAMING_SNAKE_CASE							(      self				:     Optional[Any])						->str:





           '''simple docstring'''

           A__							=   (self.rust_tokenizer_class, '''hf-internal-testing/tiny-random-nllb''', {})
           for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
                  with self.subTest(f"""{tokenizer.__class__.__name__} ({pretrained_name})"""):
                         A__							=   self.rust_tokenizer_class.from_pretrained(UpperCAmelCase__ ,			**UpperCAmelCase__)
                         A__							=   self.tokenizer_class.from_pretrained(UpperCAmelCase__ ,			**UpperCAmelCase__)

                         A__							=   tempfile.mkdtemp()

                         A__							=   tokenizer_r.save_pretrained(UpperCAmelCase__)
                         A__							=   tokenizer_p.save_pretrained(UpperCAmelCase__)

                         # Checks it save with the same files + the tokenizer.json file for the fast one
                         self.assertTrue(any('''tokenizer.json''' in f for f in tokenizer_r_files))
                         A__							=   tuple(f for f in tokenizer_r_files if '''tokenizer.json''' not in f)
                         self.assertSequenceEqual(UpperCAmelCase__ ,			UpperCAmelCase__)

                         # Checks everything loads correctly in the same way
                         A__							=   tokenizer_r.from_pretrained(UpperCAmelCase__)
                         A__							=   tokenizer_p.from_pretrained(UpperCAmelCase__)

                         # Check special tokens are set accordingly on Rust and Python
                         for key in tokenizer_pp.special_tokens_map:
                                self.assertTrue(hasattr(UpperCAmelCase__ ,			UpperCAmelCase__))

                         shutil.rmtree(UpperCAmelCase__)

                         # Save tokenizer rust, legacy_format=True
                         A__							=   tempfile.mkdtemp()

                         A__							=   tokenizer_r.save_pretrained(UpperCAmelCase__ ,			legacy_format=UpperCAmelCase__)
                         A__							=   tokenizer_p.save_pretrained(UpperCAmelCase__)

                         # Checks it save with the same files
                         self.assertSequenceEqual(UpperCAmelCase__ ,			UpperCAmelCase__)

                         # Checks everything loads correctly in the same way
                         A__							=   tokenizer_r.from_pretrained(UpperCAmelCase__)
                         A__							=   tokenizer_p.from_pretrained(UpperCAmelCase__)

                         # Check special tokens are set accordingly on Rust and Python
                         for key in tokenizer_pp.special_tokens_map:
                                self.assertTrue(hasattr(UpperCAmelCase__ ,			UpperCAmelCase__))

                         shutil.rmtree(UpperCAmelCase__)

                         # Save tokenizer rust, legacy_format=False
                         A__							=   tempfile.mkdtemp()

                         A__							=   tokenizer_r.save_pretrained(UpperCAmelCase__ ,			legacy_format=UpperCAmelCase__)
                         A__							=   tokenizer_p.save_pretrained(UpperCAmelCase__)

                         # Checks it saved the tokenizer.json file
                         self.assertTrue(any('''tokenizer.json''' in f for f in tokenizer_r_files))

                         # Checks everything loads correctly in the same way
                         A__							=   tokenizer_r.from_pretrained(UpperCAmelCase__)
                         A__							=   tokenizer_p.from_pretrained(UpperCAmelCase__)

                         # Check special tokens are set accordingly on Rust and Python
                         for key in tokenizer_pp.special_tokens_map:
                                self.assertTrue(hasattr(UpperCAmelCase__ ,			UpperCAmelCase__))

                         shutil.rmtree(UpperCAmelCase__)


    @require_torch
    def 	SCREAMING_SNAKE_CASE							(      self				:     List[str])						->str:





           '''simple docstring'''

           if not self.test_seqaseq:
                  return

           A__							=   self.get_tokenizers()
           for tokenizer in tokenizers:
                  with self.subTest(f"""{tokenizer.__class__.__name__}"""):
                         # Longer text that will definitely require truncation.
                         A__							=   [
                             ''' UN Chief Says There Is No Military Solution in Syria''',
                             ''' Secretary-General Ban Ki-moon says his response to Russia\'s stepped up military support for'''
                             ''' Syria is that \'there is no military solution\' to the nearly five-year conflict and more weapons'''
                             ''' will only worsen the violence and misery for millions of people.''',
                         ]
                         A__							=   [
                             '''Şeful ONU declară că nu există o soluţie militară în Siria''',
                             '''Secretarul General Ban Ki-moon declară că răspunsul său la intensificarea sprijinului militar al'''
                             ''' Rusiei pentru Siria este că "nu există o soluţie militară" la conflictul de aproape cinci ani şi'''
                             ''' că noi arme nu vor face decât să înrăutăţească violenţele şi mizeria pentru milioane de oameni.''',
                         ]
                         try:
                                A__							=   tokenizer.prepare_seqaseq_batch(
                                    src_texts=UpperCAmelCase__ ,			tgt_texts=UpperCAmelCase__ ,			max_length=3 ,			max_target_length=10 ,			return_tensors='''pt''' ,			src_lang='''eng_Latn''' ,			tgt_lang='''ron_Latn''' ,			)
                         except NotImplementedError:
                                return
                         self.assertEqual(batch.input_ids.shape[1] ,			3)
                         self.assertEqual(batch.labels.shape[1] ,			10)
                         # max_target_length will default to max_length if not specified
                         A__							=   tokenizer.prepare_seqaseq_batch(
                             UpperCAmelCase__ ,			tgt_texts=UpperCAmelCase__ ,			max_length=3 ,			return_tensors='''pt''')
                         self.assertEqual(batch.input_ids.shape[1] ,			3)
                         self.assertEqual(batch.labels.shape[1] ,			3)

                         A__							=   tokenizer.prepare_seqaseq_batch(
                             src_texts=UpperCAmelCase__ ,			max_length=3 ,			max_target_length=10 ,			return_tensors='''pt''')
                         self.assertEqual(batch_encoder_only.input_ids.shape[1] ,			3)
                         self.assertEqual(batch_encoder_only.attention_mask.shape[1] ,			3)
                         self.assertNotIn('''decoder_input_ids''' ,			UpperCAmelCase__)


    @unittest.skip('''Unfortunately way too slow to build a BPE with SentencePiece.''')
    def 	SCREAMING_SNAKE_CASE							(      self				:     List[Any])						->Union[str, Any]:





           '''simple docstring'''

           pass





    def 	SCREAMING_SNAKE_CASE							(      self				:     Tuple)						->Any:





           '''simple docstring'''

           for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
                  with self.subTest(f"""{tokenizer.__class__.__name__} ({pretrained_name})"""):
                         A__							=   [AddedToken('''<special>''' ,			lstrip=UpperCAmelCase__)]

                         A__							=   self.rust_tokenizer_class.from_pretrained(
                             UpperCAmelCase__ ,			additional_special_tokens=UpperCAmelCase__ ,			**UpperCAmelCase__)
                         A__							=   tokenizer_r.encode('''Hey this is a <special> token''')

                         A__							=   tokenizer_r.encode('''<special>''' ,			add_special_tokens=UpperCAmelCase__)[0]

                         self.assertTrue(special_token_id in r_output)

                         if self.test_slow_tokenizer:
                                A__							=   self.rust_tokenizer_class.from_pretrained(
                                    UpperCAmelCase__ ,			additional_special_tokens=UpperCAmelCase__ ,			**UpperCAmelCase__ ,			)
                                A__							=   self.tokenizer_class.from_pretrained(
                                    UpperCAmelCase__ ,			additional_special_tokens=UpperCAmelCase__ ,			**UpperCAmelCase__)

                                A__							=   tokenizer_p.encode('''Hey this is a <special> token''')

                                A__							=   tokenizer_cr.encode('''Hey this is a <special> token''')

                                self.assertEqual(UpperCAmelCase__ ,			UpperCAmelCase__)
                                self.assertEqual(UpperCAmelCase__ ,			UpperCAmelCase__)
                                self.assertTrue(special_token_id in p_output)
                                self.assertTrue(special_token_id in cr_output)


@require_torch
@require_sentencepiece
@require_tokenizers
class     UpperCamelCase_   (			unittest.TestCase					):

    '''simple docstring'''






    UpperCAmelCase__							  =				'''facebook/nllb-200-distilled-600M'''
    UpperCAmelCase__							  =				[
        ''' UN Chief Says There Is No Military Solution in Syria''',
        ''' Secretary-General Ban Ki-moon says his response to Russia\'s stepped up military support for Syria is that "there is no military solution" to the nearly five-year conflict and more weapons will only worsen the violence and misery for millions of people.''',
    ]
    UpperCAmelCase__							  =				[
        '''Şeful ONU declară că nu există o soluţie militară în Siria''',
        '''Secretarul General Ban Ki-moon declară că răspunsul său la intensificarea sprijinului militar al Rusiei'''
        ''' pentru Siria este că "nu există o soluţie militară" la conflictul de aproape cinci ani şi că noi arme nu vor'''
        ''' face decât să înrăutăţească violenţele şi mizeria pentru milioane de oameni.''',
    ]
    UpperCAmelCase__							  =				[
        25_6047,
        1_6297,
        13_4408,
        8165,
        24_8066,
        1_4734,
        950,
        1135,
        10_5721,
        3573,
        83,
        2_7352,
        108,
        4_9486,
        2,
    ]


    @classmethod
    def 	SCREAMING_SNAKE_CASE							(      cls				:     Optional[int])						->List[str]:





           '''simple docstring'''

           A__							=   NllbTokenizer.from_pretrained(
               cls.checkpoint_name ,			src_lang='''eng_Latn''' ,			tgt_lang='''ron_Latn''')
           A__							=   1
           return cls


    def 	SCREAMING_SNAKE_CASE							(      self				:     Optional[int])						->Any:





           '''simple docstring'''

           self.assertEqual(self.tokenizer.fairseq_tokens_to_ids['''ace_Arab'''] ,			256_001)
           self.assertEqual(self.tokenizer.fairseq_tokens_to_ids['''ace_Latn'''] ,			256_002)
           self.assertEqual(self.tokenizer.fairseq_tokens_to_ids['''fra_Latn'''] ,			256_057)


    def 	SCREAMING_SNAKE_CASE							(      self				:     str)						->Tuple:





           '''simple docstring'''

           A__							=   self.tokenizer.batch_encode_plus(self.src_text).input_ids[0]
           self.assertListEqual(self.expected_src_tokens ,			UpperCAmelCase__)


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





           '''simple docstring'''

           self.assertIn(UpperCAmelCase__ ,			self.tokenizer.all_special_ids)
           # fmt: off
           A__							=   [RO_CODE, 4_254, 98_068, 112_923, 39_072, 3_909, 713, 102_767, 26, 17_314, 35_642, 14_683, 33_118, 2_022, 66_987, 2, 256_047]
           # fmt: on

           A__							=   self.tokenizer.decode(UpperCAmelCase__ ,			skip_special_tokens=UpperCAmelCase__)
           A__							=   self.tokenizer.decode(generated_ids[1:] ,			skip_special_tokens=UpperCAmelCase__)
           self.assertEqual(UpperCAmelCase__ ,			UpperCAmelCase__)
           self.assertNotIn(self.tokenizer.eos_token ,			UpperCAmelCase__)


    def 	SCREAMING_SNAKE_CASE							(      self				:     Dict)						->str:





           '''simple docstring'''

           A__							=   ['''this is gunna be a long sentence ''' * 20]
           assert isinstance(src_text[0] ,			UpperCAmelCase__)
           A__							=   10
           A__							=   self.tokenizer(UpperCAmelCase__ ,			max_length=UpperCAmelCase__ ,			truncation=UpperCAmelCase__).input_ids[0]
           self.assertEqual(ids[-1] ,			2)
           self.assertEqual(ids[0] ,			UpperCAmelCase__)
           self.assertEqual(len(UpperCAmelCase__) ,			UpperCAmelCase__)


    def 	SCREAMING_SNAKE_CASE							(      self				:     List[str])						->int:





           '''simple docstring'''

           self.assertListEqual(self.tokenizer.convert_tokens_to_ids(['''<mask>''', '''ar_AR''']) ,			[256_203, 3])


    def 	SCREAMING_SNAKE_CASE							(      self				:     int)						->Union[str, Any]:





           '''simple docstring'''

           A__							=   tempfile.mkdtemp()
           A__							=   self.tokenizer.fairseq_tokens_to_ids
           self.tokenizer.save_pretrained(UpperCAmelCase__)
           A__							=   NllbTokenizer.from_pretrained(UpperCAmelCase__)
           self.assertDictEqual(new_tok.fairseq_tokens_to_ids ,			UpperCAmelCase__)


    @require_torch
    def 	SCREAMING_SNAKE_CASE							(      self				:     Dict)						->Tuple:





           '''simple docstring'''

           A__							=   self.tokenizer(
               self.src_text ,			text_target=self.tgt_text ,			padding=UpperCAmelCase__ ,			truncation=UpperCAmelCase__ ,			max_length=len(self.expected_src_tokens) ,			return_tensors='''pt''' ,			)
           A__							=   shift_tokens_right(
               batch['''labels'''] ,			self.tokenizer.pad_token_id ,			self.tokenizer.lang_code_to_id['''ron_Latn'''])

           self.assertIsInstance(UpperCAmelCase__ ,			UpperCAmelCase__)

           self.assertEqual((2, 15) ,			batch.input_ids.shape)
           self.assertEqual((2, 15) ,			batch.attention_mask.shape)
           A__							=   batch.input_ids.tolist()[0]
           self.assertListEqual(self.expected_src_tokens ,			UpperCAmelCase__)
           self.assertEqual(UpperCAmelCase__ ,			batch.decoder_input_ids[0, 0])  # EOS
           # Test that special tokens are reset
           self.assertEqual(self.tokenizer.prefix_tokens ,			[EN_CODE])
           self.assertEqual(self.tokenizer.suffix_tokens ,			[self.tokenizer.eos_token_id])


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





           '''simple docstring'''

           A__							=   self.tokenizer(self.src_text ,			padding=UpperCAmelCase__ ,			truncation=UpperCAmelCase__ ,			max_length=3 ,			return_tensors='''pt''')
           A__							=   self.tokenizer(
               text_target=self.tgt_text ,			padding=UpperCAmelCase__ ,			truncation=UpperCAmelCase__ ,			max_length=10 ,			return_tensors='''pt''')
           A__							=   targets['''input_ids''']
           A__							=   shift_tokens_right(
               UpperCAmelCase__ ,			self.tokenizer.pad_token_id ,			decoder_start_token_id=self.tokenizer.lang_code_to_id[self.tokenizer.tgt_lang] ,			)

           self.assertEqual(batch.input_ids.shape[1] ,			3)
           self.assertEqual(batch.decoder_input_ids.shape[1] ,			10)


    @require_torch
    def 	SCREAMING_SNAKE_CASE							(      self				:     List[Any])						->Optional[int]:





           '''simple docstring'''

           A__							=   self.tokenizer._build_translation_inputs(
               '''A test''' ,			return_tensors='''pt''' ,			src_lang='''eng_Latn''' ,			tgt_lang='''fra_Latn''')

           self.assertEqual(
               nested_simplify(UpperCAmelCase__) ,			{
                   # A, test, EOS, en_XX
                   '''input_ids''': [[256_047, 70, 7_356, 2]],
                   '''attention_mask''': [[1, 1, 1, 1]],
                   # ar_AR
                   '''forced_bos_token_id''': 256_057,
               } ,			)





    @require_torch
    def 	SCREAMING_SNAKE_CASE							(      self				:     int)						->Tuple:





           '''simple docstring'''

           A__							=   True
           A__							=   self.tokenizer(
               '''UN Chief says there is no military solution in Syria''' ,			src_lang='''eng_Latn''' ,			tgt_lang='''fra_Latn''')
           self.assertEqual(
               inputs.input_ids ,			[16_297, 134_408, 25_653, 6_370, 248, 254, 103_929, 94_995, 108, 49_486, 2, 256_047])

           A__							=   False
           A__							=   self.tokenizer(
               '''UN Chief says there is no military solution in Syria''' ,			src_lang='''eng_Latn''' ,			tgt_lang='''fra_Latn''')
           self.assertEqual(
               inputs.input_ids ,			[256_047, 16_297, 134_408, 25_653, 6_370, 248, 254, 103_929, 94_995, 108, 49_486, 2])








class     UpperCamelCase_   :

    '''simple docstring'''






    def __init__(      self				:     List[str] ,			UpperCAmelCase__				:     int)						->Union[str, Any]:





           '''simple docstring'''

           A__							=   n
           A__							=   [None] * self.n
           A__							=   0  # index of the first element
           A__							=   0
           A__							=   0


    def __len__(      self				:     List[Any])						->int:





           '''simple docstring'''

           return self.size


    def 	SCREAMING_SNAKE_CASE							(      self				:     List[Any])						->bool:





           '''simple docstring'''

           return self.size == 0


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





           '''simple docstring'''

           return False if self.is_empty() else self.array[self.front]


    def 	SCREAMING_SNAKE_CASE							(      self				:     List[str] ,			UpperCAmelCase__				:     str)						->Any:





           '''simple docstring'''

           if self.size >= self.n:
                  raise Exception('''QUEUE IS FULL''')

           A__							=   data
           A__							=   (self.rear + 1) % self.n
           self.size += 1
           return self





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





           '''simple docstring'''

           if self.size == 0:
                  raise Exception('''UNDERFLOW''')

           A__							=   self.array[self.front]
           A__							=   None
           A__							=   (self.front + 1) % self.n
           self.size -= 1
           return temp


