chendl/compositional_test

1import json2import os3import subprocess4import unittest5from ast import literal_eval6 7import pytest8from parameterized import parameterized_class9 10from . import is_sagemaker_available11 12 13if is_sagemaker_available():14    from sagemaker import Session, TrainingJobAnalytics15    from sagemaker.huggingface import HuggingFace16 17 18@pytest.mark.skipif(19    literal_eval(os.getenv("TEST_SAGEMAKER", "False")) is not True,20    reason="Skipping test because should only be run when releasing minor transformers version",21)22@pytest.mark.usefixtures("sm_env")23@parameterized_class(24    [25        {26            "framework": "pytorch",27            "script": "run_glue.py",28            "model_name_or_path": "distilbert-base-cased",29            "instance_type": "ml.g4dn.xlarge",30            "results": {"train_runtime": 650, "eval_accuracy": 0.6, "eval_loss": 0.9},31        },32        {33            "framework": "tensorflow",34            "script": "run_tf.py",35            "model_name_or_path": "distilbert-base-cased",36            "instance_type": "ml.g4dn.xlarge",37            "results": {"train_runtime": 600, "eval_accuracy": 0.3, "eval_loss": 0.9},38        },39    ]40)41class SingleNodeTest(unittest.TestCase):42    def setUp(self):43        if self.framework == "pytorch":44            subprocess.run(45                f"cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py".split(),46                encoding="utf-8",47                check=True,48            )49        assert hasattr(self, "env")50 51    def create_estimator(self, instance_count=1):52        # creates estimator53        return HuggingFace(54            entry_point=self.script,55            source_dir=self.env.test_path,56            role=self.env.role,57            image_uri=self.env.image_uri,58            base_job_name=f"{self.env.base_job_name}-single",59            instance_count=instance_count,60            instance_type=self.instance_type,61            debugger_hook_config=False,62            hyperparameters={**self.env.hyperparameters, "model_name_or_path": self.model_name_or_path},63            metric_definitions=self.env.metric_definitions,64            py_version="py36",65        )66 67    def save_results_as_csv(self, job_name):68        TrainingJobAnalytics(job_name).export_csv(f"{self.env.test_path}/{job_name}_metrics.csv")69 70    def test_glue(self):71        # create estimator72        estimator = self.create_estimator()73 74        # run training75        estimator.fit()76 77        # result dataframe78        result_metrics_df = TrainingJobAnalytics(estimator.latest_training_job.name).dataframe()79 80        # extract kpis81        eval_accuracy = list(result_metrics_df[result_metrics_df.metric_name == "eval_accuracy"]["value"])82        eval_loss = list(result_metrics_df[result_metrics_df.metric_name == "eval_loss"]["value"])83        # get train time from SageMaker job, this includes starting, preprocessing, stopping84        train_runtime = (85            Session().describe_training_job(estimator.latest_training_job.name).get("TrainingTimeInSeconds", 999999)86        )87 88        # assert kpis89        assert train_runtime <= self.results["train_runtime"]90        assert all(t >= self.results["eval_accuracy"] for t in eval_accuracy)91        assert all(t <= self.results["eval_loss"] for t in eval_loss)92 93        # dump tests result into json file to share in PR94        with open(f"{estimator.latest_training_job.name}.json", "w") as outfile:95            json.dump({"train_time": train_runtime, "eval_accuracy": eval_accuracy, "eval_loss": eval_loss}, outfile)96