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







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







											'''simple docstring'''






											UpperCamelCase   			=     mock.Mock()
											UpperCamelCase   			=     500
											UpperCamelCase   			=     {}
											UpperCamelCase   			=     HTTPError
											UpperCamelCase   			=     {}

											# Download this model to make sure it's in the cache.
											UpperCamelCase   			=     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:
																		UpperCamelCase   			=     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 						UpperCAmelCase_    (       self     )->		int:







											'''simple docstring'''






											UpperCamelCase   			=     mock.Mock()
											UpperCamelCase   			=     500
											UpperCamelCase   			=     {}
											UpperCamelCase   			=     HTTPError
											UpperCamelCase   			=     {}

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







				def 						UpperCAmelCase_    (       self     )->		Tuple:







											'''simple docstring'''






											try:
																		UpperCamelCase   			=     tempfile.mktemp()
																		with open(A_						,				'wb'     ) as f:
																									http_get('https://huggingface.co/albert-base-v1/resolve/main/spiece.model'						,				A_     )

																		UpperCamelCase   			=     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_     )
																		UpperCamelCase   			=     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 						UpperCAmelCase_    (       self     )->		Any:







											'''simple docstring'''






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


@is_staging_test
class 			SCREAMING_SNAKE_CASE__      (     unittest.TestCase):
				lowerCAmelCase_     			=							["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]







				@classmethod
				def 						UpperCAmelCase_    (       cls     )->		int:







											'''simple docstring'''






											UpperCamelCase   			=     TOKEN
											HfFolder.save_token(A_     )







				@classmethod
				def 						UpperCAmelCase_    (       cls     )->		Optional[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 						UpperCAmelCase_    (       self     )->		List[str]:







											'''simple docstring'''






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

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

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







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







											'''simple docstring'''






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

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

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







				@require_tokenizers
				def 						UpperCAmelCase_    (       self     )->		Optional[int]:







											'''simple docstring'''






											CustomTokenizer.register_for_auto_class()
											with tempfile.TemporaryDirectory() as tmp_dir:
																		UpperCamelCase   			=     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]     )     )
																		UpperCamelCase   			=     CustomTokenizer(A_     )

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

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

																		UpperCamelCase   			=     BertTokenizerFast.from_pretrained(A_     )
																		bert_tokenizer.save_pretrained(A_     )
																		UpperCamelCase   			=     CustomTokenizerFast.from_pretrained(A_     )

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

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







				def 						UpperCAmelCase_    (       self     )->		Dict:







											'''simple docstring'''






											UpperCamelCase   			=     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 						UpperCAmelCase_    (       self     )->		List[Any]:







											'''simple docstring'''






											UpperCamelCase   			=     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 						UpperCAmelCase_    (       self     )->		List[Any]:







											'''simple docstring'''






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







				def 						UpperCAmelCase_    (       self     )->		Optional[Any]:







											'''simple docstring'''






											UpperCamelCase   			=     Trie()
											trie.add('TOKEN]'     )
											trie.add('[SPECIAL_TOKEN]'     )
											self.assertEqual(trie.split('This is something [SPECIAL_TOKEN]'     )						,				['This is something ', '[SPECIAL_TOKEN]']     )







				def 						UpperCAmelCase_    (       self     )->		Any:







											'''simple docstring'''






											UpperCamelCase   			=     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 						UpperCAmelCase_    (       self     )->		List[str]:







											'''simple docstring'''






											UpperCamelCase   			=     Trie()
											trie.add('AB'     )
											trie.add('B'     )
											trie.add('C'     )
											self.assertEqual(trie.split('ABC'     )						,				['AB', 'C']     )







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







											'''simple docstring'''






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







				def 						UpperCAmelCase_    (       self     )->		Dict:







											'''simple docstring'''






											UpperCamelCase   			=     Trie()
											UpperCamelCase   			=     trie.cut_text('ABC'						,				[0, 0, 2, 1, 2, 3]     )
											self.assertEqual(A_						,				['AB', 'C']     )










import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized, parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
		from sagemaker import Session, TrainingJobAnalytics
		from sagemaker.huggingface import HuggingFace
@pytest.mark.skipif(
    literal_eval(os.getenv('''TEST_SAGEMAKER'''			,  '''False'''     )     ) is not True			,  reason='''Skipping test because should only be run when releasing minor transformers version'''			,  )
@pytest.mark.usefixtures('''sm_env'''     )
@parameterized_class(
    [
        {
            '''framework''': '''pytorch''',
            '''script''': '''run_glue.py''',
            '''model_name_or_path''': '''distilbert-base-cased''',
            '''instance_type''': '''ml.p3.16xlarge''',
            '''results''': {'''train_runtime''': 650, '''eval_accuracy''': 0.7, '''eval_loss''': 0.6},
        },
        {
            '''framework''': '''pytorch''',
            '''script''': '''run_ddp.py''',
            '''model_name_or_path''': '''distilbert-base-cased''',
            '''instance_type''': '''ml.p3.16xlarge''',
            '''results''': {'''train_runtime''': 600, '''eval_accuracy''': 0.7, '''eval_loss''': 0.6},
        },
        {
            '''framework''': '''tensorflow''',
            '''script''': '''run_tf_dist.py''',
            '''model_name_or_path''': '''distilbert-base-cased''',
            '''instance_type''': '''ml.p3.16xlarge''',
            '''results''': {'''train_runtime''': 600, '''eval_accuracy''': 0.6, '''eval_loss''': 0.7},
        },
    ]     )
class 						UpperCAmelCase  (							unittest.TestCase     ):







						'''simple docstring'''





						def 			snake_case__		(		self				:	Optional[Any]		):






							"""simple docstring"""






							if self.framework == "pytorch":
								subprocess.run(
								    f"cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py".split()			,      encoding="utf-8"			,      check=__lowercase			,      )
							assert hasattr(self			,      "env"		)






						def 			snake_case__		(		self				:	Dict			,      __lowercase				:	Optional[Any]		):






							"""simple docstring"""






							snake_case_						       =	f"{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}"
							# distributed data settings
							snake_case_						       =	{"smdistributed": {"dataparallel": {"enabled": True}}} if self.script != "run_ddp.py" else None

							# creates estimator
							return HuggingFace(
							    entry_point=self.script			,      source_dir=self.env.test_path			,      role=self.env.role			,      image_uri=self.env.image_uri			,      base_job_name=__lowercase			,      instance_count=__lowercase			,      instance_type=self.instance_type			,      debugger_hook_config=__lowercase			,      hyperparameters={**self.env.distributed_hyperparameters, "model_name_or_path": self.model_name_or_path}			,      metric_definitions=self.env.metric_definitions			,      distribution=__lowercase			,      py_version="py36"			,      )






						def 			snake_case__		(		self				:	Union[str, Any]			,      __lowercase				:	Tuple		):






							"""simple docstring"""






							TrainingJobAnalytics(__lowercase		).export_csv(f"{self.env.test_path}/{job_name}_metrics.csv"		)






						@parameterized.expand([(2,)]		)
						def 			snake_case__		(		self				:	Union[str, Any]			,      __lowercase				:	Union[str, Any]		):






							"""simple docstring"""






							snake_case_						       =	self.create_estimator(__lowercase		)

							# run training
							estimator.fit()

							# result dataframe
							snake_case_						       =	TrainingJobAnalytics(estimator.latest_training_job.name		).dataframe()

							# extract kpis
							snake_case_						       =	list(result_metrics_df[result_metrics_df.metric_name == "eval_accuracy"]["value"]		)
							snake_case_						       =	list(result_metrics_df[result_metrics_df.metric_name == "eval_loss"]["value"]		)
							# get train time from SageMaker job, this includes starting, preprocessing, stopping
							snake_case_						       =	(
							    Session().describe_training_job(estimator.latest_training_job.name		).get("TrainingTimeInSeconds"			,      99_99_99		)
							)

							# assert kpis
							assert train_runtime <= self.results["train_runtime"]
							assert all(t >= self.results["eval_accuracy"] for t in eval_accuracy		)
							assert all(t <= self.results["eval_loss"] for t in eval_loss		)

							# dump tests result into json file to share in PR
							with open(f"{estimator.latest_training_job.name}.json"			,      "w"		) as outfile:
								json.dump({"train_time": train_runtime, "eval_accuracy": eval_accuracy, "eval_loss": eval_loss}			,      __lowercase		)
