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







							'''simple docstring'''

							def 	lowercase							(self  )					->       int:
														_snake_case       =  mock.Mock()
														_snake_case       =  500
														_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 	lowercase							(self  )					->       List[str]:
														_snake_case       =  mock.Mock()
														_snake_case       =  500
														_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 	lowercase							(self  )					->       str:





														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  ,  1000  )
																					# Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

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





							def 	lowercase							(self  )					->       Tuple:
														_snake_case       =  AlbertTokenizer.from_pretrained("""https://huggingface.co/albert-base-v1/resolve/main/spiece.model"""  )


@is_staging_test
class       _lowerCAmelCase   (   unittest.TestCase				):







							'''simple docstring'''

							lowerCAmelCase_ 						=						["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]

							@classmethod
							def 	lowercase							(cls  )					->       Dict:
														_snake_case       =  TOKEN
														HfFolder.save_token(snake_case__  )

							@classmethod
							def 	lowercase							(cls  )					->       Optional[int]:





														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 	lowercase							(self  )					->       Dict:





														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 	lowercase							(self  )					->       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("""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 	lowercase							(self  )					->       int:
														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       _lowerCAmelCase   (   unittest.TestCase				):







							'''simple docstring'''

							def 	lowercase							(self  )					->       Optional[int]:
														_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 	lowercase							(self  )					->       Optional[int]:
														_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 	lowercase							(self  )					->       Dict:
														_snake_case       =  Trie()
														trie.add("""A"""  )
														self.assertEqual(trie.split("""ABC"""  )  ,  ["""A""", """BC"""]  )
														self.assertEqual(trie.split("""BCA"""  )  ,  ["""BC""", """A"""]  )

							def 	lowercase							(self  )					->       Tuple:
														_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 	lowercase							(self  )					->       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 	lowercase							(self  )					->       str:
														_snake_case       =  Trie()
														trie.add("""AB"""  )
														trie.add("""B"""  )
														trie.add("""C"""  )
														self.assertEqual(trie.split("""ABC"""  )  ,  ["""AB""", """C"""]  )

							def 	lowercase							(self  )					->       Dict:
														_snake_case       =  Trie()
														trie.add("""ABC"""  )
														trie.add("""B"""  )
														trie.add("""CD"""  )
														self.assertEqual(trie.split("""ABCD"""  )  ,  ["""ABC""", """D"""]  )





							def 	lowercase							(self  )					->       Tuple:
														_snake_case       =  Trie()
														_snake_case       =  trie.cut_text("""ABC"""  ,  [0, 0, 2, 1, 2, 3]  )
														self.assertEqual(snake_case__  ,  ["""AB""", """C"""]  )

'''simple docstring'''
from ...utils import (
    OptionalDependencyNotAvailable,
    is_flax_available,
    is_torch_available,
    is_transformers_available,
)


try:
				if not (is_transformers_available() and is_torch_available()):
								raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
				from ...utils.dummy_torch_and_transformers_objects import *  # noqa F403
else:
				from .multicontrolnet import MultiControlNetModel
				from .pipeline_controlnet import StableDiffusionControlNetPipeline
				from .pipeline_controlnet_imgaimg import StableDiffusionControlNetImgaImgPipeline
				from .pipeline_controlnet_inpaint import StableDiffusionControlNetInpaintPipeline


if is_transformers_available() and is_flax_available():
				from .pipeline_flax_controlnet import FlaxStableDiffusionControlNetPipeline