



'''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			):


	def    _a     (self							):
				# A mock response for an HTTP head request to emulate server down
				A_     :						List[Any]     =					mock.Mock()
				A_     :						Tuple     =					500
				A_     :						Any     =					{}
				A_     :						List[str]     =					HTTPError
				A_     :						List[str]     =					{}

				# Download this model to make sure it's in the cache.
				A_     :						Optional[Any]     =					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=lowercase							) as mock_head:
							A_     :						Optional[Any]     =					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
				A_     :						Any     =					mock.Mock()
				A_     :						Optional[Any]     =					500
				A_     :						Any     =					{}
				A_     :						List[str]     =					HTTPError
				A_     :						Optional[int]     =					{}

				# Download this model to make sure it's in the cache.
				A_     :						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=lowercase							) as mock_head:
							A_     :						Any     =					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:
							A_     :						List[str]     =					tempfile.mktemp()
							with open(lowercase	,				"""wb"""							) as f:
										http_get("""https://huggingface.co/albert-base-v1/resolve/main/spiece.model"""	,				lowercase							)

							A_     :						List[str]     =					AlbertTokenizer.from_pretrained(lowercase							)
				finally:
							os.remove(lowercase							)

				# 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"""	,				lowercase							)
							A_     :						int     =					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
				A_     :						Optional[int]     =					AlbertTokenizer.from_pretrained("""https://huggingface.co/albert-base-v1/resolve/main/spiece.model"""							)







@is_staging_test
class      _lowerCAmelCase							(      unittest.TestCase			):
	__SCREAMING_SNAKE_CASE       :		List[str]		      = ['[UNK]', '[CLS]', '[SEP]', '[PAD]', '[MASK]', 'bla', 'blou']


	@classmethod
	def    _a     (cls							):
				A_     :						List[str]     =					TOKEN
				HfFolder.save_token(lowercase							)


	@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:
							A_     :						List[Any]     =					os.path.join(lowercase	,				"""vocab.txt"""							)
							with open(lowercase	,				"""w"""	,				encoding="""utf-8"""							) as vocab_writer:
										vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]							)							)
							A_     :						Optional[int]     =					BertTokenizer(lowercase							)

				tokenizer.push_to_hub("""test-tokenizer"""	,				use_auth_token=self._token							)
				A_     :						Dict     =					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(lowercase	,				repo_id="""test-tokenizer"""	,				push_to_hub=lowercase	,				use_auth_token=self._token							)

				A_     :						Tuple     =					BertTokenizer.from_pretrained(F'{USER}/test-tokenizer'							)
				self.assertDictEqual(new_tokenizer.vocab	,				tokenizer.vocab							)


	def    _a     (self							):
				with tempfile.TemporaryDirectory() as tmp_dir:
							A_     :						Optional[int]     =					os.path.join(lowercase	,				"""vocab.txt"""							)
							with open(lowercase	,				"""w"""	,				encoding="""utf-8"""							) as vocab_writer:
										vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]							)							)
							A_     :						Union[str, Any]     =					BertTokenizer(lowercase							)

				tokenizer.push_to_hub("""valid_org/test-tokenizer-org"""	,				use_auth_token=self._token							)
				A_     :						int     =					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(
							    lowercase	,				repo_id="""valid_org/test-tokenizer-org"""	,				push_to_hub=lowercase	,				use_auth_token=self._token							)

				A_     :						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:
							A_     :						Optional[Any]     =					os.path.join(lowercase	,				"""vocab.txt"""							)
							with open(lowercase	,				"""w"""	,				encoding="""utf-8"""							) as vocab_writer:
										vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]							)							)
							A_     :						Dict     =					CustomTokenizer(lowercase							)

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

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

							A_     :						Optional[Any]     =					BertTokenizerFast.from_pretrained(lowercase							)
							bert_tokenizer.save_pretrained(lowercase							)
							A_     :						Any     =					CustomTokenizerFast.from_pretrained(lowercase							)

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

				A_     :						Dict     =					AutoTokenizer.from_pretrained(F'{USER}/test-dynamic-tokenizer'	,				trust_remote_code=lowercase							)
				# 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"""							)
				A_     :						Union[str, Any]     =					AutoTokenizer.from_pretrained(
				    F'{USER}/test-dynamic-tokenizer'	,				use_fast=lowercase	,				trust_remote_code=lowercase							)
				# 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			):


	def    _a     (self							):
				A_     :						Union[str, 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							):
				A_     :						Union[str, Any]     =					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							):
				A_     :						List[Any]     =					Trie()
				trie.add("""A"""							)
				self.assertEqual(trie.split("""ABC"""							)	,				["""A""", """BC"""]							)
				self.assertEqual(trie.split("""BCA"""							)	,				["""BC""", """A"""]							)


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


	def    _a     (self							):
				A_     :						Optional[Any]     =					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.
				A_     :						Any     =					Trie()
				A_     :						Union[str, Any]     =					trie.cut_text("""ABC"""	,				[0, 0, 2, 1, 2, 3]							)
				self.assertEqual(lowercase	,				["""AB""", """C"""]							)




'''simple docstring'''




import unittest
from typing import Tuple

import torch

from diffusers.utils import floats_tensor, randn_tensor, torch_all_close, torch_device
from diffusers.utils.testing_utils import require_torch







@require_torch
class      _lowerCAmelCase							:


	@property
	def    _a     (self							):
				return self.get_dummy_input()


	@property
	def    _a     (self							):
				if self.block_type == "down":
							return (4, 32, 16, 16)
				elif self.block_type == "mid":
							return (4, 32, 32, 32)
				elif self.block_type == "up":
							return (4, 32, 64, 64)

				raise ValueError(F'\'{self.block_type}\' is not a supported block_type. Set it to \'up\', \'mid\', or \'down\'.'							)


	def    _a     (self	,				lowercase=True	,				lowercase=False	,				lowercase=False	,				lowercase=False	,				):
				A_     :						List[str]     =					4
				A_     :						int     =					32
				A_     :						Optional[int]     =					(32, 32)

				A_     :						Optional[Any]     =					torch.manual_seed(0							)
				A_     :						int     =					torch.device(lowercase							)
				A_     :						int     =					(batch_size, num_channels) + sizes
				A_     :						Optional[int]     =					randn_tensor(lowercase	,				generator=lowercase	,				device=lowercase							)
				A_     :						Union[str, Any]     =					{"""hidden_states""": hidden_states}

				if include_temb:
							A_     :						str     =					128
							A_     :						List[Any]     =					randn_tensor((batch_size, temb_channels)	,				generator=lowercase	,				device=lowercase							)

				if include_res_hidden_states_tuple:
							A_     :						List[str]     =					torch.manual_seed(1							)
							A_     :						int     =					(randn_tensor(lowercase	,				generator=lowercase	,				device=lowercase							),)

				if include_encoder_hidden_states:
							A_     :						List[str]     =					floats_tensor((batch_size, 32, 32)							).to(lowercase							)

				if include_skip_sample:
							A_     :						Dict     =					randn_tensor(((batch_size, 3) + sizes)	,				generator=lowercase	,				device=lowercase							)

				return dummy_input


	def    _a     (self							):
				A_     :						Tuple     =					{
				    """in_channels""": 32,
				    """out_channels""": 32,
				    """temb_channels""": 128,
				}
				if self.block_type == "up":
							A_     :						Any     =					32

				if self.block_type == "mid":
							init_dict.pop("""out_channels"""							)

				A_     :						Optional[int]     =					self.dummy_input
				return init_dict, inputs_dict


	def    _a     (self	,				lowercase							):
				A_,   A_     :						Optional[Any]     =					self.prepare_init_args_and_inputs_for_common()
				A_     :						int     =					self.block_class(**lowercase							)
				unet_block.to(lowercase							)
				unet_block.eval()

				with torch.no_grad():
							A_     :						List[str]     =					unet_block(**lowercase							)

				if isinstance(lowercase	,				lowercase							):
							A_     :						Union[str, Any]     =					output[0]

				self.assertEqual(output.shape	,				self.output_shape							)

				A_     :						int     =					output[0, -1, -3:, -3:]
				A_     :						List[Any]     =					torch.tensor(lowercase							).to(lowercase							)
				assert torch_all_close(output_slice.flatten()	,				lowercase	,				atol=5E-3							)


	@unittest.skipIf(torch_device == """mps"""	,				"""Training is not supported in mps"""							)
	def    _a     (self							):
				A_,   A_     :						Tuple     =					self.prepare_init_args_and_inputs_for_common()
				A_     :						List[str]     =					self.block_class(**lowercase							)
				model.to(lowercase							)
				model.train()
				A_     :						Any     =					model(**lowercase							)

				if isinstance(lowercase	,				lowercase							):
							A_     :						str     =					output[0]

				A_     :						Union[str, Any]     =					torch.device(lowercase							)
				A_     :						Tuple     =					randn_tensor(output.shape	,				device=lowercase							)
				A_     :						List[str]     =					torch.nn.functional.mse_loss(lowercase	,				lowercase							)
				loss.backward()