



'''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 				lowercase__  (self : List[str]      )     ->			Any:

     """simple docstring"""

     SCREAMING_SNAKE_CASE  :						Optional[int]      					= mock.Mock()
     SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= 500
     SCREAMING_SNAKE_CASE  :						str      					= {}
     SCREAMING_SNAKE_CASE  :						List[str]      					= HTTPError
     SCREAMING_SNAKE_CASE  :						Optional[Any]      					= {}

     # Download this model to make sure it's in the cache.
     SCREAMING_SNAKE_CASE  :						List[str]      					= 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:
         SCREAMING_SNAKE_CASE  :						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 				lowercase__  (self : Optional[int]      )     ->			Tuple:

     """simple docstring"""

     SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= mock.Mock()
     SCREAMING_SNAKE_CASE  :						Any      					= 500
     SCREAMING_SNAKE_CASE  :						int      					= {}
     SCREAMING_SNAKE_CASE  :						int      					= HTTPError
     SCREAMING_SNAKE_CASE  :						Any      					= {}

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


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

     """simple docstring"""

     try:
         SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= tempfile.mktemp()
         with open(a_,					'''wb'''      ) as f:
             http_get('''https://huggingface.co/albert-base-v1/resolve/main/spiece.model''',					a_      )

         SCREAMING_SNAKE_CASE  :						Dict      					= 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_      )
         SCREAMING_SNAKE_CASE  :						Tuple      					= 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 : Optional[Any]      )     ->			int:

     """simple docstring"""

     SCREAMING_SNAKE_CASE  :						Tuple      					= AlbertTokenizer.from_pretrained('''https://huggingface.co/albert-base-v1/resolve/main/spiece.model'''      )



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


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

     """simple docstring"""

     SCREAMING_SNAKE_CASE  :						Tuple      					= TOKEN
     HfFolder.save_token(a_      )


 @classmethod
 def 				lowercase__  (cls : List[str]      )     ->			Dict:

     """simple docstring"""

     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 : Any      )     ->			Union[str, Any]:

     """simple docstring"""

     with tempfile.TemporaryDirectory() as tmp_dir:
         SCREAMING_SNAKE_CASE  :						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]      )      )
         SCREAMING_SNAKE_CASE  :						str      					= BertTokenizer(a_      )

     tokenizer.push_to_hub('''test-tokenizer''',					use_auth_token=self._token      )
     SCREAMING_SNAKE_CASE  :						List[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      )

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


 def 				lowercase__  (self : str      )     ->			List[Any]:

     """simple docstring"""

     with tempfile.TemporaryDirectory() as tmp_dir:
         SCREAMING_SNAKE_CASE  :						List[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]      )      )
         SCREAMING_SNAKE_CASE  :						Any      					= BertTokenizer(a_      )

     tokenizer.push_to_hub('''valid_org/test-tokenizer-org''',					use_auth_token=self._token      )
     SCREAMING_SNAKE_CASE  :						Dict      					= 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      )

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






 @require_tokenizers
 def 				lowercase__  (self : List[Any]      )     ->			Dict:

     """simple docstring"""

     CustomTokenizer.register_for_auto_class()
     with tempfile.TemporaryDirectory() as tmp_dir:
         SCREAMING_SNAKE_CASE  :						Optional[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]      )      )
         SCREAMING_SNAKE_CASE  :						Tuple      					= CustomTokenizer(a_      )

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

     SCREAMING_SNAKE_CASE  :						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:
         SCREAMING_SNAKE_CASE  :						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]      )      )

         SCREAMING_SNAKE_CASE  :						List[Any]      					= BertTokenizerFast.from_pretrained(a_      )
         bert_tokenizer.save_pretrained(a_      )
         SCREAMING_SNAKE_CASE  :						List[str]      					= CustomTokenizerFast.from_pretrained(a_      )

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

     SCREAMING_SNAKE_CASE  :						Dict      					= 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'''      )
     SCREAMING_SNAKE_CASE  :						Dict      					= 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 				lowercase__  (self : Tuple      )     ->			Dict:

     """simple docstring"""

     SCREAMING_SNAKE_CASE  :						Optional[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 				lowercase__  (self : int      )     ->			List[Any]:

     """simple docstring"""

     SCREAMING_SNAKE_CASE  :						int      					= 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      )     ->			Tuple:

     """simple docstring"""

     SCREAMING_SNAKE_CASE  :						Any      					= Trie()
     trie.add('''A'''      )
     self.assertEqual(trie.split('''ABC'''      ),					['''A''', '''BC''']      )
     self.assertEqual(trie.split('''BCA'''      ),					['''BC''', '''A''']      )


 def 				lowercase__  (self : str      )     ->			int:

     """simple docstring"""

     SCREAMING_SNAKE_CASE  :						Tuple      					= 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 : List[str]      )     ->			Any:

     """simple docstring"""

     SCREAMING_SNAKE_CASE  :						List[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 				lowercase__  (self : Optional[Any]      )     ->			List[str]:

     """simple docstring"""

     SCREAMING_SNAKE_CASE  :						Optional[Any]      					= Trie()
     trie.add('''AB'''      )
     trie.add('''B'''      )
     trie.add('''C'''      )
     self.assertEqual(trie.split('''ABC'''      ),					['''AB''', '''C''']      )


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

     """simple docstring"""

     SCREAMING_SNAKE_CASE  :						str      					= Trie()
     trie.add('''ABC'''      )
     trie.add('''B'''      )
     trie.add('''CD'''      )
     self.assertEqual(trie.split('''ABCD'''      ),					['''ABC''', '''D''']      )






 def 				lowercase__  (self : int      )     ->			List[str]:

     """simple docstring"""

     SCREAMING_SNAKE_CASE  :						Dict      					= Trie()
     SCREAMING_SNAKE_CASE  :						str      					= trie.cut_text('''ABC''',					[0, 0, 2, 1, 2, 3]      )
     self.assertEqual(a_,					['''AB''', '''C''']      )









'''simple docstring'''

import inspect
import os
import unittest
from dataclasses import dataclass

import torch

from accelerate import Accelerator, DistributedDataParallelKwargs, GradScalerKwargs
from accelerate.state import AcceleratorState
from accelerate.test_utils import execute_subprocess_async, require_cuda, require_multi_gpu
from accelerate.utils import KwargsHandler


@dataclass
class 	a__     (				_lowercase	):
	__magic_name__		:    int	    =     0
	__magic_name__		:    bool	    =     False
	__magic_name__		:    float	    =     3.0



class 	a__     (				unittest.TestCase	):


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

					"""simple docstring"""

					self.assertDictEqual(MockClass().to_kwargs(),					{}      )
					self.assertDictEqual(MockClass(a=2      ).to_kwargs(),					{'''a''': 2}      )
					self.assertDictEqual(MockClass(a=2,					b=__UpperCAmelCase      ).to_kwargs(),					{'''a''': 2, '''b''': True}      )
					self.assertDictEqual(MockClass(a=2,					c=2.25      ).to_kwargs(),					{'''a''': 2, '''c''': 2.25}      )


	@require_cuda
	def 				lowercase__  (self : Dict      )     ->			str:

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						List[str]      					= GradScalerKwargs(init_scale=1024,					growth_factor=2      )
					AcceleratorState._reset_state()
					SCREAMING_SNAKE_CASE  :						Tuple      					= Accelerator(mixed_precision='''fp16''',					kwargs_handlers=[scaler_handler]      )
					print(accelerator.use_fpaa      )
					SCREAMING_SNAKE_CASE  :						str      					= accelerator.scaler

					# Check the kwargs have been applied
					self.assertEqual(scaler._init_scale,					1024.0      )
					self.assertEqual(scaler._growth_factor,					2.0      )

					# Check the other values are at the default
					self.assertEqual(scaler._backoff_factor,					0.5      )
					self.assertEqual(scaler._growth_interval,					2000      )
					self.assertEqual(scaler._enabled,					__UpperCAmelCase      )






	@require_multi_gpu
	def 				lowercase__  (self : Optional[Any]      )     ->			Dict:

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						str      					= ['''torchrun''', F'''--nproc_per_node={torch.cuda.device_count()}''', inspect.getfile(self.__class__      )]
					execute_subprocess_async(__UpperCAmelCase,					env=os.environ.copy()      )


if __name__ == "__main__":
	snake_case_			=     DistributedDataParallelKwargs(bucket_cap_mb=15, find_unused_parameters=True)
	snake_case_			=     Accelerator(kwargs_handlers=[ddp_scaler])
	snake_case_			=     torch.nn.Linear(100, 200)
	snake_case_			=     accelerator.prepare(model)

	# Check the values changed in kwargs
	snake_case_			=     """"""
	snake_case_			=     model.bucket_bytes_cap // (1024 * 1024)
	if observed_bucket_cap_map != 15:
		error_msg += f"Kwargs badly passed, should have `15` but found {observed_bucket_cap_map}.\n"
	if model.find_unused_parameters is not True:
		error_msg += f"Kwargs badly passed, should have `True` but found {model.find_unused_parameters}.\n"

	# Check the values of the defaults
	if model.dim != 0:
		error_msg += f"Default value not respected, should have `0` but found {model.dim}.\n"
	if model.broadcast_buffers is not True:
		error_msg += f"Default value not respected, should have `True` but found {model.broadcast_buffers}.\n"
	if model.gradient_as_bucket_view is not False:
		error_msg += f"Default value not respected, should have `False` but found {model.gradient_as_bucket_view}.\n"

	# Raise error at the end to make sure we don't stop at the first failure.
	if len(error_msg) > 0:
		raise ValueError(error_msg)




