




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_model_parallelism.py""",
            """model_name_or_path""": """roberta-large""",
            """instance_type""": """ml.p3dn.24xlarge""",
            """results""": {"""train_runtime""": 16_00, """eval_accuracy""": 0.3, """eval_loss""": 1.2},
        },
        {
            """framework""": """pytorch""",
            """script""": """run_glue.py""",
            """model_name_or_path""": """roberta-large""",
            """instance_type""": """ml.p3dn.24xlarge""",
            """results""": {"""train_runtime""": 16_00, """eval_accuracy""": 0.3, """eval_loss""": 1.2},
        },
    ]     )
class       _A				(		unittest.TestCase     ):






		def 				UpperCAmelCase (    self			):
								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=_SCREAMING_SNAKE_CASE						,		)
								assert hasattr(self						,		"""env"""			)






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								# configuration for running training on smdistributed Model Parallel
								_UpperCAmelCase      =			{
								    """enabled""": True,
								    """processes_per_host""": 8,
								}
								_UpperCAmelCase      =			{
								    """enabled""": True,
								    """parameters""": {
								        """microbatches""": 4,
								        """placement_strategy""": """spread""",
								        """pipeline""": """interleaved""",
								        """optimize""": """speed""",
								        """partitions""": 4,
								        """ddp""": True,
								    },
								}

								_UpperCAmelCase      =			{"""smdistributed""": {"""modelparallel""": smp_options}, """mpi""": mpi_options}

								_UpperCAmelCase      =			"""trainer""" if self.script == """run_glue.py""" else """smtrainer"""
								# 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=F"{self.env.base_job_name}-{instance_count}-smp-{name_extension}"						,		instance_count=_SCREAMING_SNAKE_CASE						,		instance_type=self.instance_type						,		debugger_hook_config=_SCREAMING_SNAKE_CASE						,		hyperparameters={
								        **self.env.hyperparameters,
								        """model_name_or_path""": self.model_name_or_path,
								        """max_steps""": 500,
								    }						,		metric_definitions=self.env.metric_definitions						,		distribution=_SCREAMING_SNAKE_CASE						,		py_version="""py36"""						,		)






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								TrainingJobAnalytics(_SCREAMING_SNAKE_CASE			).export_csv(F"{self.env.test_path}/{job_name}_metrics.csv"			)






		@parameterized.expand([(1,)]			)
		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								# create estimator
								_UpperCAmelCase      =			self.create_estimator(_SCREAMING_SNAKE_CASE			)

								# run training
								estimator.fit()

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

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

								# 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}						,		_SCREAMING_SNAKE_CASE			)





import os
import sys
import unittest


a													=			os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, "utils"))

import check_dummies  # noqa: E402
from check_dummies import create_dummy_files, create_dummy_object, find_backend, read_init  # noqa: E402


# Align TRANSFORMERS_PATH in check_dummies with the current path
a													=			os.path.join(git_repo_path, "src", "transformers")

a													=			"\n{0} = None\n"

a													=			"\nclass {0}(metaclass=DummyObject):\n    _backends = {1}\n\n    def __init__(self, *args, **kwargs):\n        requires_backends(self, {1})\n"


a													=			"\ndef {0}(*args, **kwargs):\n    requires_backends({0}, {1})\n"




class       _A				(		unittest.TestCase     ):






		def 				UpperCAmelCase (    self			):
								_UpperCAmelCase      =			find_backend("""    _import_structure[\"models.albert\"].append(\"AlbertTokenizerFast\")"""			)
								self.assertIsNone(_SCREAMING_SNAKE_CASE			)

								_UpperCAmelCase      =			find_backend("""    if not is_tokenizers_available():"""			)
								self.assertEqual(_SCREAMING_SNAKE_CASE						,		"""tokenizers"""			)

								_UpperCAmelCase      =			find_backend("""    if not is_tensorflow_text_available():"""			)
								self.assertEqual(_SCREAMING_SNAKE_CASE						,		"""tensorflow_text"""			)

								_UpperCAmelCase      =			find_backend("""    if not (is_sentencepiece_available() and is_tokenizers_available()):"""			)
								self.assertEqual(_SCREAMING_SNAKE_CASE						,		"""sentencepiece_and_tokenizers"""			)

								_UpperCAmelCase      =			find_backend(
								    """    if not (is_sentencepiece_available() and is_tensorflow_text_available()):"""			)
								self.assertEqual(_SCREAMING_SNAKE_CASE						,		"""sentencepiece_and_tensorflow_text"""			)

								_UpperCAmelCase      =			find_backend(
								    """    if not (is_sentencepiece_available() and is_tokenizers_available() and is_vision_available()):"""			)
								self.assertEqual(_SCREAMING_SNAKE_CASE						,		"""sentencepiece_and_tokenizers_and_vision"""			)






		def 				UpperCAmelCase (    self			):
								_UpperCAmelCase      =			read_init()
								# We don't assert on the exact list of keys to allow for smooth grow of backend-specific objects
								self.assertIn("""torch"""						,		_SCREAMING_SNAKE_CASE			)
								self.assertIn("""tensorflow_text"""						,		_SCREAMING_SNAKE_CASE			)
								self.assertIn("""sentencepiece_and_tokenizers"""						,		_SCREAMING_SNAKE_CASE			)

								# Likewise, we can't assert on the exact content of a key
								self.assertIn("""BertModel"""						,		objects["""torch"""]			)
								self.assertIn("""TFBertModel"""						,		objects["""tf"""]			)
								self.assertIn("""FlaxBertModel"""						,		objects["""flax"""]			)
								self.assertIn("""BertModel"""						,		objects["""torch"""]			)
								self.assertIn("""TFBertTokenizer"""						,		objects["""tensorflow_text"""]			)
								self.assertIn("""convert_slow_tokenizer"""						,		objects["""sentencepiece_and_tokenizers"""]			)






		def 				UpperCAmelCase (    self			):
								_UpperCAmelCase      =			create_dummy_object("""CONSTANT"""						,		"""'torch'"""			)
								self.assertEqual(_SCREAMING_SNAKE_CASE						,		"""\nCONSTANT = None\n"""			)

								_UpperCAmelCase      =			create_dummy_object("""function"""						,		"""'torch'"""			)
								self.assertEqual(
								    _SCREAMING_SNAKE_CASE						,		"""\ndef function(*args, **kwargs):\n    requires_backends(function, 'torch')\n"""			)

								_UpperCAmelCase      =			"""
class FakeClass(metaclass=DummyObject):
    _backends = 'torch'

    def __init__(self, *args, **kwargs):
        requires_backends(self, 'torch')
"""
								_UpperCAmelCase      =			create_dummy_object("""FakeClass"""						,		"""'torch'"""			)
								self.assertEqual(_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE			)






		def 				UpperCAmelCase (    self			):
								_UpperCAmelCase      =			"""# This file is autogenerated by the command `make fix-copies`, do not edit.
from ..utils import DummyObject, requires_backends


CONSTANT = None


def function(*args, **kwargs):
    requires_backends(function, [\"torch\"])


class FakeClass(metaclass=DummyObject):
    _backends = [\"torch\"]

    def __init__(self, *args, **kwargs):
        requires_backends(self, [\"torch\"])
"""
								_UpperCAmelCase      =			create_dummy_files({"""torch""": ["""CONSTANT""", """function""", """FakeClass"""]}			)
								self.assertEqual(dummy_files["""torch"""]						,		_SCREAMING_SNAKE_CASE			)