

"""simple docstring"""





import os
import unittest

from huggingface_hub.utils import are_progress_bars_disabled

import transformers.models.bart.tokenization_bart
from transformers import logging
from transformers.testing_utils import CaptureLogger, mockenv, mockenv_context
from transformers.utils.logging import disable_progress_bar, enable_progress_bar




class 					_lowerCAmelCase  (	unittest.TestCase   ):







 """simple docstring"""







 def 					_lowercase (			self							:			List[str]	):
    __lowercase								=						logging.get_logger()

    # the current default level is logging.WARNING
    __lowercase								=						logging.get_verbosity()

    logging.set_verbosity_error()
    self.assertEqual(logger.getEffectiveLevel(),					logging.get_verbosity()	)

    logging.set_verbosity_warning()
    self.assertEqual(logger.getEffectiveLevel(),					logging.get_verbosity()	)

    logging.set_verbosity_info()
    self.assertEqual(logger.getEffectiveLevel(),					logging.get_verbosity()	)

    logging.set_verbosity_debug()
    self.assertEqual(logger.getEffectiveLevel(),					logging.get_verbosity()	)

    # restore to the original level
    logging.set_verbosity(UpperCAmelCase__	)


 def 					_lowercase (			self							:			str	):
    __lowercase								=						logging.get_verbosity()

    __lowercase								=						logging.get_logger("transformers.models.bart.tokenization_bart"	)
    __lowercase								=						"Testing 1, 2, 3"

    # should be able to log warnings (if default settings weren't overridden by `pytest --log-level-all`)
    if level_origin <= logging.WARNING:
       with CaptureLogger(UpperCAmelCase__	) as cl:
          logger.warning(UpperCAmelCase__	)
       self.assertEqual(cl.out,					msg + "\n"	)

    # this is setting the level for all of `transformers.*` loggers
    logging.set_verbosity_error()

    # should not be able to log warnings
    with CaptureLogger(UpperCAmelCase__	) as cl:
       logger.warning(UpperCAmelCase__	)
    self.assertEqual(cl.out,					""	)

    # should be able to log warnings again
    logging.set_verbosity_warning()
    with CaptureLogger(UpperCAmelCase__	) as cl:
       logger.warning(UpperCAmelCase__	)
    self.assertEqual(cl.out,					msg + "\n"	)

    # restore to the original level
    logging.set_verbosity(UpperCAmelCase__	)


 @mockenv(TRANSFORMERS_VERBOSITY="error"	)
 def 					_lowercase (			self							:			Any	):
    # reset for the env var to take effect, next time some logger call is made
    transformers.utils.logging._reset_library_root_logger()
    # this action activates the env var
    __lowercase								=						logging.get_logger("transformers.models.bart.tokenization_bart"	)

    __lowercase								=						os.getenv("TRANSFORMERS_VERBOSITY",					UpperCAmelCase__	)
    __lowercase								=						logging.log_levels[env_level_str]

    __lowercase								=						logging.get_verbosity()
    self.assertEqual(
        UpperCAmelCase__,					UpperCAmelCase__,					F"""TRANSFORMERS_VERBOSITY={env_level_str}/{env_level}, but internal verbosity is {current_level}""",					)

    # restore to the original level
    __lowercase								=						""
    transformers.utils.logging._reset_library_root_logger()


 @mockenv(TRANSFORMERS_VERBOSITY="super-error"	)
 def 					_lowercase (			self							:			str	):
    # reset for the env var to take effect, next time some logger call is made
    transformers.utils.logging._reset_library_root_logger()
    __lowercase								=						logging.logging.getLogger()
    with CaptureLogger(UpperCAmelCase__	) as cl:
       # this action activates the env var
       logging.get_logger("transformers.models.bart.tokenization_bart"	)
    self.assertIn("Unknown option TRANSFORMERS_VERBOSITY=super-error",					cl.out	)

    # no need to restore as nothing was changed



 def 					_lowercase (			self							:			Union[str, Any]	):
    # testing `logger.warning_advice()`
    transformers.utils.logging._reset_library_root_logger()

    __lowercase								=						logging.get_logger("transformers.models.bart.tokenization_bart"	)
    __lowercase								=						"Testing 1, 2, 3"

    with mockenv_context(TRANSFORMERS_NO_ADVISORY_WARNINGS="1"	):
       # nothing should be logged as env var disables this method
       with CaptureLogger(UpperCAmelCase__	) as cl:
          logger.warning_advice(UpperCAmelCase__	)
       self.assertEqual(cl.out,					""	)

    with mockenv_context(TRANSFORMERS_NO_ADVISORY_WARNINGS=""	):
       # should log normally as TRANSFORMERS_NO_ADVISORY_WARNINGS is unset
       with CaptureLogger(UpperCAmelCase__	) as cl:
          logger.warning_advice(UpperCAmelCase__	)
       self.assertEqual(cl.out,					msg + "\n"	)







def   _A  (  )	->						int:
   '''simple docstring'''


   disable_progress_bar()
   assert are_progress_bars_disabled()

   enable_progress_bar()
   assert not are_progress_bars_disabled()







"""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": 1_6_0_0, "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": 1_6_0_0, "eval_accuracy": 0.3, "eval_loss": 1.2},
        },
    ]   )
class 					_lowerCAmelCase  (	unittest.TestCase   ):







 """simple docstring"""







 def 					_lowercase (			self							:			Optional[int]	):
    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=UpperCAmelCase__,					)
    assert hasattr(self,					"env"	)


 def 					_lowercase (			self							:			str,					UpperCAmelCase__							:			List[Any]	):
    # configuration for running training on smdistributed Model Parallel
    __lowercase								=						{
        "enabled": True,
        "processes_per_host": 8,
    }
    __lowercase								=						{
        "enabled": True,
        "parameters": {
            "microbatches": 4,
            "placement_strategy": "spread",
            "pipeline": "interleaved",
            "optimize": "speed",
            "partitions": 4,
            "ddp": True,
        },
    }

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

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


 def 					_lowercase (			self							:			Tuple,					UpperCAmelCase__							:			int	):
    TrainingJobAnalytics(UpperCAmelCase__	).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv"""	)



 @parameterized.expand([(1,)]	)
 def 					_lowercase (			self							:			str,					UpperCAmelCase__							:			Union[str, Any]	):
    # create estimator
    __lowercase								=						self.create_estimator(UpperCAmelCase__	)

    # run training
    estimator.fit()

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

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

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




