



import json
import os
import unittest

from transformers import BatchEncoding, LEDTokenizer, LEDTokenizerFast
from transformers.models.led.tokenization_led import VOCAB_FILES_NAMES
from transformers.testing_utils import require_tokenizers, require_torch
from transformers.utils import cached_property

from ...test_tokenization_common import TokenizerTesterMixin



@require_tokenizers
class 							lowerCAmelCase_						(    a__			,       unittest.TestCase				):
       UpperCAmelCase__							:   Dict  	=    LEDTokenizer
       UpperCAmelCase__							:   Union[str, Any]  	=    LEDTokenizerFast
       UpperCAmelCase__							:   int  	=    True
       def    snake_case_			(     self			)	->						Tuple:
          super().setUp()
          UpperCamelCase       :    str		      =      [
              'l',
              'o',
              'w',
              'e',
              'r',
              's',
              't',
              'i',
              'd',
              'n',
              '\u0120',
              '\u0120l',
              '\u0120n',
              '\u0120lo',
              '\u0120low',
              'er',
              '\u0120lowest',
              '\u0120newer',
              '\u0120wider',
              '<unk>',
          ]
          UpperCamelCase       :    str		      =      dict(zip(SCREAMING_SNAKE_CASE_,     range(len(SCREAMING_SNAKE_CASE_			)			)			)			)
          UpperCamelCase       :    int		      =      ['#version: 0.2', '\u0120 l', '\u0120l o', '\u0120lo w', 'e r', '']
          UpperCamelCase       :    List[str]		      =      {'unk_token': '<unk>'}

          UpperCamelCase       :    Tuple		      =      os.path.join(self.tmpdirname,     VOCAB_FILES_NAMES['vocab_file']			)
          UpperCamelCase       :    Any		      =      os.path.join(self.tmpdirname,     VOCAB_FILES_NAMES['merges_file']			)
          with open(self.vocab_file,     'w',     encoding='utf-8'			) as fp:
             fp.write(json.dumps(SCREAMING_SNAKE_CASE_			) + '\n'			)
          with open(self.merges_file,     'w',     encoding='utf-8'			) as fp:
             fp.write('\n'.join(SCREAMING_SNAKE_CASE_			)			)
       def    snake_case_			(     self,     **SCREAMING_SNAKE_CASE_			)	->						Union[str, Any]:
          kwargs.update(self.special_tokens_map			)
          return self.tokenizer_class.from_pretrained(self.tmpdirname,     **SCREAMING_SNAKE_CASE_			)
       def    snake_case_			(     self,     **SCREAMING_SNAKE_CASE_			)	->						Dict:
          kwargs.update(self.special_tokens_map			)
          return self.rust_tokenizer_class.from_pretrained(self.tmpdirname,     **SCREAMING_SNAKE_CASE_			)
       def    snake_case_			(     self,     SCREAMING_SNAKE_CASE_			)	->						Tuple:
          return "lower newer", "lower newer"
       @cached_property
       def    snake_case_			(     self			)	->						Any:
          return LEDTokenizer.from_pretrained('allenai/led-base-16384'			)
       @cached_property
       def    snake_case_			(     self			)	->						Dict:
          return LEDTokenizerFast.from_pretrained('allenai/led-base-16384'			)
       @require_torch
       def    snake_case_			(     self			)	->						List[Any]:
          UpperCamelCase       :    str		      =      ['A long paragraph for summarization.', 'Another paragraph for summarization.']
          UpperCamelCase       :    Dict		      =      [0, 250, 251, 1_7818, 13, 3_9186, 1938, 4, 2]

          for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
             UpperCamelCase       :    Dict		      =      tokenizer(SCREAMING_SNAKE_CASE_,     max_length=len(SCREAMING_SNAKE_CASE_			),     padding=SCREAMING_SNAKE_CASE_,     return_tensors='pt'			)
             self.assertIsInstance(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			)

             self.assertEqual((2, 9),     batch.input_ids.shape			)
             self.assertEqual((2, 9),     batch.attention_mask.shape			)
             UpperCamelCase       :    Optional[int]		      =      batch.input_ids.tolist()[0]
             self.assertListEqual(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			)
       @require_torch
       def    snake_case_			(     self			)	->						Optional[int]:
          UpperCamelCase       :    List[str]		      =      ['A long paragraph for summarization.', 'Another paragraph for summarization.']
          for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
             UpperCamelCase       :    Dict		      =      tokenizer(SCREAMING_SNAKE_CASE_,     padding=SCREAMING_SNAKE_CASE_,     return_tensors='pt'			)
             self.assertIn('input_ids',     SCREAMING_SNAKE_CASE_			)
             self.assertIn('attention_mask',     SCREAMING_SNAKE_CASE_			)
             self.assertNotIn('labels',     SCREAMING_SNAKE_CASE_			)
             self.assertNotIn('decoder_attention_mask',     SCREAMING_SNAKE_CASE_			)
       @require_torch
       def    snake_case_			(     self			)	->						int:
          UpperCamelCase       :    List[str]		      =      [
              'Summary of the text.',
              'Another summary.',
          ]
          for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
             UpperCamelCase       :    str		      =      tokenizer(text_target=SCREAMING_SNAKE_CASE_,     max_length=32,     padding='max_length',     return_tensors='pt'			)
             self.assertEqual(32,     targets['input_ids'].shape[1]			)
       @require_torch
       def    snake_case_			(     self			)	->						Any:
          for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
             UpperCamelCase       :    int		      =      tokenizer(
                 ['I am a small frog' * 1024, 'I am a small frog'],     padding=SCREAMING_SNAKE_CASE_,     truncation=SCREAMING_SNAKE_CASE_,     return_tensors='pt'			)
             self.assertIsInstance(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			)
             self.assertEqual(batch.input_ids.shape,     (2, 5122)			)
       @require_torch
       def    snake_case_			(     self			)	->						Dict:
          UpperCamelCase       :    List[str]		      =      ['A long paragraph for summarization.']
          UpperCamelCase       :    Tuple		      =      [
              'Summary of the text.',
          ]
          for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
             UpperCamelCase       :    Union[str, Any]		      =      tokenizer(SCREAMING_SNAKE_CASE_,     return_tensors='pt'			)
             UpperCamelCase       :    str		      =      tokenizer(text_target=SCREAMING_SNAKE_CASE_,     return_tensors='pt'			)
             UpperCamelCase       :    List[Any]		      =      inputs['input_ids']
             UpperCamelCase       :    str		      =      targets['input_ids']
             self.assertTrue((input_ids[:, 0] == tokenizer.bos_token_id).all().item()			)
             self.assertTrue((labels[:, 0] == tokenizer.bos_token_id).all().item()			)
             self.assertTrue((input_ids[:, -1] == tokenizer.eos_token_id).all().item()			)
             self.assertTrue((labels[:, -1] == tokenizer.eos_token_id).all().item()			)
       @require_torch
       def    snake_case_			(     self			)	->						Union[str, Any]:
          for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
             UpperCamelCase       :    Dict		      =      ['Summary of the text.', 'Another summary.']
             UpperCamelCase       :    int		      =      [[0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, -1, -1]]

             UpperCamelCase       :    Any		      =      tokenizer(SCREAMING_SNAKE_CASE_,     padding=SCREAMING_SNAKE_CASE_			)
             UpperCamelCase       :    Tuple		      =      [[0] * len(SCREAMING_SNAKE_CASE_			) for x in encoded_output['input_ids']]
             UpperCamelCase       :    Dict		      =      tokenizer.pad(SCREAMING_SNAKE_CASE_			)
             self.assertSequenceEqual(outputs['global_attention_mask'],     SCREAMING_SNAKE_CASE_			)
       def    snake_case_			(     self			)	->						Optional[int]:
          pass






       def    snake_case_			(     self			)	->						Dict:
          for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
             with self.subTest(F"""{tokenizer.__class__.__name__} ({pretrained_name})"""			):
                UpperCamelCase       :    Optional[int]		      =      self.rust_tokenizer_class.from_pretrained(SCREAMING_SNAKE_CASE_,     **SCREAMING_SNAKE_CASE_			)
                UpperCamelCase       :    str		      =      self.tokenizer_class.from_pretrained(SCREAMING_SNAKE_CASE_,     **SCREAMING_SNAKE_CASE_			)
                UpperCamelCase       :    Dict		      =      'A, <mask> AllenNLP sentence.'
                UpperCamelCase       :    int		      =      tokenizer_r.encode_plus(SCREAMING_SNAKE_CASE_,     add_special_tokens=SCREAMING_SNAKE_CASE_,     return_token_type_ids=SCREAMING_SNAKE_CASE_			)
                UpperCamelCase       :    Tuple		      =      tokenizer_p.encode_plus(SCREAMING_SNAKE_CASE_,     add_special_tokens=SCREAMING_SNAKE_CASE_,     return_token_type_ids=SCREAMING_SNAKE_CASE_			)
                self.assertEqual(sum(tokens_r['token_type_ids']			),     sum(tokens_p['token_type_ids']			)			)
                self.assertEqual(
                    sum(tokens_r['attention_mask']			) / len(tokens_r['attention_mask']			),     sum(tokens_p['attention_mask']			) / len(tokens_p['attention_mask']			),     )

                UpperCamelCase       :    Union[str, Any]		      =      tokenizer_r.convert_ids_to_tokens(tokens_r['input_ids']			)
                UpperCamelCase       :    Optional[Any]		      =      tokenizer_p.convert_ids_to_tokens(tokens_p['input_ids']			)
                self.assertSequenceEqual(tokens_p['input_ids'],     [0, 250, 6, 5_0264, 3823, 487, 2_1992, 3645, 4, 2]			)
                self.assertSequenceEqual(tokens_r['input_ids'],     [0, 250, 6, 5_0264, 3823, 487, 2_1992, 3645, 4, 2]			)

                self.assertSequenceEqual(
                    SCREAMING_SNAKE_CASE_,     ['<s>', 'A', ',', '<mask>', 'ĠAllen', 'N', 'LP', 'Ġsentence', '.', '</s>']			)
                self.assertSequenceEqual(
                    SCREAMING_SNAKE_CASE_,     ['<s>', 'A', ',', '<mask>', 'ĠAllen', 'N', 'LP', 'Ġsentence', '.', '</s>']			)











import os
import shutil
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
from datasets import Dataset

from transformers.models.realm.configuration_realm import RealmConfig
from transformers.models.realm.retrieval_realm import _REALM_BLOCK_RECORDS_FILENAME, RealmRetriever
from transformers.models.realm.tokenization_realm import VOCAB_FILES_NAMES, RealmTokenizer


class    __lowercase   (						lowercase_	):







     '''simple docstring'''




     def 		lowerCAmelCase_  (					self							:      int					):


         """simple docstring"""



         __A       =						tempfile.mkdtemp()
         __A       =						5

         # Realm tok
         __A       =						[
             """[UNK]""",
             """[CLS]""",
             """[SEP]""",
             """[PAD]""",
             """[MASK]""",
             """test""",
             """question""",
             """this""",
             """is""",
             """the""",
             """first""",
             """second""",
             """third""",
             """fourth""",
             """fifth""",
             """record""",
             """want""",
             """##want""",
             """##ed""",
             """wa""",
             """un""",
             """runn""",
             """##ing""",
             """,""",
             """low""",
             """lowest""",
         ]
         __A       =						os.path.join(self.tmpdirname			,						"""realm_tokenizer"""					)
         os.makedirs(UpperCamelCase_			,						exist_ok=UpperCamelCase_					)
         __A       =						os.path.join(UpperCamelCase_			,						VOCAB_FILES_NAMES["""vocab_file"""]					)
         with open(self.vocab_file			,						"""w"""			,						encoding="""utf-8"""					) as vocab_writer:
             vocab_writer.write("""""".join([x + """\n""" for x in vocab_tokens]					)					)

         __A       =						os.path.join(self.tmpdirname			,						"""realm_block_records"""					)
         os.makedirs(UpperCamelCase_			,						exist_ok=UpperCamelCase_					)




     def 		lowerCAmelCase_  (					self							:      List[Any]					):


         """simple docstring"""



         return RealmTokenizer.from_pretrained(os.path.join(self.tmpdirname			,						"""realm_tokenizer"""					)					)




     def 		lowerCAmelCase_  (					self							:      Union[str, Any]					):


         """simple docstring"""



         shutil.rmtree(self.tmpdirname					)




     def 		lowerCAmelCase_  (					self							:      str					):


         """simple docstring"""



         __A       =						RealmConfig(num_block_records=self.num_block_records					)
         return config




     def 		lowerCAmelCase_  (					self							:      Dict					):


         """simple docstring"""



         __A       =						Dataset.from_dict(
             {
                 """id""": ["""0""", """1"""],
                 """question""": ["""foo""", """bar"""],
                 """answers""": [["""Foo""", """Bar"""], ["""Bar"""]],
             }					)
         return dataset




     def 		lowerCAmelCase_  (					self							:      str					):


         """simple docstring"""



         __A       =						np.array(
             [
                 B"""This is the first record""",
                 B"""This is the second record""",
                 B"""This is the third record""",
                 B"""This is the fourth record""",
                 B"""This is the fifth record""",
                 B"""This is a longer longer longer record""",
             ]			,						dtype=UpperCamelCase_			,						)
         return block_records




     def 		lowerCAmelCase_  (					self							:      List[Any]					):


         """simple docstring"""



         __A       =						RealmRetriever(
             block_records=self.get_dummy_block_records()			,						tokenizer=self.get_tokenizer()			,						)
         return retriever




     def 		lowerCAmelCase_  (					self							:      Dict					):


         """simple docstring"""



         __A       =						self.get_config()
         __A       =						self.get_dummy_retriever()
         __A       =						retriever.tokenizer

         __A       =						np.array([0, 3]			,						dtype="""long"""					)
         __A       =						tokenizer(["""Test question"""]					).input_ids
         __A       =						tokenizer(
             ["""the fourth"""]			,						add_special_tokens=UpperCamelCase_			,						return_token_type_ids=UpperCamelCase_			,						return_attention_mask=UpperCamelCase_			,						).input_ids
         __A       =						config.reader_seq_len

         __A     ,						__A     ,						__A     ,						__A       =						retriever(
             UpperCamelCase_			,						UpperCamelCase_			,						answer_ids=UpperCamelCase_			,						max_length=UpperCamelCase_			,						return_tensors="""np"""					)

         self.assertEqual(len(UpperCamelCase_					)			,						2					)
         self.assertEqual(len(UpperCamelCase_					)			,						2					)
         self.assertEqual(len(UpperCamelCase_					)			,						2					)
         self.assertEqual(concat_inputs.input_ids.shape			,						(2, 10)					)
         self.assertEqual(concat_inputs.attention_mask.shape			,						(2, 10)					)
         self.assertEqual(concat_inputs.token_type_ids.shape			,						(2, 10)					)
         self.assertEqual(concat_inputs.special_tokens_mask.shape			,						(2, 10)					)
         self.assertEqual(
             tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[0]					)			,						["""[CLS]""", """test""", """question""", """[SEP]""", """this""", """is""", """the""", """first""", """record""", """[SEP]"""]			,						)
         self.assertEqual(
             tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[1]					)			,						["""[CLS]""", """test""", """question""", """[SEP]""", """this""", """is""", """the""", """fourth""", """record""", """[SEP]"""]			,						)




     def 		lowerCAmelCase_  (					self							:      str					):


         """simple docstring"""



         __A       =						self.get_config()
         __A       =						self.get_dummy_retriever()
         __A       =						retriever.tokenizer

         __A       =						np.array([0, 3, 5]			,						dtype="""long"""					)
         __A       =						tokenizer(["""Test question"""]					).input_ids
         __A       =						tokenizer(
             ["""the fourth""", """longer longer"""]			,						add_special_tokens=UpperCamelCase_			,						return_token_type_ids=UpperCamelCase_			,						return_attention_mask=UpperCamelCase_			,						).input_ids
         __A       =						config.reader_seq_len

         __A     ,						__A     ,						__A     ,						__A       =						retriever(
             UpperCamelCase_			,						UpperCamelCase_			,						answer_ids=UpperCamelCase_			,						max_length=UpperCamelCase_			,						return_tensors="""np"""					)

         self.assertEqual([False, True, True]			,						UpperCamelCase_					)
         self.assertEqual([[-1, -1, -1], [6, -1, -1], [6, 7, 8]]			,						UpperCamelCase_					)
         self.assertEqual([[-1, -1, -1], [7, -1, -1], [7, 8, 9]]			,						UpperCamelCase_					)





     def 		lowerCAmelCase_  (					self							:      str					):


         """simple docstring"""



         __A       =						self.get_dummy_retriever()
         retriever.save_pretrained(os.path.join(self.tmpdirname			,						"""realm_block_records"""					)					)

         # Test local path
         __A       =						retriever.from_pretrained(os.path.join(self.tmpdirname			,						"""realm_block_records"""					)					)
         self.assertEqual(retriever.block_records[0]			,						B"""This is the first record"""					)

         # Test mocked remote path
         with patch("""transformers.models.realm.retrieval_realm.hf_hub_download"""					) as mock_hf_hub_download:
             __A       =						os.path.join(
                 os.path.join(self.tmpdirname			,						"""realm_block_records"""					)			,						_REALM_BLOCK_RECORDS_FILENAME					)
             __A       =						RealmRetriever.from_pretrained("""google/realm-cc-news-pretrained-openqa"""					)

         self.assertEqual(retriever.block_records[0]			,						B"""This is the first record"""					)




