



import unittest
from pathlib import Path
from tempfile import TemporaryDirectory

from transformers import AutoConfig, TFAutoModel, is_tensorflow_text_available, is_tf_available
from transformers.models.bert.tokenization_bert import BertTokenizer
from transformers.testing_utils import require_tensorflow_text, require_tf, slow


if is_tf_available():
				import tensorflow as tf

if is_tensorflow_text_available():
				from transformers.models.bert import TFBertTokenizer


lowercase_	       =			['''bert-base-uncased''', '''bert-base-cased''']
lowercase_	       =			'''hf-internal-testing/tiny-bert-tf-only'''

if is_tf_available():






				class 						__a (     tf.keras.Model				):



					def __init__(						self							:    str				,	snake_case_							:    Dict)->							Any:
										super().__init__()
										__lowerCAmelCase	   =tokenizer
										__lowerCAmelCase	   =AutoConfig.from_pretrained(snake_case_)
										__lowerCAmelCase	   =TFAutoModel.from_config(snake_case_)



					def 		UpperCamelCase							(						self							:    Any				,	snake_case_							:    Any)->							List[Any]:
										__lowerCAmelCase	   =self.tokenizer(snake_case_)
										__lowerCAmelCase	   =self.bert(**snake_case_)
										return out["pooler_output"]






@require_tf
@require_tensorflow_text
class 						__a (     unittest.TestCase				):



	def 		UpperCamelCase							(						self							:    List[str])->							List[Any]:
						super().setUp()

						__lowerCAmelCase	   =[
						    BertTokenizer.from_pretrained(snake_case_) for checkpoint in (TOKENIZER_CHECKPOINTS * 2)
						]  # repeat for when fast_bert_tokenizer=false
						__lowerCAmelCase	   =[TFBertTokenizer.from_pretrained(snake_case_) for checkpoint in TOKENIZER_CHECKPOINTS] + [
						    TFBertTokenizer.from_pretrained(snake_case_				,	use_fast_bert_tokenizer=snake_case_)
						    for checkpoint in TOKENIZER_CHECKPOINTS
						]
						assert len(self.tokenizers) == len(self.tf_tokenizers)

						__lowerCAmelCase	   =[
						    """This is a straightforward English test sentence.""",
						    """This one has some weird characters\rto\nsee\r\nif  those\u00E9break things.""",
						    """Now we're going to add some Chinese: 一 二 三 一二三""",
						    """And some much more rare Chinese: 齉 堃 齉堃""",
						    """Je vais aussi écrire en français pour tester les accents""",
						    """Classical Irish also has some unusual characters, so in they go: Gaelaċ, ꝼ""",
						]
						__lowerCAmelCase	   =list(zip(self.test_sentences				,	self.test_sentences[::-1]))



	def 		UpperCamelCase							(						self							:    str)->							Optional[Any]:
						for tokenizer, tf_tokenizer in zip(self.tokenizers				,	self.tf_tokenizers):
											for test_inputs in (self.test_sentences, self.paired_sentences):
																__lowerCAmelCase	   =tokenizer(snake_case_				,	return_tensors="""tf"""				,	padding="""longest""")
																__lowerCAmelCase	   =tf_tokenizer(snake_case_)

																for key in python_outputs.keys():
																					self.assertTrue(tf.reduce_all(python_outputs[key].shape == tf_outputs[key].shape))
																					self.assertTrue(tf.reduce_all(tf.cast(python_outputs[key]				,	tf.intaa) == tf_outputs[key]))



	@slow
	def 		UpperCamelCase							(						self							:    Tuple)->							str:
						for tf_tokenizer in self.tf_tokenizers:
											__lowerCAmelCase	   =tf_tokenizer(self.paired_sentences)
											__lowerCAmelCase	   =tf_tokenizer(
											    text=[sentence[0] for sentence in self.paired_sentences]				,	text_pair=[sentence[1] for sentence in self.paired_sentences]				,	)
											for key in merged_outputs.keys():
																self.assertTrue(tf.reduce_all(tf.cast(merged_outputs[key]				,	tf.intaa) == separated_outputs[key]))



	@slow
	def 		UpperCamelCase							(						self							:    List[Any])->							Optional[int]:
						for tf_tokenizer in self.tf_tokenizers:
											__lowerCAmelCase	   =tf.function(snake_case_)
											for test_inputs in (self.test_sentences, self.paired_sentences):
																__lowerCAmelCase	   =tf.constant(snake_case_)
																__lowerCAmelCase	   =compiled_tokenizer(snake_case_)
																__lowerCAmelCase	   =tf_tokenizer(snake_case_)

																for key in eager_outputs.keys():
																					self.assertTrue(tf.reduce_all(eager_outputs[key] == compiled_outputs[key]))



	@slow
	def 		UpperCamelCase							(						self							:    Optional[Any])->							Optional[int]:
						for tf_tokenizer in self.tf_tokenizers:
											__lowerCAmelCase	   =ModelToSave(tokenizer=snake_case_)
											__lowerCAmelCase	   =tf.convert_to_tensor(self.test_sentences)
											__lowerCAmelCase	   =model(snake_case_)  # Build model with some sample inputs
											with TemporaryDirectory() as tempdir:
																__lowerCAmelCase	   =Path(snake_case_) / """saved.model"""
																model.save(snake_case_)
																__lowerCAmelCase	   =tf.keras.models.load_model(snake_case_)
											__lowerCAmelCase	   =loaded_model(snake_case_)
											# We may see small differences because the loaded model is compiled, so we need an epsilon for the test
											self.assertLessEqual(tf.reduce_max(tf.abs(out - loaded_output))				,	1e-5)









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



	def 		UpperCamelCase							(						self							:    Dict)->							Any:
						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=snake_case_				,	)
						assert hasattr(self				,	"""env""")



	def 		UpperCamelCase							(						self							:    Dict				,	snake_case_							:    str)->							Dict:
						__lowerCAmelCase	   =F"""{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}"""
						# distributed data settings
						__lowerCAmelCase	   ={"""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=snake_case_				,	instance_count=snake_case_				,	instance_type=self.instance_type				,	debugger_hook_config=snake_case_				,	hyperparameters={**self.env.distributed_hyperparameters, """model_name_or_path""": self.model_name_or_path}				,	metric_definitions=self.env.metric_definitions				,	distribution=snake_case_				,	py_version="""py36"""				,	)



	def 		UpperCamelCase							(						self							:    int				,	snake_case_							:    List[Any])->							int:
						TrainingJobAnalytics(snake_case_).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv""")



	@parameterized.expand([(2,)])
	def 		UpperCamelCase							(						self							:    Any				,	snake_case_							:    Dict)->							Optional[Any]:
						# create estimator
						__lowerCAmelCase	   =self.create_estimator(snake_case_)

						# run training
						estimator.fit()

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

						# extract kpis
						__lowerCAmelCase	   =list(result_metrics_df[result_metrics_df.metric_name == """eval_accuracy"""]["""value"""])
						__lowerCAmelCase	   =list(result_metrics_df[result_metrics_df.metric_name == """eval_loss"""]["""value"""])
						# get train time from SageMaker job, this includes starting, preprocessing, stopping
						__lowerCAmelCase	   =(
						    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}				,	snake_case_)




