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


						def  __A ( self			:       Dict		):





												'''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  __A ( self			:       str		):





												'''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  __A ( self			:       Any		):





												'''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  __A ( self			:       Tuple		):





												'''simple docstring'''





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






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


						@classmethod
						def  __A ( cls			:       Optional[int]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Tuple           =							TOKEN
												HfFolder.save_token(a_		)


						@classmethod
						def  __A ( cls			:       Any		):





												'''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  __A ( self			:       int		):





												'''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  __A ( self			:       str		):





												'''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  __A ( self			:       Any		):





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


						def  __A ( self			:       List[str]		):





												'''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  __A ( self			:       Union[str, 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  __A ( self			:       Dict		):





												'''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  __A ( self			:       str		):





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





												'''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  __A ( self			:       Dict		):





												'''simple docstring'''





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



						def  __A ( self			:       Union[str, Any]		):





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






import argparse
import json
import logging
import os
import sys
from unittest.mock import patch

from transformers.testing_utils import TestCasePlus, get_gpu_count, slow


__UpperCamelCase	: str    		=  [
    os.path.join(os.path.dirname(__file__), dirname)
    for dirname in [
        'text-classification',
        'language-modeling',
        'summarization',
        'token-classification',
        'question-answering',
    ]
]
sys.path.extend(SRC_DIRS)


if SRC_DIRS is not None:
					import run_clm_flax
					import run_flax_glue
					import run_flax_ner
					import run_mlm_flax
					import run_qa
					import run_summarization_flax
					import run_ta_mlm_flax


logging.basicConfig(level=logging.DEBUG)

__UpperCamelCase	: int    		=  logging.getLogger()
def    A			(					):
						SCREAMING_SNAKE_CASE							:					str           =							argparse.ArgumentParser()
						parser.add_argument('''-f'''		)
						SCREAMING_SNAKE_CASE							:					List[str]           =							parser.parse_args()
						return args.f
def    A			(					_lowercase	,						_lowercase="eval"		):
						SCREAMING_SNAKE_CASE							:					Dict           =							os.path.join(_lowercase	,						f"""{split}_results.json"""		)
						if os.path.exists(_lowercase		):
												with open(_lowercase	,						'''r'''		) as f:
																		return json.load(_lowercase		)
						raise ValueError(f"""can't find {path}"""		)


__UpperCamelCase	: Optional[Any]    		=  logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)





class       lowercase__		(     UpperCamelCase_):


						def  __A ( self			:       Optional[int]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					int           =							self.get_auto_remove_tmp_dir()
												SCREAMING_SNAKE_CASE							:					Tuple           =							f"""
            run_glue.py
            --model_name_or_path distilbert-base-uncased
            --output_dir {tmp_dir}
            --train_file ./tests/fixtures/tests_samples/MRPC/train.csv
            --validation_file ./tests/fixtures/tests_samples/MRPC/dev.csv
            --per_device_train_batch_size=2
            --per_device_eval_batch_size=1
            --learning_rate=1e-4
            --eval_steps=2
            --warmup_steps=2
            --seed=42
            --max_seq_length=128
            """.split()

												with patch.object(UpperCamelCase__					,	'''argv'''					,	UpperCamelCase__		):
																		run_flax_glue.main()
																		SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							get_results(UpperCamelCase__		)
																		self.assertGreaterEqual(result['''eval_accuracy''']					,	0.75		)


						@slow
						def  __A ( self			:       Optional[int]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Dict           =							self.get_auto_remove_tmp_dir()
												SCREAMING_SNAKE_CASE							:					str           =							f"""
            run_clm_flax.py
            --model_name_or_path distilgpt2
            --train_file ./tests/fixtures/sample_text.txt
            --validation_file ./tests/fixtures/sample_text.txt
            --do_train
            --do_eval
            --block_size 128
            --per_device_train_batch_size 4
            --per_device_eval_batch_size 4
            --num_train_epochs 2
            --logging_steps 2 --eval_steps 2
            --output_dir {tmp_dir}
            --overwrite_output_dir
            """.split()

												with patch.object(UpperCamelCase__					,	'''argv'''					,	UpperCamelCase__		):
																		run_clm_flax.main()
																		SCREAMING_SNAKE_CASE							:					Dict           =							get_results(UpperCamelCase__		)
																		self.assertLess(result['''eval_perplexity''']					,	100		)


						@slow
						def  __A ( self			:       Any		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					List[str]           =							self.get_auto_remove_tmp_dir()
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							f"""
            run_summarization.py
            --model_name_or_path t5-small
            --train_file tests/fixtures/tests_samples/xsum/sample.json
            --validation_file tests/fixtures/tests_samples/xsum/sample.json
            --test_file tests/fixtures/tests_samples/xsum/sample.json
            --output_dir {tmp_dir}
            --overwrite_output_dir
            --num_train_epochs=3
            --warmup_steps=8
            --do_train
            --do_eval
            --do_predict
            --learning_rate=2e-4
            --per_device_train_batch_size=2
            --per_device_eval_batch_size=1
            --predict_with_generate
        """.split()

												with patch.object(UpperCamelCase__					,	'''argv'''					,	UpperCamelCase__		):
																		run_summarization_flax.main()
																		SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							get_results(UpperCamelCase__					,	split='''test'''		)
																		self.assertGreaterEqual(result['''test_rouge1''']					,	10		)
																		self.assertGreaterEqual(result['''test_rouge2''']					,	2		)
																		self.assertGreaterEqual(result['''test_rougeL''']					,	7		)
																		self.assertGreaterEqual(result['''test_rougeLsum''']					,	7		)


						@slow
						def  __A ( self			:       Optional[int]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					List[Any]           =							self.get_auto_remove_tmp_dir()
												SCREAMING_SNAKE_CASE							:					Dict           =							f"""
            run_mlm.py
            --model_name_or_path distilroberta-base
            --train_file ./tests/fixtures/sample_text.txt
            --validation_file ./tests/fixtures/sample_text.txt
            --output_dir {tmp_dir}
            --overwrite_output_dir
            --max_seq_length 128
            --per_device_train_batch_size 4
            --per_device_eval_batch_size 4
            --logging_steps 2 --eval_steps 2
            --do_train
            --do_eval
            --num_train_epochs=1
        """.split()

												with patch.object(UpperCamelCase__					,	'''argv'''					,	UpperCamelCase__		):
																		run_mlm_flax.main()
																		SCREAMING_SNAKE_CASE							:					List[Any]           =							get_results(UpperCamelCase__		)
																		self.assertLess(result['''eval_perplexity''']					,	42		)


						@slow
						def  __A ( self			:       List[Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Dict           =							self.get_auto_remove_tmp_dir()
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							f"""
            run_t5_mlm_flax.py
            --model_name_or_path t5-small
            --train_file ./tests/fixtures/sample_text.txt
            --validation_file ./tests/fixtures/sample_text.txt
            --do_train
            --do_eval
            --max_seq_length 128
            --per_device_train_batch_size 4
            --per_device_eval_batch_size 4
            --num_train_epochs 2
            --logging_steps 2 --eval_steps 2
            --output_dir {tmp_dir}
            --overwrite_output_dir
            """.split()

												with patch.object(UpperCamelCase__					,	'''argv'''					,	UpperCamelCase__		):
																		run_ta_mlm_flax.main()
																		SCREAMING_SNAKE_CASE							:					Optional[int]           =							get_results(UpperCamelCase__		)
																		self.assertGreaterEqual(result['''eval_accuracy''']					,	0.42		)


						@slow
						def  __A ( self			:       Union[str, Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Tuple           =							7 if get_gpu_count() > 1 else 2

												SCREAMING_SNAKE_CASE							:					Dict           =							self.get_auto_remove_tmp_dir()
												SCREAMING_SNAKE_CASE							:					Any           =							f"""
            run_flax_ner.py
            --model_name_or_path bert-base-uncased
            --train_file tests/fixtures/tests_samples/conll/sample.json
            --validation_file tests/fixtures/tests_samples/conll/sample.json
            --output_dir {tmp_dir}
            --overwrite_output_dir
            --do_train
            --do_eval
            --warmup_steps=2
            --learning_rate=2e-4
            --logging_steps 2 --eval_steps 2
            --per_device_train_batch_size=2
            --per_device_eval_batch_size=2
            --num_train_epochs={epochs}
            --seed 7
        """.split()

												with patch.object(UpperCamelCase__					,	'''argv'''					,	UpperCamelCase__		):
																		run_flax_ner.main()
																		SCREAMING_SNAKE_CASE							:					List[str]           =							get_results(UpperCamelCase__		)
																		self.assertGreaterEqual(result['''eval_accuracy''']					,	0.75		)
																		self.assertGreaterEqual(result['''eval_f1''']					,	0.3		)



						@slow
						def  __A ( self			:       List[str]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					List[str]           =							self.get_auto_remove_tmp_dir()
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							f"""
            run_qa.py
            --model_name_or_path bert-base-uncased
            --version_2_with_negative
            --train_file tests/fixtures/tests_samples/SQUAD/sample.json
            --validation_file tests/fixtures/tests_samples/SQUAD/sample.json
            --output_dir {tmp_dir}
            --overwrite_output_dir
            --num_train_epochs=3
            --warmup_steps=2
            --do_train
            --do_eval
            --logging_steps 2 --eval_steps 2
            --learning_rate=2e-4
            --per_device_train_batch_size=2
            --per_device_eval_batch_size=1
        """.split()

												with patch.object(UpperCamelCase__					,	'''argv'''					,	UpperCamelCase__		):
																		run_qa.main()
																		SCREAMING_SNAKE_CASE							:					str           =							get_results(UpperCamelCase__		)
																		self.assertGreaterEqual(result['''eval_f1''']					,	30		)
																		self.assertGreaterEqual(result['''eval_exact''']					,	30		)





