





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": 6_5_0, "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": 6_0_0, "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": 6_0_0, "eval_accuracy": 0.6, "eval_loss": 0.7},
        },
    ]   )
class SCREAMING_SNAKE_CASE_ (					unittest.TestCase   ):





					"""simple docstring"""


					def 					__magic_name__ ( 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=_lowerCAmelCase	,  )
										assert hasattr(self	,  "env"	)





					def 					__magic_name__ ( self	,  _lowerCAmelCase	):
										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=_lowerCAmelCase	,  instance_count=_lowerCAmelCase	,  instance_type=self.instance_type	,  debugger_hook_config=_lowerCAmelCase	,  hyperparameters={**self.env.distributed_hyperparameters, "model_name_or_path": self.model_name_or_path}	,  metric_definitions=self.env.metric_definitions	,  distribution=_lowerCAmelCase	,  py_version="py36"	,  )





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






					@parameterized.expand([(2,)]	)
					def 					__magic_name__ ( self	,  _lowerCAmelCase	):
										# create estimator
										lowerCamelCase__       =       self.create_estimator(_lowerCAmelCase	)

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








import os
import tempfile
from functools import partial
from unittest import TestCase
from unittest.mock import patch

import datasets
import datasets.config

from .utils import require_beam

class SCREAMING_SNAKE_CASE_ (					datasets.BeamBasedBuilder   ):





					"""simple docstring"""


					def 					__magic_name__ ( self	):
										return datasets.DatasetInfo(
										    features=datasets.Features({"content": datasets.Value("string"	)}	)	,  supervised_keys=_lowerCAmelCase	,  )





					def 					__magic_name__ ( self	,  _lowerCAmelCase	,  _lowerCAmelCase	):
										return [datasets.SplitGenerator(name=datasets.Split.TRAIN	,  gen_kwargs={"examples": get_test_dummy_examples()}	)]






					def 					__magic_name__ ( self	,  _lowerCAmelCase	,  _lowerCAmelCase	):
										import apache_beam as beam

										return pipeline | "Load Examples" >> beam.Create(_lowerCAmelCase	)

class SCREAMING_SNAKE_CASE_ (					datasets.BeamBasedBuilder   ):





					"""simple docstring"""


					def 					__magic_name__ ( self	):
										return datasets.DatasetInfo(
										    features=datasets.Features({"a": datasets.Sequence({"b": datasets.Value("string"	)}	)}	)	,  supervised_keys=_lowerCAmelCase	,  )





					def 					__magic_name__ ( self	,  _lowerCAmelCase	,  _lowerCAmelCase	):
										return [
										    datasets.SplitGenerator(name=datasets.Split.TRAIN	,  gen_kwargs={"examples": get_test_nested_examples()}	)
										]






					def 					__magic_name__ ( self	,  _lowerCAmelCase	,  _lowerCAmelCase	):
										import apache_beam as beam

										return pipeline | "Load Examples" >> beam.Create(_lowerCAmelCase	)





def    __UpperCamelCase      (    ) ->Optional[Any]:
					return [(i, {"content": content}) for i, content in enumerate(["foo", "bar", "foobar"])]





def    __UpperCamelCase      (    ) ->Optional[Any]:
					return [(i, {"a": {"b": [content]}}) for i, content in enumerate(["foo", "bar", "foobar"])]







class SCREAMING_SNAKE_CASE_ (					lowercase_   ):





					"""simple docstring"""


					@require_beam
					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       len(get_test_dummy_examples()	)
										with tempfile.TemporaryDirectory() as tmp_cache_dir:
															lowerCamelCase__       =       DummyBeamDataset(cache_dir=_lowerCAmelCase	,  beam_runner="DirectRunner"	)
															builder.download_and_prepare()
															self.assertTrue(
															    os.path.exists(
															        os.path.join(_lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  F"{builder.name}-train.arrow"	)	)	)
															self.assertDictEqual(builder.info.features	,  datasets.Features({"content": datasets.Value("string"	)}	)	)
															lowerCamelCase__       =       builder.as_dataset()
															self.assertEqual(dset["train"].num_rows	,  _lowerCAmelCase	)
															self.assertEqual(dset["train"].info.splits["train"].num_examples	,  _lowerCAmelCase	)
															self.assertDictEqual(dset["train"][0]	,  get_test_dummy_examples()[0][1]	)
															self.assertDictEqual(
															    dset["train"][expected_num_examples - 1]	,  get_test_dummy_examples()[expected_num_examples - 1][1]	)
															self.assertTrue(
															    os.path.exists(os.path.join(_lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  "dataset_info.json"	)	)	)
															del dset





					@require_beam
					def 					__magic_name__ ( self	):
										import apache_beam as beam

										lowerCamelCase__       =       beam.io.parquetio.WriteToParquet

										lowerCamelCase__       =       len(get_test_dummy_examples()	)
										with tempfile.TemporaryDirectory() as tmp_cache_dir:
															lowerCamelCase__       =       DummyBeamDataset(cache_dir=_lowerCAmelCase	,  beam_runner="DirectRunner"	)
															with patch("apache_beam.io.parquetio.WriteToParquet"	) as write_parquet_mock:
																				lowerCamelCase__       =       partial(_lowerCAmelCase	,  num_shards=2	)
																				builder.download_and_prepare()
															self.assertTrue(
															    os.path.exists(
															        os.path.join(
															            _lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  F"{builder.name}-train-00000-of-00002.arrow"	)	)	)
															self.assertTrue(
															    os.path.exists(
															        os.path.join(
															            _lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  F"{builder.name}-train-00000-of-00002.arrow"	)	)	)
															self.assertDictEqual(builder.info.features	,  datasets.Features({"content": datasets.Value("string"	)}	)	)
															lowerCamelCase__       =       builder.as_dataset()
															self.assertEqual(dset["train"].num_rows	,  _lowerCAmelCase	)
															self.assertEqual(dset["train"].info.splits["train"].num_examples	,  _lowerCAmelCase	)
															# Order is not preserved when sharding, so we just check that all the elements are there
															self.assertListEqual(sorted(dset["train"]["content"]	)	,  sorted(["foo", "bar", "foobar"]	)	)
															self.assertTrue(
															    os.path.exists(os.path.join(_lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  "dataset_info.json"	)	)	)
															del dset





					@require_beam
					def 					__magic_name__ ( self	):
										with tempfile.TemporaryDirectory() as tmp_cache_dir:
															lowerCamelCase__       =       DummyBeamDataset(cache_dir=_lowerCAmelCase	)
															self.assertRaises(datasets.builder.MissingBeamOptions	,  builder.download_and_prepare	)






					@require_beam
					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       len(get_test_nested_examples()	)
										with tempfile.TemporaryDirectory() as tmp_cache_dir:
															lowerCamelCase__       =       NestedBeamDataset(cache_dir=_lowerCAmelCase	,  beam_runner="DirectRunner"	)
															builder.download_and_prepare()
															self.assertTrue(
															    os.path.exists(
															        os.path.join(_lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  F"{builder.name}-train.arrow"	)	)	)
															self.assertDictEqual(
															    builder.info.features	,  datasets.Features({"a": datasets.Sequence({"b": datasets.Value("string"	)}	)}	)	)
															lowerCamelCase__       =       builder.as_dataset()
															self.assertEqual(dset["train"].num_rows	,  _lowerCAmelCase	)
															self.assertEqual(dset["train"].info.splits["train"].num_examples	,  _lowerCAmelCase	)
															self.assertDictEqual(dset["train"][0]	,  get_test_nested_examples()[0][1]	)
															self.assertDictEqual(
															    dset["train"][expected_num_examples - 1]	,  get_test_nested_examples()[expected_num_examples - 1][1]	)
															self.assertTrue(
															    os.path.exists(os.path.join(_lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  "dataset_info.json"	)	)	)
															del dset

