import inspect
import os

import torch
from transformers import AutoModel
from transformers.testing_utils import mockenv_context
from transformers.trainer_utils import set_seed

import accelerate
from accelerate.accelerator import Accelerator
from accelerate.state import AcceleratorState
from accelerate.test_utils.testing import (
    AccelerateTestCase,
    TempDirTestCase,
    execute_subprocess_async,
    require_cuda,
    require_fsdp,
    require_multi_gpu,
    slow,
)
from accelerate.utils.constants import (
    FSDP_AUTO_WRAP_POLICY,
    FSDP_BACKWARD_PREFETCH,
    FSDP_SHARDING_STRATEGY,
    FSDP_STATE_DICT_TYPE,
)
from accelerate.utils.dataclasses import FullyShardedDataParallelPlugin
from accelerate.utils.other import patch_environment


set_seed(4_2)

__a  :     str       =  """bert-base-cased"""
__a  :     Dict       =  """fp16"""
__a  :     Dict       =  """bf16"""
__a  :     Optional[int]       =  [FPaa, BFaa]




@require_fsdp
@require_cuda
class    _UpperCamelCase      (						_UpperCAmelCase ):


			"""simple docstring"""






			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					List[str]:

							'''simple docstring'''


							super().setUp()

							__lowercase 			=							dict(
							    ACCELERATE_USE_FSDP='''true'''      ,   MASTER_ADDR='''localhost'''      ,   MASTER_PORT='''10999'''      ,   RANK='''0'''      ,   LOCAL_RANK='''0'''      ,   WORLD_SIZE='''1'''      ,   )





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					int:

							'''simple docstring'''


							from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

							for i, strategy in enumerate(lowerCAmelCase__						):
											__lowercase 			=							self.dist_env.copy()
											__lowercase 			=							F"{i + 1}"
											__lowercase 			=							strategy
											with mockenv_context(**lowerCAmelCase__						):
															__lowercase 			=							FullyShardedDataParallelPlugin()
															self.assertEqual(fsdp_plugin.sharding_strategy      ,   ShardingStrategy(i + 1						)						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					str:

							'''simple docstring'''


							from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

							for i, prefetch_policy in enumerate(lowerCAmelCase__						):
											__lowercase 			=							self.dist_env.copy()
											__lowercase 			=							prefetch_policy
											with mockenv_context(**lowerCAmelCase__						):
															__lowercase 			=							FullyShardedDataParallelPlugin()
															if prefetch_policy == "NO_PREFETCH":
																			self.assertIsNone(fsdp_plugin.backward_prefetch						)
															else:
																			self.assertEqual(fsdp_plugin.backward_prefetch      ,   BackwardPrefetch(i + 1						)						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					Tuple:

							'''simple docstring'''


							from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

							for i, state_dict_type in enumerate(lowerCAmelCase__						):
											__lowercase 			=							self.dist_env.copy()
											__lowercase 			=							state_dict_type
											with mockenv_context(**lowerCAmelCase__						):
															__lowercase 			=							FullyShardedDataParallelPlugin()
															self.assertEqual(fsdp_plugin.state_dict_type      ,   StateDictType(i + 1						)						)
															if state_dict_type == "FULL_STATE_DICT":
																			self.assertTrue(fsdp_plugin.state_dict_config.offload_to_cpu						)
																			self.assertTrue(fsdp_plugin.state_dict_config.ranka_only						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					str:

							'''simple docstring'''


							__lowercase 			=							AutoModel.from_pretrained(lowerCAmelCase__						)
							for policy in FSDP_AUTO_WRAP_POLICY:
											__lowercase 			=							self.dist_env.copy()
											__lowercase 			=							policy
											if policy == "TRANSFORMER_BASED_WRAP":
															__lowercase 			=							'''BertLayer'''
											elif policy == "SIZE_BASED_WRAP":
															__lowercase 			=							'''2000'''
											with mockenv_context(**lowerCAmelCase__						):
															__lowercase 			=							FullyShardedDataParallelPlugin()
															fsdp_plugin.set_auto_wrap_policy(lowerCAmelCase__						)
															if policy == "NO_WRAP":
																			self.assertIsNone(fsdp_plugin.auto_wrap_policy						)
															else:
																			self.assertIsNotNone(fsdp_plugin.auto_wrap_policy						)

							__lowercase 			=							self.dist_env.copy()
							__lowercase 			=							'''TRANSFORMER_BASED_WRAP'''
							__lowercase 			=							'''T5Layer'''
							with mockenv_context(**lowerCAmelCase__						):
											__lowercase 			=							FullyShardedDataParallelPlugin()
											with self.assertRaises(lowerCAmelCase__						) as cm:
															fsdp_plugin.set_auto_wrap_policy(lowerCAmelCase__						)
											self.assertTrue('''Could not find the transformer layer class to wrap in the model.''' in str(cm.exception						)						)

							__lowercase 			=							self.dist_env.copy()
							__lowercase 			=							'''SIZE_BASED_WRAP'''
							__lowercase 			=							'''0'''
							with mockenv_context(**lowerCAmelCase__						):
											__lowercase 			=							FullyShardedDataParallelPlugin()
											fsdp_plugin.set_auto_wrap_policy(lowerCAmelCase__						)
											self.assertIsNone(fsdp_plugin.auto_wrap_policy						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					Dict:

							'''simple docstring'''


							from torch.distributed.fsdp.fully_sharded_data_parallel import MixedPrecision
							from torch.distributed.fsdp.sharded_grad_scaler import ShardedGradScaler

							for mp_dtype in dtypes:
											__lowercase 			=							self.dist_env.copy()
											__lowercase 			=							mp_dtype
											with mockenv_context(**lowerCAmelCase__						):
															__lowercase 			=							Accelerator()
															if mp_dtype == "fp16":
																			__lowercase 			=							torch.floataa
															elif mp_dtype == "bf16":
																			__lowercase 			=							torch.bfloataa
															__lowercase 			=							MixedPrecision(param_dtype=lowerCAmelCase__      ,   reduce_dtype=lowerCAmelCase__      ,   buffer_dtype=lowerCAmelCase__						)
															self.assertEqual(accelerator.state.fsdp_plugin.mixed_precision_policy      ,   lowerCAmelCase__						)
															if mp_dtype == FPaa:
																			self.assertTrue(isinstance(accelerator.scaler      ,   lowerCAmelCase__						)						)
															elif mp_dtype == BFaa:
																			self.assertIsNone(accelerator.scaler						)
															AcceleratorState._reset_state(lowerCAmelCase__						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					Optional[int]:

							'''simple docstring'''


							from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

							for flag in [True, False]:
											__lowercase 			=							self.dist_env.copy()
											__lowercase 			=							str(lowerCAmelCase__						).lower()
											with mockenv_context(**lowerCAmelCase__						):
															__lowercase 			=							FullyShardedDataParallelPlugin()
															self.assertEqual(fsdp_plugin.cpu_offload      ,   CPUOffload(offload_params=lowerCAmelCase__						)						)




@require_fsdp
@require_multi_gpu
@slow
class    _UpperCamelCase      (						_UpperCAmelCase ):


			"""simple docstring"""






			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					str:

							'''simple docstring'''


							super().setUp()
							__lowercase 			=							0.82
							__lowercase 			=							[
							    '''fsdp_shard_grad_op_transformer_based_wrap''',
							    '''fsdp_full_shard_transformer_based_wrap''',
							]
							__lowercase 			=							{
							    '''multi_gpu_fp16''': 32_00,
							    '''fsdp_shard_grad_op_transformer_based_wrap_fp16''': 20_00,
							    '''fsdp_full_shard_transformer_based_wrap_fp16''': 19_00,
							    # Disabling below test as it overwhelms the RAM memory usage
							    # on CI self-hosted runner leading to tests getting killed.
							    # "fsdp_full_shard_cpu_offload_transformer_based_wrap_fp32": 1500,  # fp16 was leading to indefinite hang
							}
							__lowercase 			=							1_60
							__lowercase 			=							1_60

							__lowercase 			=							inspect.getfile(accelerate.test_utils						)
							__lowercase 			=							os.path.sep.join(mod_file.split(os.path.sep						)[:-1] + ['''scripts''', '''external_deps''']						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					List[Any]:

							'''simple docstring'''


							__lowercase 			=							os.path.join(self.test_scripts_folder      ,   '''test_performance.py'''						)
							__lowercase 			=							['''accelerate''', '''launch''', '''--num_processes=2''', '''--num_machines=1''', '''--machine_rank=0''', '''--use_fsdp''']
							for config in self.performance_configs:
											__lowercase 			=							cmd.copy()
											for i, strategy in enumerate(lowerCAmelCase__						):
															if strategy.lower() in config:
																			cmd_config.append(F"--fsdp_sharding_strategy={i+1}"						)
																			break

											if "fp32" in config:
															cmd_config.append('''--mixed_precision=no'''						)
											else:
															cmd_config.append('''--mixed_precision=fp16'''						)

											if "cpu_offload" in config:
															cmd_config.append('''--fsdp_offload_params=True'''						)

											for policy in FSDP_AUTO_WRAP_POLICY:
															if policy.lower() in config:
																			cmd_config.append(F"--fsdp_auto_wrap_policy={policy}"						)
																			break

											if policy == "TRANSFORMER_BASED_WRAP":
															cmd_config.append('''--fsdp_transformer_layer_cls_to_wrap=BertLayer'''						)
											elif policy == "SIZE_BASED_WRAP":
															cmd_config.append('''--fsdp_min_num_params=2000'''						)

											cmd_config.extend(
											    [
											        self.test_file_path,
											        F"--output_dir={self.tmpdir}",
											        F"--performance_lower_bound={self.performance_lower_bound}",
											    ]						)
											with patch_environment(omp_num_threads=1						):
															execute_subprocess_async(lowerCAmelCase__      ,   env=os.environ.copy()						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					List[str]:

							'''simple docstring'''


							__lowercase 			=							os.path.join(self.test_scripts_folder      ,   '''test_checkpointing.py'''						)
							__lowercase 			=							[
							    '''accelerate''',
							    '''launch''',
							    '''--num_processes=2''',
							    '''--num_machines=1''',
							    '''--machine_rank=0''',
							    '''--use_fsdp''',
							    '''--mixed_precision=fp16''',
							    '''--fsdp_transformer_layer_cls_to_wrap=BertLayer''',
							]

							for i, strategy in enumerate(lowerCAmelCase__						):
											__lowercase 			=							cmd.copy()
											cmd_config.append(F"--fsdp_sharding_strategy={i+1}"						)
											if strategy != "FULL_SHARD":
															continue
											__lowercase 			=							len(lowerCAmelCase__						)
											for state_dict_type in FSDP_STATE_DICT_TYPE:
															__lowercase 			=							cmd_config[:state_dict_config_index]
															cmd_config.append(F"--fsdp_state_dict_type={state_dict_type}"						)
															cmd_config.extend(
															    [
															        self.test_file_path,
															        F"--output_dir={self.tmpdir}",
															        '''--partial_train_epoch=1''',
															    ]						)
															with patch_environment(omp_num_threads=1						):
																			execute_subprocess_async(lowerCAmelCase__      ,   env=os.environ.copy()						)

															__lowercase 			=							cmd_config[:-1]
															__lowercase 			=							os.path.join(self.tmpdir      ,   '''epoch_0'''						)
															cmd_config.extend(
															    [
															        F"--resume_from_checkpoint={resume_from_checkpoint}",
															    ]						)
															with patch_environment(omp_num_threads=1						):
																			execute_subprocess_async(lowerCAmelCase__      ,   env=os.environ.copy()						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					int:

							'''simple docstring'''


							__lowercase 			=							os.path.join(self.test_scripts_folder      ,   '''test_peak_memory_usage.py'''						)
							__lowercase 			=							[
							    '''accelerate''',
							    '''launch''',
							    '''--num_processes=2''',
							    '''--num_machines=1''',
							    '''--machine_rank=0''',
							]
							for spec, peak_mem_upper_bound in self.peak_memory_usage_upper_bound.items():
											__lowercase 			=							cmd.copy()
											if "fp16" in spec:
															cmd_config.extend(['''--mixed_precision=fp16''']						)
											else:
															cmd_config.extend(['''--mixed_precision=no''']						)

											if "multi_gpu" in spec:
															continue
											else:
															cmd_config.extend(['''--use_fsdp''']						)
															for i, strategy in enumerate(lowerCAmelCase__						):
																			if strategy.lower() in spec:
																							cmd_config.append(F"--fsdp_sharding_strategy={i+1}"						)
																							break

															if "cpu_offload" in spec:
																			cmd_config.append('''--fsdp_offload_params=True'''						)

															for policy in FSDP_AUTO_WRAP_POLICY:
																			if policy.lower() in spec:
																							cmd_config.append(F"--fsdp_auto_wrap_policy={policy}"						)
																							break

															if policy == "TRANSFORMER_BASED_WRAP":
																			cmd_config.append('''--fsdp_transformer_layer_cls_to_wrap=BertLayer'''						)
															elif policy == "SIZE_BASED_WRAP":
																			cmd_config.append('''--fsdp_min_num_params=2000'''						)

											cmd_config.extend(
											    [
											        self.test_file_path,
											        F"--output_dir={self.tmpdir}",
											        F"--peak_memory_upper_bound={peak_mem_upper_bound}",
											        F"--n_train={self.n_train}",
											        F"--n_val={self.n_val}",
											    ]						)
											with patch_environment(omp_num_threads=1						):
															execute_subprocess_async(lowerCAmelCase__      ,   env=os.environ.copy()						)
import json
import os
import re
import shutil
import tempfile
import unittest
from typing import Tuple

from transformers import AddedToken, BatchEncoding, ByTaTokenizer
from transformers.utils import cached_property, is_tf_available, is_torch_available

from ...test_tokenization_common import TokenizerTesterMixin


if is_torch_available():
		__a  :     Tuple       =  """pt"""
elif is_tf_available():
		__a  :     int       =  """tf"""
else:
		__a  :     Tuple       =  """jax"""




class    _UpperCamelCase      (						_UpperCAmelCase						,unittest.TestCase ):


			"""simple docstring"""






			__a  : List[Any]			 =						ByTaTokenizer
			__a  : str			 =						False





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					List[Any]:

							'''simple docstring'''


							super().setUp()
							__lowercase 			=							ByTaTokenizer()
							tokenizer.save_pretrained(self.tmpdirname						)





			@cached_property
			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					Any:

							'''simple docstring'''


							return ByTaTokenizer.from_pretrained('''google/byt5-small'''						)





			def        _SCREAMING_SNAKE_CASE			(   self      ,   **lowerCAmelCase__						)     ->					ByTaTokenizer:

							'''simple docstring'''


							return self.tokenizer_class.from_pretrained(self.tmpdirname      ,   **lowerCAmelCase__						)





			def        _SCREAMING_SNAKE_CASE			(   self      ,   lowerCAmelCase__      ,   lowerCAmelCase__=False      ,   lowerCAmelCase__=20      ,   lowerCAmelCase__=5						)     ->					Tuple[str, list]:

							'''simple docstring'''


							__lowercase 			=							[]
							for i in range(len(lowerCAmelCase__						)						):
											try:
															__lowercase 			=							tokenizer.decode([i]      ,   clean_up_tokenization_spaces=lowerCAmelCase__						)
											except UnicodeDecodeError:
															pass
											toks.append((i, tok)						)

							__lowercase 			=							list(filter(lambda lowerCAmelCase__						: re.match(R'''^[ a-zA-Z]+$'''      ,   t[1]						)      ,   lowerCAmelCase__						)						)
							__lowercase 			=							list(filter(lambda lowerCAmelCase__						: [t[0]] == tokenizer.encode(t[1]      ,   add_special_tokens=lowerCAmelCase__						)      ,   lowerCAmelCase__						)						)
							if max_length is not None and len(lowerCAmelCase__						) > max_length:
											__lowercase 			=							toks[:max_length]
							if min_length is not None and len(lowerCAmelCase__						) < min_length and len(lowerCAmelCase__						) > 0:
											while len(lowerCAmelCase__						) < min_length:
															__lowercase 			=							toks + toks
        # toks_str = [t[1] for t in toks]
							__lowercase 			=							[t[0] for t in toks]

							# Ensure consistency
							__lowercase 			=							tokenizer.decode(lowerCAmelCase__      ,   clean_up_tokenization_spaces=lowerCAmelCase__						)
							if " " not in output_txt and len(lowerCAmelCase__						) > 1:
											__lowercase 			=							(
											    tokenizer.decode([toks_ids[0]]      ,   clean_up_tokenization_spaces=lowerCAmelCase__						)
											    + ''' '''
											    + tokenizer.decode(toks_ids[1:]      ,   clean_up_tokenization_spaces=lowerCAmelCase__						)
											)
							if with_prefix_space:
											__lowercase 			=							''' ''' + output_txt
							__lowercase 			=							tokenizer.encode(lowerCAmelCase__      ,   add_special_tokens=lowerCAmelCase__						)
							return output_txt, output_ids





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					Tuple:

							'''simple docstring'''


							__lowercase 			=							self.ta_base_tokenizer
							__lowercase 			=							tokenizer(['''hi</s>''', '''I went to the gym</s>''', '''</s>''']						)
							__lowercase 			=							tokenizer(['''hi''', '''I went to the gym''', '''''']						)
							self.assertListEqual(batch_with_eos_added['''input_ids''']      ,   batch_without_eos_added['''input_ids''']						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					Dict:

							'''simple docstring'''


							__lowercase 			=							self.ta_base_tokenizer
							__lowercase 			=							'''Unicode €.'''
							__lowercase 			=							tokenizer(lowerCAmelCase__						)
							__lowercase 			=							[88, 1_13, 1_08, 1_02, 1_14, 1_03, 1_04, 35, 2_29, 1_33, 1_75, 49, 1]
							self.assertEqual(encoded['''input_ids''']      ,   lowerCAmelCase__						)

							# decoding
							__lowercase 			=							tokenizer.decode(lowerCAmelCase__						)
							self.assertEqual(lowerCAmelCase__      ,   '''Unicode €.</s>'''						)

							__lowercase 			=							tokenizer('''e è é ê ë'''						)
							__lowercase 			=							[1_04, 35, 1_98, 1_71, 35, 1_98, 1_72, 35, 1_98, 1_73, 35, 1_98, 1_74, 1]
							self.assertEqual(encoded['''input_ids''']      ,   lowerCAmelCase__						)
							# decoding
							__lowercase 			=							tokenizer.decode(lowerCAmelCase__						)
							self.assertEqual(lowerCAmelCase__      ,   '''e è é ê ë</s>'''						)

							# encode/decode, but with `encode` instead of `__call__`
							self.assertEqual(tokenizer.decode(tokenizer.encode('''e è é ê ë'''						)						)      ,   '''e è é ê ë</s>'''						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					List[Any]:

							'''simple docstring'''


							__lowercase 			=							self.ta_base_tokenizer
							__lowercase 			=							['''A long paragraph for summarization.''', '''Another paragraph for summarization.''']
							# fmt: off
							__lowercase 			=							[68, 35, 1_11, 1_14, 1_13, 1_06, 35, 1_15, 1_00, 1_17, 1_00, 1_06, 1_17, 1_00, 1_15, 1_07, 35, 1_05, 1_14, 1_17, 35, 1_18, 1_20, 1_12, 1_12, 1_00, 1_17, 1_08, 1_25, 1_00, 1_19, 1_08, 1_14, 1_13, 49, 1, 0]
							# fmt: on
							__lowercase 			=							tokenizer(lowerCAmelCase__      ,   padding=lowerCAmelCase__      ,   return_tensors=lowerCAmelCase__						)
							self.assertIsInstance(lowerCAmelCase__      ,   lowerCAmelCase__						)

							if FRAMEWORK != "jax":
											__lowercase 			=							list(batch.input_ids.numpy()[0]						)
							else:
											__lowercase 			=							list(batch.input_ids.tolist()[0]						)

							self.assertListEqual(lowerCAmelCase__      ,   lowerCAmelCase__						)

							self.assertEqual((2, 37)      ,   batch.input_ids.shape						)
							self.assertEqual((2, 37)      ,   batch.attention_mask.shape						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					int:

							'''simple docstring'''


							__lowercase 			=							self.ta_base_tokenizer
							__lowercase 			=							['''A long paragraph for summarization.''', '''Another paragraph for summarization.''']
							__lowercase 			=							tokenizer(lowerCAmelCase__      ,   padding=lowerCAmelCase__      ,   return_tensors=lowerCAmelCase__						)
							# check if input_ids are returned and no decoder_input_ids
							self.assertIn('''input_ids'''      ,   lowerCAmelCase__						)
							self.assertIn('''attention_mask'''      ,   lowerCAmelCase__						)
							self.assertNotIn('''decoder_input_ids'''      ,   lowerCAmelCase__						)
							self.assertNotIn('''decoder_attention_mask'''      ,   lowerCAmelCase__						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					Tuple:

							'''simple docstring'''


							__lowercase 			=							self.ta_base_tokenizer
							__lowercase 			=							[
							    '''Summary of the text.''',
							    '''Another summary.''',
							]
							__lowercase 			=							tokenizer(
							    text_target=lowerCAmelCase__      ,   max_length=32      ,   padding='''max_length'''      ,   truncation=lowerCAmelCase__      ,   return_tensors=lowerCAmelCase__						)
							self.assertEqual(32      ,   targets['''input_ids'''].shape[1]						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					int:

							'''simple docstring'''


							__lowercase 			=							self.ta_base_tokenizer
							__lowercase 			=							['''A long paragraph for summarization. </s>''']
							__lowercase 			=							['''Summary of the text. </s>''']
							# fmt: off
							__lowercase 			=							[68, 35, 1_11, 1_14, 1_13, 1_06, 35, 1_15, 1_00, 1_17, 1_00, 1_06, 1_17, 1_00, 1_15, 1_07, 35, 1_05, 1_14, 1_17, 35, 1_18, 1_20, 1_12, 1_12, 1_00, 1_17, 1_08, 1_25, 1_00, 1_19, 1_08, 1_14, 1_13, 49, 35, 1]
							__lowercase 			=							[86, 1_20, 1_12, 1_12, 1_00, 1_17, 1_24, 35, 1_14, 1_05, 35, 1_19, 1_07, 1_04, 35, 1_19, 1_04, 1_23, 1_19, 49, 35, 1]
							# fmt: on

							__lowercase 			=							tokenizer(lowerCAmelCase__      ,   text_target=lowerCAmelCase__						)

							self.assertEqual(lowerCAmelCase__      ,   batch['''input_ids'''][0]						)
							self.assertEqual(lowerCAmelCase__      ,   batch['''labels'''][0]						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					List[str]:

							'''simple docstring'''


							__lowercase 			=							self.get_tokenizers()
							for tokenizer in tokenizers:
											with self.subTest(F"{tokenizer.__class__.__name__}"						):
															self.assertNotEqual(tokenizer.model_max_length      ,   42						)

        # Now let's start the test
							__lowercase 			=							self.get_tokenizers()
							for tokenizer in tokenizers:
											with self.subTest(F"{tokenizer.__class__.__name__}"						):
															# Isolate this from the other tests because we save additional tokens/etc
															__lowercase 			=							tempfile.mkdtemp()

															__lowercase 			=							''' He is very happy, UNwant\u00E9d,running'''
															__lowercase 			=							tokenizer.encode(lowerCAmelCase__      ,   add_special_tokens=lowerCAmelCase__						)
															tokenizer.save_pretrained(lowerCAmelCase__						)

															__lowercase 			=							tokenizer.__class__.from_pretrained(lowerCAmelCase__						)
															__lowercase 			=							after_tokenizer.encode(lowerCAmelCase__      ,   add_special_tokens=lowerCAmelCase__						)
															self.assertListEqual(lowerCAmelCase__      ,   lowerCAmelCase__						)

															shutil.rmtree(lowerCAmelCase__						)

							__lowercase 			=							self.get_tokenizers(model_max_length=42						)
							for tokenizer in tokenizers:
											with self.subTest(F"{tokenizer.__class__.__name__}"						):
															# Isolate this from the other tests because we save additional tokens/etc
															__lowercase 			=							tempfile.mkdtemp()

															__lowercase 			=							''' He is very happy, UNwant\u00E9d,running'''
															tokenizer.add_tokens(['''bim''', '''bambam''']						)
															__lowercase 			=							tokenizer.additional_special_tokens
															additional_special_tokens.append('''new_additional_special_token'''						)
															tokenizer.add_special_tokens({'''additional_special_tokens''': additional_special_tokens}						)
															__lowercase 			=							tokenizer.encode(lowerCAmelCase__      ,   add_special_tokens=lowerCAmelCase__						)
															tokenizer.save_pretrained(lowerCAmelCase__						)

															__lowercase 			=							tokenizer.__class__.from_pretrained(lowerCAmelCase__						)
															__lowercase 			=							after_tokenizer.encode(lowerCAmelCase__      ,   add_special_tokens=lowerCAmelCase__						)
															self.assertListEqual(lowerCAmelCase__      ,   lowerCAmelCase__						)
															self.assertIn('''new_additional_special_token'''      ,   after_tokenizer.additional_special_tokens						)
															self.assertEqual(after_tokenizer.model_max_length      ,   42						)

															__lowercase 			=							tokenizer.__class__.from_pretrained(lowerCAmelCase__      ,   model_max_length=43						)
															self.assertEqual(tokenizer.model_max_length      ,   43						)

															shutil.rmtree(lowerCAmelCase__						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					Dict:

							'''simple docstring'''


							__lowercase 			=							[]
							if self.test_slow_tokenizer:
											tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())						)

							if self.test_rust_tokenizer:
											tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())						)

							for tokenizer_class, tokenizer_utils in tokenizer_list:
											with tempfile.TemporaryDirectory() as tmp_dir:
															tokenizer_utils.save_pretrained(lowerCAmelCase__						)

															with open(os.path.join(lowerCAmelCase__      ,   '''special_tokens_map.json'''						)      ,   encoding='''utf-8'''						) as json_file:
																			__lowercase 			=							json.load(lowerCAmelCase__						)

															with open(os.path.join(lowerCAmelCase__      ,   '''tokenizer_config.json'''						)      ,   encoding='''utf-8'''						) as json_file:
																			__lowercase 			=							json.load(lowerCAmelCase__						)

															__lowercase 			=							[F"<extra_id_{i}>" for i in range(1_25						)]

															__lowercase 			=							added_tokens_extra_ids + [
															    '''an_additional_special_token'''
															]
															__lowercase 			=							added_tokens_extra_ids + [
															    '''an_additional_special_token'''
															]

															with open(os.path.join(lowerCAmelCase__      ,   '''special_tokens_map.json'''						)      ,   '''w'''      ,   encoding='''utf-8'''						) as outfile:
																			json.dump(lowerCAmelCase__      ,   lowerCAmelCase__						)
															with open(os.path.join(lowerCAmelCase__      ,   '''tokenizer_config.json'''						)      ,   '''w'''      ,   encoding='''utf-8'''						) as outfile:
																			json.dump(lowerCAmelCase__      ,   lowerCAmelCase__						)

															# the following checks allow us to verify that our test works as expected, i.e. that the tokenizer takes
															# into account the new value of additional_special_tokens given in the "tokenizer_config.json" and
															# "special_tokens_map.json" files
															__lowercase 			=							tokenizer_class.from_pretrained(
															    lowerCAmelCase__      ,   )
															self.assertIn(
															    '''an_additional_special_token'''      ,   tokenizer_without_change_in_init.additional_special_tokens						)
															# self.assertIn("an_additional_special_token",tokenizer_without_change_in_init.get_vocab()) # ByT5Tokenization no vocab
															self.assertEqual(
															    ['''an_additional_special_token''']      ,   tokenizer_without_change_in_init.convert_ids_to_tokens(
															        tokenizer_without_change_in_init.convert_tokens_to_ids(['''an_additional_special_token''']						)						)      ,   )

															# Now we test that we can change the value of additional_special_tokens in the from_pretrained
															__lowercase 			=							added_tokens_extra_ids + [AddedToken('''a_new_additional_special_token'''      ,   lstrip=lowerCAmelCase__						)]
															__lowercase 			=							tokenizer_class.from_pretrained(
															    lowerCAmelCase__      ,   additional_special_tokens=lowerCAmelCase__      ,   )

															self.assertIn('''a_new_additional_special_token'''      ,   tokenizer.additional_special_tokens						)
															self.assertEqual(
															    ['''a_new_additional_special_token''']      ,   tokenizer.convert_ids_to_tokens(
															        tokenizer.convert_tokens_to_ids(['''a_new_additional_special_token''']						)						)      ,   )





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					List[str]:

							'''simple docstring'''


							__lowercase 			=							[]
							if self.test_slow_tokenizer:
											tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())						)

							if self.test_rust_tokenizer:
											tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())						)

							for tokenizer_class, tokenizer_utils in tokenizer_list:
											with tempfile.TemporaryDirectory() as tmp_dir:
															tokenizer_utils.save_pretrained(lowerCAmelCase__						)

															__lowercase 			=							tokenizer_class.from_pretrained(lowerCAmelCase__						)

															self.assertTrue(tokenizer.decode([2_55]						) == ''''''						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					Optional[int]:

							'''simple docstring'''


							pass





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					Union[str, Any]:

							'''simple docstring'''


							pass





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					int:

							'''simple docstring'''


							pass





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					str:

							'''simple docstring'''


							pass





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					Optional[int]:

							'''simple docstring'''


							__lowercase 			=							self.get_tokenizers(fast=lowerCAmelCase__      ,   do_lower_case=lowerCAmelCase__						)
							for tokenizer in tokenizers:
											with self.subTest(F"{tokenizer.__class__.__name__}"						):
															__lowercase 			=							['''t''', '''h''', '''i''', '''s''', ''' ''', '''i''', '''s''', ''' ''', '''a''', ''' ''', '''t''', '''e''', '''x''', '''t''', '''</s>''']
															__lowercase 			=							tokenizer.convert_tokens_to_string(lowerCAmelCase__						)

															self.assertIsInstance(lowerCAmelCase__      ,   lowerCAmelCase__						)





			def        _SCREAMING_SNAKE_CASE			(   self						)     ->					int:

							'''simple docstring'''


							__lowercase 			=							self.get_tokenizers()
							for tokenizer in tokenizers:
											with self.subTest(F"{tokenizer.__class__.__name__}"						):
															__lowercase 			=							[
															    '''bos_token''',
															    '''eos_token''',
															    '''unk_token''',
															    '''sep_token''',
															    '''pad_token''',
															    '''cls_token''',
															    '''mask_token''',
															]

															__lowercase 			=							0
															__lowercase 			=							tokenizer.convert_ids_to_tokens(
															    lowerCAmelCase__      ,   skip_special_tokens=lowerCAmelCase__						)

															for attr in attributes_list:
																			setattr(lowerCAmelCase__      ,   attr + '''_id'''      ,   lowerCAmelCase__						)
																			self.assertEqual(getattr(lowerCAmelCase__      ,   lowerCAmelCase__						)      ,   lowerCAmelCase__						)
																			self.assertEqual(getattr(lowerCAmelCase__      ,   attr + '''_id'''						)      ,   lowerCAmelCase__						)

																			setattr(lowerCAmelCase__      ,   attr + '''_id'''      ,   lowerCAmelCase__						)
																			self.assertEqual(getattr(lowerCAmelCase__      ,   lowerCAmelCase__						)      ,   lowerCAmelCase__						)
																			self.assertEqual(getattr(lowerCAmelCase__      ,   attr + '''_id'''						)      ,   lowerCAmelCase__						)

															setattr(lowerCAmelCase__      ,   '''additional_special_tokens_ids'''      ,   []						)
															self.assertListEqual(getattr(lowerCAmelCase__      ,   '''additional_special_tokens'''						)      ,   []						)
															self.assertListEqual(getattr(lowerCAmelCase__      ,   '''additional_special_tokens_ids'''						)      ,   []						)

															setattr(lowerCAmelCase__      ,   '''additional_special_tokens_ids'''      ,   [token_id_to_test_setters]						)
															self.assertListEqual(getattr(lowerCAmelCase__      ,   '''additional_special_tokens'''						)      ,   [token_to_test_setters]						)
															self.assertListEqual(getattr(lowerCAmelCase__      ,   '''additional_special_tokens_ids'''						)      ,   [token_id_to_test_setters]						)