"""simple docstring"""




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 		__A							(				unittest.TestCase	):






     def 					__A						(	self						):
       # A mock response for an HTTP head request to emulate server down
       _lowerCAmelCase		: str						    =  mock.Mock()
       _lowerCAmelCase		: Any						    =  500
       _lowerCAmelCase		: Optional[int]						    =  {}
       _lowerCAmelCase		: str						    =  HTTPError
       _lowerCAmelCase		: str						    =  {}

       # Download this model to make sure it's in the cache.
       _lowerCAmelCase		: int						    =  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=a__						) as mock_head:
         _lowerCAmelCase		: Optional[int]						    =  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 					__A						(	self						):
       # A mock response for an HTTP head request to emulate server down
       _lowerCAmelCase		: Any						    =  mock.Mock()
       _lowerCAmelCase		: Any						    =  500
       _lowerCAmelCase		: Union[str, Any]						    =  {}
       _lowerCAmelCase		: List[Any]						    =  HTTPError
       _lowerCAmelCase		: List[str]						    =  {}

       # Download this model to make sure it's in the cache.
       _lowerCAmelCase		: List[str]						    =  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=a__						) as mock_head:
         _lowerCAmelCase		: Optional[int]						    =  GPTaTokenizerFast.from_pretrained("""gpt2"""						)
         # This check we did call the fake head request
         mock_head.assert_called()






     def 					__A						(	self						):
       # This test is for deprecated behavior and can be removed in v5
       try:
         _lowerCAmelCase		: Tuple						    =  tempfile.mktemp()
         with open(a__       ,  """wb"""						) as f:
           http_get("""https://huggingface.co/albert-base-v1/resolve/main/spiece.model"""       ,  a__						)

         _lowerCAmelCase		: Optional[int]						    =  AlbertTokenizer.from_pretrained(a__						)
       finally:
         os.remove(a__						)

       # 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"""       ,  a__						)
         _lowerCAmelCase		: Optional[Any]						    =  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       ,  1000						)
         # Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

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






     def 					__A						(	self						):
       # This test is for deprecated behavior and can be removed in v5
       _lowerCAmelCase		: Optional[int]						    =  AlbertTokenizer.from_pretrained("""https://huggingface.co/albert-base-v1/resolve/main/spiece.model"""						)







@is_staging_test
class 		__A							(				unittest.TestCase	):
     _UpperCamelCase							:    int	      =	["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]






     @classmethod
     def 					__A						(	cls						):
       _lowerCAmelCase		: int						    =  TOKEN
       HfFolder.save_token(a__						)






     @classmethod
     def 					__A						(	cls						):
       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 					__A						(	self						):
       with tempfile.TemporaryDirectory() as tmp_dir:
         _lowerCAmelCase		: Optional[Any]						    =  os.path.join(a__       ,  """vocab.txt"""						)
         with open(a__       ,  """w"""       ,  encoding="""utf-8"""						) as vocab_writer:
           vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]						)						)
         _lowerCAmelCase		: Any						    =  BertTokenizer(a__						)

       tokenizer.push_to_hub("""test-tokenizer"""       ,  use_auth_token=self._token						)
       _lowerCAmelCase		: Optional[Any]						    =  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(a__       ,  repo_id="""test-tokenizer"""       ,  push_to_hub=a__       ,  use_auth_token=self._token						)

       _lowerCAmelCase		: str						    =  BertTokenizer.from_pretrained(F"{USER}/test-tokenizer"						)
       self.assertDictEqual(new_tokenizer.vocab       ,  tokenizer.vocab						)






     def 					__A						(	self						):
       with tempfile.TemporaryDirectory() as tmp_dir:
         _lowerCAmelCase		: Dict						    =  os.path.join(a__       ,  """vocab.txt"""						)
         with open(a__       ,  """w"""       ,  encoding="""utf-8"""						) as vocab_writer:
           vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]						)						)
         _lowerCAmelCase		: Optional[int]						    =  BertTokenizer(a__						)

       tokenizer.push_to_hub("""valid_org/test-tokenizer-org"""       ,  use_auth_token=self._token						)
       _lowerCAmelCase		: List[str]						    =  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(
             a__       ,  repo_id="""valid_org/test-tokenizer-org"""       ,  push_to_hub=a__       ,  use_auth_token=self._token						)

       _lowerCAmelCase		: List[str]						    =  BertTokenizer.from_pretrained("""valid_org/test-tokenizer-org"""						)
       self.assertDictEqual(new_tokenizer.vocab       ,  tokenizer.vocab						)






     @require_tokenizers
     def 					__A						(	self						):
       CustomTokenizer.register_for_auto_class()
       with tempfile.TemporaryDirectory() as tmp_dir:
         _lowerCAmelCase		: str						    =  os.path.join(a__       ,  """vocab.txt"""						)
         with open(a__       ,  """w"""       ,  encoding="""utf-8"""						) as vocab_writer:
           vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]						)						)
         _lowerCAmelCase		: Optional[int]						    =  CustomTokenizer(a__						)

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

       _lowerCAmelCase		: int						    =  AutoTokenizer.from_pretrained(F"{USER}/test-dynamic-tokenizer"       ,  trust_remote_code=a__						)
       # 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:
         _lowerCAmelCase		: int						    =  os.path.join(a__       ,  """vocab.txt"""						)
         with open(a__       ,  """w"""       ,  encoding="""utf-8"""						) as vocab_writer:
           vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]						)						)

         _lowerCAmelCase		: List[Any]						    =  BertTokenizerFast.from_pretrained(a__						)
         bert_tokenizer.save_pretrained(a__						)
         _lowerCAmelCase		: Union[str, Any]						    =  CustomTokenizerFast.from_pretrained(a__						)

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

       _lowerCAmelCase		: List[Any]						    =  AutoTokenizer.from_pretrained(F"{USER}/test-dynamic-tokenizer"       ,  trust_remote_code=a__						)
       # 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"""						)
       _lowerCAmelCase		: List[str]						    =  AutoTokenizer.from_pretrained(
           F"{USER}/test-dynamic-tokenizer"       ,  use_fast=a__       ,  trust_remote_code=a__						)
       # 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 		__A							(				unittest.TestCase	):






     def 					__A						(	self						):
       _lowerCAmelCase		: List[Any]						    =  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 					__A						(	self						):
       _lowerCAmelCase		: Dict						    =  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 					__A						(	self						):
       _lowerCAmelCase		: Optional[Any]						    =  Trie()
       trie.add("""A"""						)
       self.assertEqual(trie.split("""ABC"""						)       ,  ["""A""", """BC"""]						)
       self.assertEqual(trie.split("""BCA"""						)       ,  ["""BC""", """A"""]						)






     def 					__A						(	self						):
       _lowerCAmelCase		: int						    =  Trie()
       trie.add("""TOKEN]"""						)
       trie.add("""[SPECIAL_TOKEN]"""						)
       self.assertEqual(trie.split("""This is something [SPECIAL_TOKEN]"""						)       ,  ["""This is something """, """[SPECIAL_TOKEN]"""]						)






     def 					__A						(	self						):
       _lowerCAmelCase		: int						    =  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 					__A						(	self						):
       _lowerCAmelCase		: List[Any]						    =  Trie()
       trie.add("""AB"""						)
       trie.add("""B"""						)
       trie.add("""C"""						)
       self.assertEqual(trie.split("""ABC"""						)       ,  ["""AB""", """C"""]						)






     def 					__A						(	self						):
       _lowerCAmelCase		: str						    =  Trie()
       trie.add("""ABC"""						)
       trie.add("""B"""						)
       trie.add("""CD"""						)
       self.assertEqual(trie.split("""ABCD"""						)       ,  ["""ABC""", """D"""]						)






     def 					__A						(	self						):
       # Even if the offsets are wrong, we necessarily output correct string
       # parts.
       _lowerCAmelCase		: Optional[int]						    =  Trie()
       _lowerCAmelCase		: Any						    =  trie.cut_text("""ABC"""       ,  [0, 0, 2, 1, 2, 3]						)
       self.assertEqual(a__       ,  ["""AB""", """C"""]						)





"""simple docstring"""




from ...utils import (
    OptionalDependencyNotAvailable,
    is_torch_available,
    is_transformers_available,
    is_transformers_version,
)


try:
     if not (is_transformers_available() and is_torch_available() and is_transformers_version('>=', '4.25.0')):
          raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
     from ...utils.dummy_torch_and_transformers_objects import UnCLIPImageVariationPipeline, UnCLIPPipeline
else:
     from .pipeline_unclip import UnCLIPPipeline
     from .pipeline_unclip_image_variation import UnCLIPImageVariationPipeline
     from .text_proj import UnCLIPTextProjModel




