





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

  """simple docstring"""


  def 				lowercase		(				self :		int     )							->						str:
     # A mock response for an HTTP head request to emulate server down
     __lowerCAmelCase								=	mock.Mock()
     __lowerCAmelCase								=	5_0_0
     __lowerCAmelCase								=	{}
     __lowerCAmelCase								=	HTTPError
     __lowerCAmelCase								=	{}

     # Download this model to make sure it's in the cache.
     __lowerCAmelCase								=	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:
        __lowerCAmelCase								=	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 :		Any     )							->						Optional[Any]:
     # A mock response for an HTTP head request to emulate server down
     __lowerCAmelCase								=	mock.Mock()
     __lowerCAmelCase								=	5_0_0
     __lowerCAmelCase								=	{}
     __lowerCAmelCase								=	HTTPError
     __lowerCAmelCase								=	{}

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






  def 				lowercase		(				self :		Dict     )							->						int:
     # This test is for deprecated behavior and can be removed in v5
     try:
        __lowerCAmelCase								=	tempfile.mktemp()
        with open(lowerCAmelCase_					,  'wb'     ) as f:
           http_get('https://huggingface.co/albert-base-v1/resolve/main/spiece.model'					,  lowerCAmelCase_     )

        __lowerCAmelCase								=	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_     )
        __lowerCAmelCase								=	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 				lowercase		(				self :		Union[str, Any]     )							->						Optional[Any]:
     # This test is for deprecated behavior and can be removed in v5
     __lowerCAmelCase								=	AlbertTokenizer.from_pretrained('https://huggingface.co/albert-base-v1/resolve/main/spiece.model'     )





@is_staging_test
class 				_UpperCAmelCase    (    unittest.TestCase    ):

  """simple docstring"""


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






  @classmethod
  def 				lowercase		(				cls :		Optional[int]     )							->						Dict:
     __lowerCAmelCase								=	TOKEN
     HfFolder.save_token(lowerCAmelCase_     )






  @classmethod
  def 				lowercase		(				cls :		Optional[Any]     )							->						Union[str, Any]:
     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 :		Tuple     )							->						Union[str, Any]:
     with tempfile.TemporaryDirectory() as tmp_dir:
        __lowerCAmelCase								=	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]     )     )
        __lowerCAmelCase								=	BertTokenizer(lowerCAmelCase_     )

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

     __lowerCAmelCase								=	BertTokenizer.from_pretrained(f"""{USER}/test-tokenizer"""     )
     self.assertDictEqual(new_tokenizer.vocab					,  tokenizer.vocab     )






  def 				lowercase		(				self :		List[Any]     )							->						Tuple:
     with tempfile.TemporaryDirectory() as tmp_dir:
        __lowerCAmelCase								=	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]     )     )
        __lowerCAmelCase								=	BertTokenizer(lowerCAmelCase_     )

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

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






  @require_tokenizers
  def 				lowercase		(				self :		Any     )							->						Union[str, Any]:
     CustomTokenizer.register_for_auto_class()
     with tempfile.TemporaryDirectory() as tmp_dir:
        __lowerCAmelCase								=	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]     )     )
        __lowerCAmelCase								=	CustomTokenizer(lowerCAmelCase_     )

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

     __lowerCAmelCase								=	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:
        __lowerCAmelCase								=	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]     )     )

        __lowerCAmelCase								=	BertTokenizerFast.from_pretrained(lowerCAmelCase_     )
        bert_tokenizer.save_pretrained(lowerCAmelCase_     )
        __lowerCAmelCase								=	CustomTokenizerFast.from_pretrained(lowerCAmelCase_     )

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

     __lowerCAmelCase								=	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'     )
     __lowerCAmelCase								=	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 				_UpperCAmelCase    (    unittest.TestCase    ):

  """simple docstring"""


  def 				lowercase		(				self :		Union[str, Any]     )							->						Optional[Any]:
     __lowerCAmelCase								=	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]     )							->						Dict:
     __lowerCAmelCase								=	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 :		Optional[int]     )							->						List[str]:
     __lowerCAmelCase								=	Trie()
     trie.add('A'     )
     self.assertEqual(trie.split('ABC'     )					,  ['A', 'BC']     )
     self.assertEqual(trie.split('BCA'     )					,  ['BC', 'A']     )






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






  def 				lowercase		(				self :		int     )							->						List[Any]:
     __lowerCAmelCase								=	Trie()
     trie.add('ABC'     )
     trie.add('B'     )
     trie.add('CD'     )
     self.assertEqual(trie.split('ABCD'     )					,  ['ABC', 'D']     )






  def 				lowercase		(				self :		str     )							->						str:
     # Even if the offsets are wrong, we necessarily output correct string
     # parts.
     __lowerCAmelCase								=	Trie()
     __lowerCAmelCase								=	trie.cut_text('ABC'					,  [0, 0, 2, 1, 2, 3]     )
     self.assertEqual(lowerCAmelCase_					,  ['AB', 'C']     )








import pytest

from datasets.utils.sharding import _distribute_shards, _number_of_shards_in_gen_kwargs, _split_gen_kwargs

@pytest.mark.parametrize(
    'kwargs, expected',	[
        ({'num_shards': 0, 'max_num_jobs': 1}, []),
        ({'num_shards': 10, 'max_num_jobs': 1}, [range(10	)]),
        ({'num_shards': 10, 'max_num_jobs': 10}, [range(lowerCAmelCase_,	i + 1	) for i in range(10	)]),
        ({'num_shards': 1, 'max_num_jobs': 10}, [range(1	)]),
        ({'num_shards': 10, 'max_num_jobs': 3}, [range(0,	4	), range(4,	7	), range(7,	10	)]),
        ({'num_shards': 3, 'max_num_jobs': 10}, [range(0,	1	), range(1,	2	), range(2,	3	)]),
    ],	)
def   a_    ( lowerCAmelCase_      :			Optional[Any],	lowerCAmelCase_      :			Any	):
   __lowerCAmelCase								=	_distribute_shards(**lowerCAmelCase_	)
   assert out == expected

@pytest.mark.parametrize(
    'gen_kwargs, max_num_jobs, expected',	[
        ({'foo': 0}, 10, [{'foo': 0}]),
        ({'shards': [0, 1, 2, 3]}, 1, [{'shards': [0, 1, 2, 3]}]),
        ({'shards': [0, 1, 2, 3]}, 4, [{'shards': [0]}, {'shards': [1]}, {'shards': [2]}, {'shards': [3]}]),
        ({'shards': [0, 1]}, 4, [{'shards': [0]}, {'shards': [1]}]),
        ({'shards': [0, 1, 2, 3]}, 2, [{'shards': [0, 1]}, {'shards': [2, 3]}]),
    ],	)
def   a_    ( lowerCAmelCase_      :			Optional[Any],	lowerCAmelCase_      :			List[str],	lowerCAmelCase_      :			Optional[int]	):
   __lowerCAmelCase								=	_split_gen_kwargs(lowerCAmelCase_,	lowerCAmelCase_	)
   assert out == expected




@pytest.mark.parametrize(
    'gen_kwargs, expected',	[
        ({'foo': 0}, 1),
        ({'shards': [0]}, 1),
        ({'shards': [0, 1, 2, 3]}, 4),
        ({'shards': [0, 1, 2, 3], 'foo': 0}, 4),
        ({'shards': [0, 1, 2, 3], 'other': (0, 1)}, 4),
        ({'shards': [0, 1, 2, 3], 'shards2': [0, 1]}, RuntimeError),
    ],	)
def   a_    ( lowerCAmelCase_      :			int,	lowerCAmelCase_      :			Any	):
   if expected is RuntimeError:
      with pytest.raises(lowerCAmelCase_	):
         _number_of_shards_in_gen_kwargs(lowerCAmelCase_	)
   else:
      __lowerCAmelCase								=	_number_of_shards_in_gen_kwargs(lowerCAmelCase_	)
      assert out == expected

