




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






			'''simple docstring'''





			def 							UpperCAmelCase__       ( self			:   List[str]  )						->   Union[str, 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=lowerCamelCase_       ,						)
								assert hasattr(self       ,						'''env'''  )

			def 							UpperCAmelCase__       ( self			:   Optional[Any]       ,						lowerCamelCase_			:   Any  )						->   Tuple:
								__magic_name__							:	int    							=							F'''{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}'''
								# distributed data settings
								__magic_name__							:	Tuple    							=							{'''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 							UpperCAmelCase__       ( self			:   Optional[Any]       ,						lowerCamelCase_			:   Optional[int]  )						->   int:
								TrainingJobAnalytics(lowerCamelCase_  ).export_csv(F'''{self.env.test_path}/{job_name}_metrics.csv'''  )


			@parameterized.expand([(2,)]  )
			def 							UpperCAmelCase__       ( self			:   Union[str, Any]       ,						lowerCamelCase_			:   Any  )						->   Optional[Any]:
								# create estimator
								__magic_name__							:	Dict    							=							self.create_estimator(lowerCamelCase_  )

								# run training
								estimator.fit()

								# result dataframe
								__magic_name__							:	str    							=							TrainingJobAnalytics(estimator.latest_training_job.name  ).dataframe()

								# extract kpis
								__magic_name__							:	Optional[Any]    							=							list(result_metrics_df[result_metrics_df.metric_name == '''eval_accuracy''']['''value''']  )
								__magic_name__							:	Union[str, Any]    							=							list(result_metrics_df[result_metrics_df.metric_name == '''eval_loss''']['''value''']  )
								# get train time from SageMaker job, this includes starting, preprocessing, stopping
								__magic_name__							:	List[str]    							=							(
								    Session().describe_training_job(estimator.latest_training_job.name  ).get('''TrainingTimeInSeconds'''       ,						999999  )
								)

								# 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 inspect
import os
import unittest
from pathlib import Path

import torch

import accelerate
from accelerate.test_utils import execute_subprocess_async
from accelerate.test_utils.testing import run_command


class    lowerCamelCase    (		unittest.TestCase						):






			'''simple docstring'''





			UpperCamelCase__								=inspect.getfile(accelerate.test_utils						)
			UpperCamelCase__								=os.path.sep.join(mod_file.split(os.path.sep						)[:-1] + ['''scripts''', '''test_cli.py''']						)

			UpperCamelCase__								=['''accelerate''', '''launch''']
			UpperCamelCase__								=Path.home() / '''.cache/huggingface/accelerate'''
			UpperCamelCase__								='''default_config.yaml'''
			UpperCamelCase__								=config_folder / config_file
			UpperCamelCase__								=config_folder / '''_default_config.yaml'''

			UpperCamelCase__								=Path('''tests/test_configs'''						)

			@classmethod
			def 							UpperCAmelCase__       ( cls			:   Optional[int]  )						->   int:
								if cls.config_path.is_file():
													cls.config_path.rename(cls.changed_path  )

			@classmethod
			def 							UpperCAmelCase__       ( cls			:   Dict  )						->   List[str]:
								if cls.changed_path.is_file():
													cls.changed_path.rename(cls.config_path  )

			def 							UpperCAmelCase__       ( self			:   str  )						->   Optional[Any]:
								__magic_name__							:	Dict    							=							self.base_cmd
								if torch.cuda.is_available() and (torch.cuda.device_count() > 1):
													cmd += ["--multi_gpu"]
								execute_subprocess_async(cmd + [self.test_file_path]       ,						env=os.environ.copy()  )

			def 							UpperCAmelCase__       ( self			:   int  )						->   Optional[int]:
								for config in sorted(self.test_config_path.glob('''**/*.yaml'''  )  ):
													with self.subTest(config_file=lowerCamelCase_  ):
																		execute_subprocess_async(
																		    self.base_cmd + ['''--config_file''', str(lowerCamelCase_  ), self.test_file_path]       ,						env=os.environ.copy()  )


			def 							UpperCAmelCase__       ( self			:   int  )						->   Union[str, Any]:
								execute_subprocess_async(['''accelerate''', '''test''']       ,						env=os.environ.copy()  )






class    lowerCamelCase    (		unittest.TestCase						):






			'''simple docstring'''





			UpperCamelCase__								='''test-tpu'''
			UpperCamelCase__								='''us-central1-a'''
			UpperCamelCase__								='''ls'''
			UpperCamelCase__								=['''accelerate''', '''tpu-config''']
			UpperCamelCase__								='''cd /usr/share'''
			UpperCamelCase__								='''tests/test_samples/test_command_file.sh'''
			UpperCamelCase__								='''Running gcloud compute tpus tpu-vm ssh'''

			def 							UpperCAmelCase__       ( self			:   Optional[Any]  )						->   Optional[Any]:
								__magic_name__							:	str    							=							run_command(
								    self.cmd
								    + ['''--command''', self.command, '''--tpu_zone''', self.tpu_zone, '''--tpu_name''', self.tpu_name, '''--debug''']       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   Union[str, Any]  )						->   int:
								__magic_name__							:	int    							=							run_command(
								    self.cmd
								    + [
								        '''--config_file''',
								        '''tests/test_configs/0_12_0.yaml''',
								        '''--command''',
								        self.command,
								        '''--tpu_zone''',
								        self.tpu_zone,
								        '''--tpu_name''',
								        self.tpu_name,
								        '''--debug''',
								    ]       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   List[Any]  )						->   Tuple:
								__magic_name__							:	Any    							=							run_command(
								    self.cmd + ['''--config_file''', '''tests/test_configs/latest.yaml''', '''--debug''']       ,						return_stdout=lowerCamelCase_  )
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   Optional[int]  )						->   Tuple:
								__magic_name__							:	Tuple    							=							run_command(
								    self.cmd + ['''--config_file''', '''tests/test_configs/latest.yaml''', '''--command''', self.command, '''--debug''']       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   List[Any]  )						->   Dict:
								__magic_name__							:	str    							=							run_command(
								    self.cmd
								    + [
								        '''--config_file''',
								        '''tests/test_configs/latest.yaml''',
								        '''--command''',
								        self.command,
								        '''--command''',
								        '''echo "Hello World"''',
								        '''--debug''',
								    ]       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls; echo "Hello World" --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   Optional[Any]  )						->   Optional[int]:
								__magic_name__							:	Union[str, Any]    							=							run_command(
								    self.cmd
								    + ['''--config_file''', '''tests/test_configs/latest.yaml''', '''--command_file''', self.command_file, '''--debug''']       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   int  )						->   List[Any]:
								__magic_name__							:	List[Any]    							=							run_command(
								    self.cmd
								    + [
								        '''--config_file''',
								        '''tests/test_configs/0_12_0.yaml''',
								        '''--command_file''',
								        self.command_file,
								        '''--tpu_zone''',
								        self.tpu_zone,
								        '''--tpu_name''',
								        self.tpu_name,
								        '''--debug''',
								    ]       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   str  )						->   Dict:
								__magic_name__							:	Tuple    							=							run_command(
								    self.cmd + ['''--config_file''', '''tests/test_configs/latest.yaml''', '''--install_accelerate''', '''--debug''']       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; pip install accelerate -U; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)


			def 							UpperCAmelCase__       ( self			:   Any  )						->   Tuple:
								__magic_name__							:	Tuple    							=							run_command(
								    self.cmd
								    + [
								        '''--config_file''',
								        '''tests/test_configs/latest.yaml''',
								        '''--install_accelerate''',
								        '''--accelerate_version''',
								        '''12.0.0''',
								        '''--debug''',
								    ]       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; pip install accelerate==12.0.0; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)



