





'''simple docstring'''







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': 1600, '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': 1600, 'eval_accuracy': 0.3, 'eval_loss': 1.2},
        },
    ]      )
class 				__SCREAMING_SNAKE_CASE  (unittest.TestCase      ):






				"""simple docstring"""







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




				def   UpperCamelCase__   (  self     :					Optional[int]					,				__a     :					Optional[int]					):
								# configuration for running training on smdistributed Model Parallel
								_a								= {
								    "enabled": True,
								    "processes_per_host": 8,
								}
								_a								= {
								    "enabled": True,
								    "parameters": {
								        "microbatches": 4,
								        "placement_strategy": "spread",
								        "pipeline": "interleaved",
								        "optimize": "speed",
								        "partitions": 4,
								        "ddp": True,
								    },
								}

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

								_a								= "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=__a					,				instance_type=self.instance_type					,				debugger_hook_config=__a					,				hyperparameters={
								        **self.env.hyperparameters,
								        "model_name_or_path": self.model_name_or_path,
								        "max_steps": 5_00,
								    }					,				metric_definitions=self.env.metric_definitions					,				distribution=__a					,				py_version="py36"					,				)




				def   UpperCamelCase__   (  self     :					List[Any]					,				__a     :					Union[str, Any]					):
								TrainingJobAnalytics(__a					).export_csv(f'{self.env.test_path}/{job_name}_metrics.csv'					)




				@parameterized.expand([(1,)]					)
				def   UpperCamelCase__   (  self     :					Optional[int]					,				__a     :					Tuple					):
								# create estimator
								_a								= self.create_estimator(__a					)

								# run training
								estimator.fit()

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

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







import os

import pytest

from transformers.dynamic_module_utils import get_imports


snake_case_				=			'\nimport os\n'

snake_case_				=			'\ndef foo():\n    import os\n    return False\n'

snake_case_				=			'\ndef foo():\n    def bar():\n        if True:\n            import os\n        return False\n    return bar()\n'

snake_case_				=			'\nimport os\n\ntry:\n    import bar\nexcept ImportError:\n    raise ValueError()\n'

snake_case_				=			'\nimport os\n\ndef foo():\n    try:\n        import bar\n    except ImportError:\n        raise ValueError()\n'

snake_case_				=			'\nimport os\n\ntry:\n    import bar\nexcept (ImportError, AttributeError):\n    raise ValueError()\n'

snake_case_				=			'\nimport os\n\ntry:\n    import bar\nexcept ImportError as e:\n    raise ValueError()\n'

snake_case_				=			'\nimport os\n\ntry:\n    import bar\nexcept:\n    raise ValueError()\n'

snake_case_				=			'\nimport os\n\ntry:\n    import bar\n    import baz\nexcept ImportError:\n    raise ValueError()\n'

snake_case_				=			'\nimport os\n\ntry:\n    import bar\n    import baz\nexcept ImportError:\n    x = 1\n    raise ValueError()\n'

snake_case_				=			[
    TOP_LEVEL_IMPORT,
    IMPORT_IN_FUNCTION,
    DEEPLY_NESTED_IMPORT,
    TOP_LEVEL_TRY_IMPORT,
    GENERIC_EXCEPT_IMPORT,
    MULTILINE_TRY_IMPORT,
    MULTILINE_BOTH_IMPORT,
    MULTIPLE_EXCEPTS_IMPORT,
    EXCEPT_AS_IMPORT,
    TRY_IMPORT_IN_FUNCTION,
]






@pytest.mark.parametrize('''case'''       ,      snake_case_       )
def lowerCamelCase__    (							snake_case_	:  str       ,      snake_case_	:  Optional[int]       )      ->				Dict:
   __snake_case							       =   os.path.join(snake_case_       ,      '''test_file.py'''       )
   with open(snake_case_       ,      '''w'''       ) as _tmp_file:
      _tmp_file.write(snake_case_       )

   __snake_case							       =   get_imports(snake_case_       )
   assert parsed_imports == ["os"]

