






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      _a       (   self				):
         # A mock response for an HTTP head request to emulate server down
         UpperCamelCase_:   Any        =							mock.Mock()
         UpperCamelCase_:   Optional[int]        =							5_0_0
         UpperCamelCase_:   Optional[int]        =							{}
         UpperCamelCase_:   List[str]        =							HTTPError
         UpperCamelCase_:   str        =							{}

         # Download this model to make sure it's in the cache.
         UpperCamelCase_:   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=_lowerCamelCase				) as mock_head:
           UpperCamelCase_:   Dict        =							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
         UpperCamelCase_:   Tuple        =							mock.Mock()
         UpperCamelCase_:   Union[str, Any]        =							5_0_0
         UpperCamelCase_:   Optional[int]        =							{}
         UpperCamelCase_:   str        =							HTTPError
         UpperCamelCase_:   str        =							{}

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

           UpperCamelCase_:   Tuple        =							AlbertTokenizer.from_pretrained(_lowerCamelCase				)
         finally:
           os.remove(_lowerCamelCase				)

         # 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'    ,     _lowerCamelCase				)
           UpperCamelCase_:   Union[str, 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    ,     1_0_0_0				)
           # 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
         UpperCamelCase_:   str        =							AlbertTokenizer.from_pretrained('https://huggingface.co/albert-base-v1/resolve/main/spiece.model'				)







@is_staging_test
class 							_lowerCAmelCase(			unittest.TestCase   ):





       """simple docstring"""







       a							:						Dict										=['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']





       @classmethod
       def      _a       (   cls				):
         UpperCamelCase_:   Optional[int]        =							TOKEN
         HfFolder.save_token(_lowerCamelCase				)





       @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:
           UpperCamelCase_:   Union[str, Any]        =							os.path.join(_lowerCamelCase    ,     'vocab.txt'				)
           with open(_lowerCamelCase    ,     'w'    ,     encoding='utf-8'				) as vocab_writer:
             vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]				)				)
           UpperCamelCase_:   Union[str, Any]        =							BertTokenizer(_lowerCamelCase				)

         tokenizer.push_to_hub('test-tokenizer'    ,     use_auth_token=self._token				)
         UpperCamelCase_:   Union[str, 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(_lowerCamelCase    ,     repo_id='test-tokenizer'    ,     push_to_hub=_lowerCamelCase    ,     use_auth_token=self._token				)

         UpperCamelCase_:   Optional[Any]        =							BertTokenizer.from_pretrained(f'''{USER}/test-tokenizer'''				)
         self.assertDictEqual(new_tokenizer.vocab    ,     tokenizer.vocab				)





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

         tokenizer.push_to_hub('valid_org/test-tokenizer-org'    ,     use_auth_token=self._token				)
         UpperCamelCase_:   Optional[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(
               _lowerCamelCase    ,     repo_id='valid_org/test-tokenizer-org'    ,     push_to_hub=_lowerCamelCase    ,     use_auth_token=self._token				)

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

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

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

           UpperCamelCase_:   str        =							BertTokenizerFast.from_pretrained(_lowerCamelCase				)
           bert_tokenizer.save_pretrained(_lowerCamelCase				)
           UpperCamelCase_:   Any        =							CustomTokenizerFast.from_pretrained(_lowerCamelCase				)

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

         UpperCamelCase_:   Dict        =							AutoTokenizer.from_pretrained(f'''{USER}/test-dynamic-tokenizer'''    ,     trust_remote_code=_lowerCamelCase				)
         # 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'				)
         UpperCamelCase_:   str        =							AutoTokenizer.from_pretrained(
             f'''{USER}/test-dynamic-tokenizer'''    ,     use_fast=_lowerCamelCase    ,     trust_remote_code=_lowerCamelCase				)
         # 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      _a       (   self				):
         UpperCamelCase_:   str        =							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				):
         UpperCamelCase_:   Optional[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      _a       (   self				):
         UpperCamelCase_:   int        =							Trie()
         trie.add('A'				)
         self.assertEqual(trie.split('ABC'				)    ,     ['A', 'BC']				)
         self.assertEqual(trie.split('BCA'				)    ,     ['BC', 'A']				)





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





       def      _a       (   self				):
         UpperCamelCase_:   Dict        =							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.
         UpperCamelCase_:   Tuple        =							Trie()
         UpperCamelCase_:   Optional[int]        =							trie.cut_text('ABC'    ,     [0, 0, 2, 1, 2, 3]				)
         self.assertEqual(_lowerCamelCase    ,     ['AB', 'C']				)






'''simple docstring'''






import json
import os
from pathlib import Path

import pytest

from datasets.download.download_config import DownloadConfig
from datasets.download.download_manager import DownloadManager
from datasets.utils.file_utils import hash_url_to_filename


__UpperCamelCase					    =	"http://www.mocksite.com/file1.txt"
__UpperCamelCase					    =	"\"text\": [\"foo\", \"foo\"]"
__UpperCamelCase					    =	"6d8ce9aa78a471c7477201efbeabd3bb01ac2e7d100a6dc024ba1608361f90a8"




class  _A :
		lowercase__:   str				     =   200
		lowercase__:   List[str]				     =   {'''Content-Length''': '''100'''}
		lowercase__:   Union[str, Any]				     =   {}

		def 	lowercase__     (   self					:   Any  ,      **__magic_name__					:   List[Any]			)					->    Dict:


					"""simple docstring"""

					return [bytes(__magic_name__  ,      """utf-8"""			)]







def 					_a						(       *_lowerCamelCase    ,	**_lowerCamelCase ) ->				List[str]:



			"""simple docstring"""


			return MockResponse()







@pytest.mark.parametrize("""urls_type"""    ,	[str, list, dict] )
def 					_a						(       _lowerCamelCase    ,	_lowerCamelCase    ,	_lowerCamelCase ) ->				List[str]:



			"""simple docstring"""


			import requests

			monkeypatch.setattr(_lowerCamelCase    ,	"""request"""    ,	_lowerCamelCase )

			__snake_case					:      Union[str, Any]							      =     URL
			if issubclass(_lowerCamelCase    ,	_lowerCamelCase ):
						__snake_case					:      str							      =     url
			elif issubclass(_lowerCamelCase    ,	_lowerCamelCase ):
						__snake_case					:      Dict							      =     [url]
			elif issubclass(_lowerCamelCase    ,	_lowerCamelCase ):
						__snake_case					:      Union[str, Any]							      =     {"""train""": url}
			__snake_case					:      Dict							      =     """dummy"""
			__snake_case					:      List[str]							      =     """downloads"""
			__snake_case					:      List[Any]							      =     tmp_path
			__snake_case					:      List[Any]							      =     DownloadConfig(
			    cache_dir=os.path.join(_lowerCamelCase    ,	_lowerCamelCase )    ,	use_etag=_lowerCamelCase    ,	)
			__snake_case					:      List[str]							      =     DownloadManager(dataset_name=_lowerCamelCase    ,	download_config=_lowerCamelCase )
			__snake_case					:      int							      =     dl_manager.download(_lowerCamelCase )
			__snake_case					:      Tuple							      =     urls
			for downloaded_paths in [downloaded_paths]:
						if isinstance(_lowerCamelCase    ,	_lowerCamelCase ):
									__snake_case					:      Any							      =     [downloaded_paths]
									__snake_case					:      List[Any]							      =     [urls]
						elif isinstance(_lowerCamelCase    ,	_lowerCamelCase ):
									assert "train" in downloaded_paths.keys()
									__snake_case					:      Tuple							      =     downloaded_paths.values()
									__snake_case					:      Optional[int]							      =     urls.values()
						assert downloaded_paths
						for downloaded_path, input_url in zip(_lowerCamelCase    ,	_lowerCamelCase ):
									assert downloaded_path == dl_manager.downloaded_paths[input_url]
									__snake_case					:      List[str]							      =     Path(_lowerCamelCase )
									__snake_case					:      Any							      =     downloaded_path.parts
									assert parts[-1] == HASH
									assert parts[-2] == cache_subdir
									assert downloaded_path.exists()
									__snake_case					:      Union[str, Any]							      =     downloaded_path.read_text()
									assert content == CONTENT
									__snake_case					:      List[str]							      =     downloaded_path.with_suffix(""".json""" )
									assert metadata_downloaded_path.exists()
									__snake_case					:      Union[str, Any]							      =     json.loads(metadata_downloaded_path.read_text() )
									assert metadata_content == {"url": URL, "etag": None}







@pytest.mark.parametrize("""paths_type"""    ,	[str, list, dict] )
def 					_a						(       _lowerCamelCase    ,	_lowerCamelCase    ,	_lowerCamelCase ) ->				List[Any]:



			"""simple docstring"""


			__snake_case					:      Any							      =     str(_lowerCamelCase )
			if issubclass(_lowerCamelCase    ,	_lowerCamelCase ):
						__snake_case					:      Optional[int]							      =     filename
			elif issubclass(_lowerCamelCase    ,	_lowerCamelCase ):
						__snake_case					:      Tuple							      =     [filename]
			elif issubclass(_lowerCamelCase    ,	_lowerCamelCase ):
						__snake_case					:      Dict							      =     {"""train""": filename}
			__snake_case					:      Optional[Any]							      =     """dummy"""
			__snake_case					:      List[Any]							      =     xz_file.parent
			__snake_case					:      int							      =     """extracted"""
			__snake_case					:      Dict							      =     DownloadConfig(
			    cache_dir=_lowerCamelCase    ,	use_etag=_lowerCamelCase    ,	)
			__snake_case					:      List[str]							      =     DownloadManager(dataset_name=_lowerCamelCase    ,	download_config=_lowerCamelCase )
			__snake_case					:      Optional[Any]							      =     dl_manager.extract(_lowerCamelCase )
			__snake_case					:      Union[str, Any]							      =     paths
			for extracted_paths in [extracted_paths]:
						if isinstance(_lowerCamelCase    ,	_lowerCamelCase ):
									__snake_case					:      Dict							      =     [extracted_paths]
									__snake_case					:      int							      =     [paths]
						elif isinstance(_lowerCamelCase    ,	_lowerCamelCase ):
									assert "train" in extracted_paths.keys()
									__snake_case					:      int							      =     extracted_paths.values()
									__snake_case					:      int							      =     paths.values()
						assert extracted_paths
						for extracted_path, input_path in zip(_lowerCamelCase    ,	_lowerCamelCase ):
									assert extracted_path == dl_manager.extracted_paths[input_path]
									__snake_case					:      Any							      =     Path(_lowerCamelCase )
									__snake_case					:      str							      =     extracted_path.parts
									assert parts[-1] == hash_url_to_filename(_lowerCamelCase    ,	etag=_lowerCamelCase )
									assert parts[-2] == extracted_subdir
									assert extracted_path.exists()
									__snake_case					:      Optional[int]							      =     extracted_path.read_text()
									__snake_case					:      str							      =     text_file.read_text()
									assert extracted_file_content == expected_file_content







def 					_a						(       _lowerCamelCase    ,	_lowerCamelCase ) ->				Optional[int]:



			"""simple docstring"""


			assert path.endswith(""".jsonl""" )
			for num_items, line in enumerate(_lowerCamelCase    ,	start=1 ):
						__snake_case					:      Tuple							      =     json.loads(line.decode("""utf-8""" ) )
						assert item.keys() == {"col_1", "col_2", "col_3"}
			assert num_items == 4







@pytest.mark.parametrize("""archive_jsonl"""    ,	["""tar_jsonl_path""", """zip_jsonl_path"""] )
def 					_a						(       _lowerCamelCase    ,	_lowerCamelCase ) ->				Optional[int]:



			"""simple docstring"""


			__snake_case					:      Any							      =     request.getfixturevalue(_lowerCamelCase )
			__snake_case					:      str							      =     DownloadManager()
			for num_jsonl, (path, file) in enumerate(dl_manager.iter_archive(_lowerCamelCase )    ,	start=1 ):
						_test_jsonl(_lowerCamelCase    ,	_lowerCamelCase )
			assert num_jsonl == 2







@pytest.mark.parametrize("""archive_nested_jsonl"""    ,	["""tar_nested_jsonl_path""", """zip_nested_jsonl_path"""] )
def 					_a						(       _lowerCamelCase    ,	_lowerCamelCase ) ->				List[str]:



			"""simple docstring"""


			__snake_case					:      int							      =     request.getfixturevalue(_lowerCamelCase )
			__snake_case					:      List[str]							      =     DownloadManager()
			for num_tar, (path, file) in enumerate(dl_manager.iter_archive(_lowerCamelCase )    ,	start=1 ):
						for num_jsonl, (subpath, subfile) in enumerate(dl_manager.iter_archive(_lowerCamelCase )    ,	start=1 ):
									_test_jsonl(_lowerCamelCase    ,	_lowerCamelCase )
			assert num_tar == 1
			assert num_jsonl == 2







def 					_a						(       _lowerCamelCase ) ->				Any:



			"""simple docstring"""


			__snake_case					:      List[str]							      =     DownloadManager()
			for num_file, file in enumerate(dl_manager.iter_files(_lowerCamelCase )    ,	start=1 ):
						assert os.path.basename(_lowerCamelCase ) == ("test.txt" if num_file == 1 else "train.txt")
			assert num_file == 2


