






import os
import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from huggingface_hub.file_download import http_get
from requests.exceptions import HTTPError

from transformers import (
    AlbertTokenizer,
    AutoTokenizer,
    BertTokenizer,
    BertTokenizerFast,
    GPTaTokenizerFast,
    is_tokenizers_available,
)
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_tokenizers
from transformers.tokenization_utils import Trie


sys.path.append(str(Path(__file__).parent.parent / """utils"""))

from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
							from test_module.custom_tokenization_fast import CustomTokenizerFast


class      __SCREAMING_SNAKE_CASE(		unittest.TestCase		):
							def lowerCAmelCase_       (   self:					Dict  )						->						Dict:
											snake_case__													=   mock.Mock()
											snake_case__													=   5_00
											snake_case__													=   {}
											snake_case__													=   HTTPError
											snake_case__													=   {}

											# Download this model to make sure it's in the cache.
											snake_case__													=   BertTokenizer.from_pretrained('hf-internal-testing/tiny-random-bert'  )

											# Under the mock environment we get a 500 error when trying to reach the tokenizer.
											with mock.patch('requests.Session.request'				,		return_value=snake_case_  ) as mock_head:
															snake_case__													=   BertTokenizer.from_pretrained('hf-internal-testing/tiny-random-bert'  )
															# This check we did call the fake head request
															mock_head.assert_called()
							@require_tokenizers
							def lowerCAmelCase_       (   self:					List[Any]  )						->						Optional[int]:
											snake_case__													=   mock.Mock()
											snake_case__													=   5_00
											snake_case__													=   {}
											snake_case__													=   HTTPError
											snake_case__													=   {}

											# Download this model to make sure it's in the cache.
											snake_case__													=   GPTaTokenizerFast.from_pretrained('gpt2'  )

											# Under the mock environment we get a 500 error when trying to reach the tokenizer.
											with mock.patch('requests.Session.request'				,		return_value=snake_case_  ) as mock_head:
															snake_case__													=   GPTaTokenizerFast.from_pretrained('gpt2'  )
															# This check we did call the fake head request
															mock_head.assert_called()
							def lowerCAmelCase_       (   self:					int  )						->						Tuple:

											try:
															snake_case__													=   tempfile.mktemp()
															with open(snake_case_				,		'wb'  ) as f:
																			http_get('https://huggingface.co/albert-base-v1/resolve/main/spiece.model'				,		snake_case_  )

															snake_case__													=   AlbertTokenizer.from_pretrained(snake_case_  )
											finally:
															os.remove(snake_case_  )

											# Supporting this legacy load introduced a weird bug where the tokenizer would load local files if they are in
											# the current folder and have the right name.
											if os.path.isfile('tokenizer.json'  ):
															# We skip the test if the user has a `tokenizer.json` in this folder to avoid deleting it.
															return
											try:
															with open('tokenizer.json'				,		'wb'  ) as f:
																			http_get('https://huggingface.co/hf-internal-testing/tiny-random-bert/blob/main/tokenizer.json'				,		snake_case_  )
															snake_case__													=   AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2'  )
															# The tiny random BERT has a vocab size of 1024, tiny gpt2 as a vocab size of 1000
															self.assertEqual(tokenizer.vocab_size				,		10_00  )
															# Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

											finally:
															os.remove('tokenizer.json'  )


							def lowerCAmelCase_       (   self:					int  )						->						Union[str, Any]:
											snake_case__													=   AlbertTokenizer.from_pretrained('https://huggingface.co/albert-base-v1/resolve/main/spiece.model'  )


@is_staging_test
class      __SCREAMING_SNAKE_CASE(		unittest.TestCase		):
							_UpperCAmelCase       			=				["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]
							@classmethod
							def lowerCAmelCase_       (   cls:					int  )						->						List[str]:
											snake_case__													=   TOKEN
											HfFolder.save_token(snake_case_  )
							@classmethod
							def lowerCAmelCase_       (   cls:					Union[str, Any]  )						->						Union[str, Any]:

											try:
															delete_repo(token=cls._token				,		repo_id='test-tokenizer'  )
											except HTTPError:
															pass

											try:
															delete_repo(token=cls._token				,		repo_id='valid_org/test-tokenizer-org'  )
											except HTTPError:
															pass

											try:
															delete_repo(token=cls._token				,		repo_id='test-dynamic-tokenizer'  )
											except HTTPError:
															pass
							def lowerCAmelCase_       (   self:					List[Any]  )						->						List[Any]:

											with tempfile.TemporaryDirectory() as tmp_dir:
															snake_case__													=   os.path.join(snake_case_				,		'vocab.txt'  )
															with open(snake_case_				,		'w'				,		encoding='utf-8'  ) as vocab_writer:
																			vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]  )  )
															snake_case__													=   BertTokenizer(snake_case_  )

											tokenizer.push_to_hub('test-tokenizer'				,		use_auth_token=self._token  )
											snake_case__													=   BertTokenizer.from_pretrained(F'''{USER}/test-tokenizer'''  )
											self.assertDictEqual(new_tokenizer.vocab				,		tokenizer.vocab  )

											# Reset repo
											delete_repo(token=self._token				,		repo_id='test-tokenizer'  )

											# Push to hub via save_pretrained
											with tempfile.TemporaryDirectory() as tmp_dir:
															tokenizer.save_pretrained(snake_case_				,		repo_id='test-tokenizer'				,		push_to_hub=snake_case_				,		use_auth_token=self._token  )

											snake_case__													=   BertTokenizer.from_pretrained(F'''{USER}/test-tokenizer'''  )
											self.assertDictEqual(new_tokenizer.vocab				,		tokenizer.vocab  )
							def lowerCAmelCase_       (   self:					Tuple  )						->						List[str]:

											with tempfile.TemporaryDirectory() as tmp_dir:
															snake_case__													=   os.path.join(snake_case_				,		'vocab.txt'  )
															with open(snake_case_				,		'w'				,		encoding='utf-8'  ) as vocab_writer:
																			vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]  )  )
															snake_case__													=   BertTokenizer(snake_case_  )

											tokenizer.push_to_hub('valid_org/test-tokenizer-org'				,		use_auth_token=self._token  )
											snake_case__													=   BertTokenizer.from_pretrained('valid_org/test-tokenizer-org'  )
											self.assertDictEqual(new_tokenizer.vocab				,		tokenizer.vocab  )

											# Reset repo
											delete_repo(token=self._token				,		repo_id='valid_org/test-tokenizer-org'  )

											# Push to hub via save_pretrained
											with tempfile.TemporaryDirectory() as tmp_dir:
															tokenizer.save_pretrained(
															    snake_case_				,		repo_id='valid_org/test-tokenizer-org'				,		push_to_hub=snake_case_				,		use_auth_token=self._token  )

											snake_case__													=   BertTokenizer.from_pretrained('valid_org/test-tokenizer-org'  )
											self.assertDictEqual(new_tokenizer.vocab				,		tokenizer.vocab  )


							@require_tokenizers
							def lowerCAmelCase_       (   self:					List[Any]  )						->						List[str]:
											CustomTokenizer.register_for_auto_class()
											with tempfile.TemporaryDirectory() as tmp_dir:
															snake_case__													=   os.path.join(snake_case_				,		'vocab.txt'  )
															with open(snake_case_				,		'w'				,		encoding='utf-8'  ) as vocab_writer:
																			vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]  )  )
															snake_case__													=   CustomTokenizer(snake_case_  )

											# No fast custom tokenizer
											tokenizer.push_to_hub('test-dynamic-tokenizer'				,		use_auth_token=self._token  )

											snake_case__													=   AutoTokenizer.from_pretrained(F'''{USER}/test-dynamic-tokenizer'''				,		trust_remote_code=snake_case_  )
											# Can't make an isinstance check because the new_model.config is from the CustomTokenizer class of a dynamic module
											self.assertEqual(tokenizer.__class__.__name__				,		'CustomTokenizer'  )

											# Fast and slow custom tokenizer
											CustomTokenizerFast.register_for_auto_class()
											with tempfile.TemporaryDirectory() as tmp_dir:
															snake_case__													=   os.path.join(snake_case_				,		'vocab.txt'  )
															with open(snake_case_				,		'w'				,		encoding='utf-8'  ) as vocab_writer:
																			vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]  )  )

															snake_case__													=   BertTokenizerFast.from_pretrained(snake_case_  )
															bert_tokenizer.save_pretrained(snake_case_  )
															snake_case__													=   CustomTokenizerFast.from_pretrained(snake_case_  )

											tokenizer.push_to_hub('test-dynamic-tokenizer'				,		use_auth_token=self._token  )

											snake_case__													=   AutoTokenizer.from_pretrained(F'''{USER}/test-dynamic-tokenizer'''				,		trust_remote_code=snake_case_  )
											# Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
											self.assertEqual(tokenizer.__class__.__name__				,		'CustomTokenizerFast'  )
											snake_case__													=   AutoTokenizer.from_pretrained(
											    F'''{USER}/test-dynamic-tokenizer'''				,		use_fast=snake_case_				,		trust_remote_code=snake_case_  )
											# Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
											self.assertEqual(tokenizer.__class__.__name__				,		'CustomTokenizer'  )


class      __SCREAMING_SNAKE_CASE(		unittest.TestCase		):
							def lowerCAmelCase_       (   self:					List[str]  )						->						List[Any]:
											snake_case__													=   Trie()
											trie.add('Hello 友達'  )
											self.assertEqual(trie.data				,		{'H': {'e': {'l': {'l': {'o': {' ': {'友': {'達': {'': 1}}}}}}}}}  )
											trie.add('Hello'  )
											trie.data
											self.assertEqual(trie.data				,		{'H': {'e': {'l': {'l': {'o': {'': 1, ' ': {'友': {'達': {'': 1}}}}}}}}}  )
							def lowerCAmelCase_       (   self:					Tuple  )						->						Dict:
											snake_case__													=   Trie()
											self.assertEqual(trie.split('[CLS] This is a extra_id_100'  )				,		['[CLS] This is a extra_id_100']  )
											trie.add('[CLS]'  )
											trie.add('extra_id_1'  )
											trie.add('extra_id_100'  )
											self.assertEqual(trie.split('[CLS] This is a extra_id_100'  )				,		['[CLS]', ' This is a ', 'extra_id_100']  )
							def lowerCAmelCase_       (   self:					List[Any]  )						->						Union[str, Any]:
											snake_case__													=   Trie()
											trie.add('A'  )
											self.assertEqual(trie.split('ABC'  )				,		['A', 'BC']  )
											self.assertEqual(trie.split('BCA'  )				,		['BC', 'A']  )
							def lowerCAmelCase_       (   self:					Optional[int]  )						->						List[Any]:
											snake_case__													=   Trie()
											trie.add('TOKEN]'  )
											trie.add('[SPECIAL_TOKEN]'  )
											self.assertEqual(trie.split('This is something [SPECIAL_TOKEN]'  )				,		['This is something ', '[SPECIAL_TOKEN]']  )
							def lowerCAmelCase_       (   self:					List[Any]  )						->						Union[str, Any]:
											snake_case__													=   Trie()
											trie.add('A'  )
											trie.add('P'  )
											trie.add('[SPECIAL_TOKEN]'  )
											self.assertEqual(trie.split('This is something [SPECIAL_TOKEN]'  )				,		['This is something ', '[SPECIAL_TOKEN]']  )
							def lowerCAmelCase_       (   self:					Dict  )						->						List[str]:
											snake_case__													=   Trie()
											trie.add('AB'  )
											trie.add('B'  )
											trie.add('C'  )
											self.assertEqual(trie.split('ABC'  )				,		['AB', 'C']  )
							def lowerCAmelCase_       (   self:					List[Any]  )						->						List[str]:
											snake_case__													=   Trie()
											trie.add('ABC'  )
											trie.add('B'  )
											trie.add('CD'  )
											self.assertEqual(trie.split('ABCD'  )				,		['ABC', 'D']  )


							def lowerCAmelCase_       (   self:					Dict  )						->						Any:
											snake_case__													=   Trie()
											snake_case__													=   trie.cut_text('ABC'				,		[0, 0, 2, 1, 2, 3]  )
											self.assertEqual(snake_case_				,		['AB', 'C']  )











"""simple docstring"""




import copy
import random

from transformers import CLIPTokenizer

class  _UpperCAmelCase  (   UpperCAmelCase__		):



      '''simple docstring'''

      def __init__(     self					,      *snake_case_					,      **snake_case_  ):

        """simple docstring"""

        super().__init__(*snake_case_					,      **snake_case_  )
        A_  :       Tuple										=  {}

      def 					lowerCamelCase_			(     self					,      snake_case_					,      *snake_case_					,      **snake_case_  ):

        """simple docstring"""

        A_  :       str										=  super().add_tokens(snake_case_					,      *snake_case_					,      **snake_case_  )
        if num_added_tokens == 0:
          raise ValueError(
              F"""The tokenizer already contains the token {placeholder_token}. Please pass a different"""
              ' `placeholder_token` that is not already in the tokenizer.'  )

      def 					lowerCamelCase_			(     self					,      snake_case_					,      *snake_case_					,      snake_case_=1					,      **snake_case_  ):

        """simple docstring"""

        A_  :       Tuple										=  []
        if num_vec_per_token == 1:
          self.try_adding_tokens(snake_case_					,      *snake_case_					,      **snake_case_  )
          output.append(snake_case_  )
        else:
          A_  :       Tuple										=  []
          for i in range(snake_case_  ):
            A_  :       List[str]										=  placeholder_token + F"""_{i}"""
            self.try_adding_tokens(snake_case_					,      *snake_case_					,      **snake_case_  )
            output.append(snake_case_  )
        # handle cases where there is a new placeholder token that contains the current placeholder token but is larger
        for token in self.token_map:
          if token in placeholder_token:
            raise ValueError(
                F"""The tokenizer already has placeholder token {token} that can get confused with"""
                F""" {placeholder_token}keep placeholder tokens independent"""  )
        A_  :       Any										=  output

      def 					lowerCamelCase_			(     self					,      snake_case_					,      snake_case_=False					,      snake_case_=1.0  ):

        """simple docstring"""

        if isinstance(snake_case_					,      snake_case_  ):
          A_  :       Optional[Any]										=  []
          for i in range(len(snake_case_  )  ):
            output.append(self.replace_placeholder_tokens_in_text(text[i]					,      vector_shuffle=snake_case_  )  )
          return output
        for placeholder_token in self.token_map:
          if placeholder_token in text:
            A_  :       List[Any]										=  self.token_map[placeholder_token]
            A_  :       Optional[int]										=  tokens[: 1 + int(len(snake_case_  ) * prop_tokens_to_load  )]
            if vector_shuffle:
              A_  :       Optional[Any]										=  copy.copy(snake_case_  )
              random.shuffle(snake_case_  )
            A_  :       List[str]										=  text.replace(snake_case_					,      ' '.join(snake_case_  )  )
        return text

      def __call__(     self					,      snake_case_					,      *snake_case_					,      snake_case_=False					,      snake_case_=1.0					,      **snake_case_  ):

        """simple docstring"""

        return super().__call__(
            self.replace_placeholder_tokens_in_text(
                snake_case_					,      vector_shuffle=snake_case_					,      prop_tokens_to_load=snake_case_  )					,      *snake_case_					,      **snake_case_					,      )

      def 					lowerCamelCase_			(     self					,      snake_case_					,      *snake_case_					,      snake_case_=False					,      snake_case_=1.0					,      **snake_case_  ):

        """simple docstring"""

        return super().encode(
            self.replace_placeholder_tokens_in_text(
                snake_case_					,      vector_shuffle=snake_case_					,      prop_tokens_to_load=snake_case_  )					,      *snake_case_					,      **snake_case_					,      )