



"""simple docstring"""


import time
from contextlib import contextmanager
from pathlib import Path

import pytest
import requests
from huggingface_hub.hf_api import HfApi, HfFolder


UpperCamelCase						:				List[str]             =      "__DUMMY_TRANSFORMERS_USER__"
UpperCamelCase						:				Optional[int]             =      "Dummy User"
UpperCamelCase						:				Union[str, Any]             =      "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"

UpperCamelCase						:				Any             =      "https://hub-ci.huggingface.co"
UpperCamelCase						:				str             =      CI_HUB_ENDPOINT + "/datasets/{repo_id}/resolve/{revision}/{path}"
UpperCamelCase						:				Optional[int]             =      CI_HUB_ENDPOINT + "/{repo_id}/resolve/{revision}/{filename}"
UpperCamelCase						:				int             =      Path("~/.huggingface/hub_ci_token").expanduser()







@pytest.fixture
def  __snake_case       (   UpperCamelCase__	)      ->		Optional[int]:







		"""simple docstring"""





		monkeypatch.setattr(
		    'huggingface_hub.file_download.HUGGINGFACE_CO_URL_TEMPLATE' ,			UpperCamelCase__	)







@pytest.fixture
def  __snake_case       (   UpperCamelCase__	)      ->		str:







		"""simple docstring"""





		monkeypatch.setattr('datasets.config.HF_ENDPOINT' ,			UpperCamelCase__	)
		monkeypatch.setattr('datasets.config.HUB_DATASETS_URL' ,			UpperCamelCase__	)







@pytest.fixture
def  __snake_case       (   UpperCamelCase__	)      ->		Dict:







		"""simple docstring"""





		monkeypatch.setattr('huggingface_hub.hf_api.HfFolder.path_token' ,			UpperCamelCase__	)







@pytest.fixture
def  __snake_case       (   UpperCamelCase__ ,			UpperCamelCase__	)      ->		int:







		"""simple docstring"""





		HfFolder.save_token(UpperCamelCase__	)
		yield
		HfFolder.delete_token()







@pytest.fixture(scope='session'	)
def  __snake_case       (   )      ->		Optional[int]:







		"""simple docstring"""





		return HfApi(endpoint=UpperCamelCase__	)







@pytest.fixture(scope='session'	)
def  __snake_case       (   UpperCamelCase__	)      ->		Union[str, Any]:







		"""simple docstring"""





		A					       = HfFolder.get_token()
		HfFolder.save_token(UpperCamelCase__	)
		yield CI_HUB_USER_TOKEN
		if previous_token is not None:
				HfFolder.save_token(UpperCamelCase__	)







@pytest.fixture
def  __snake_case       (   UpperCamelCase__	)      ->		Tuple:







		"""simple docstring"""





		def _cleanup_repo(UpperCamelCase__	):
				hf_api.delete_repo(UpperCamelCase__ ,			token=UpperCamelCase__ ,			repo_type='dataset'	)

		return _cleanup_repo







@pytest.fixture
def  __snake_case       (   UpperCamelCase__	)      ->		int:







		"""simple docstring"""





		@contextmanager
		def _temporary_repo(UpperCamelCase__	):
				try:
						yield repo_id
				finally:
						cleanup_repo(UpperCamelCase__	)

		return _temporary_repo







@pytest.fixture(scope='session'	)
def  __snake_case       (   UpperCamelCase__ ,			UpperCamelCase__ ,			UpperCamelCase__	)      ->		List[Any]:







		"""simple docstring"""





		A					       = f'repo_txt_data-{int(time.time() * 10E3	)}'
		A					       = f'{CI_HUB_USER}/{repo_name}'
		hf_api.create_repo(UpperCamelCase__ ,			token=UpperCamelCase__ ,			repo_type='dataset' ,			private=UpperCamelCase__	)
		hf_api.upload_file(
		    token=UpperCamelCase__ ,			path_or_fileobj=str(UpperCamelCase__	) ,			path_in_repo='data/text_data.txt' ,			repo_id=UpperCamelCase__ ,			repo_type='dataset' ,			)
		yield repo_id
		try:
				hf_api.delete_repo(UpperCamelCase__ ,			token=UpperCamelCase__ ,			repo_type='dataset'	)
		except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
				pass







@pytest.fixture()
def  __snake_case       (   UpperCamelCase__ ,			UpperCamelCase__ ,			UpperCamelCase__	)      ->		List[Any]:







		"""simple docstring"""





		return hf_private_dataset_repo_txt_data_







@pytest.fixture(scope='session'	)
def  __snake_case       (   UpperCamelCase__ ,			UpperCamelCase__ ,			UpperCamelCase__	)      ->		Tuple:







		"""simple docstring"""





		A					       = f'repo_zipped_txt_data-{int(time.time() * 10E3	)}'
		A					       = f'{CI_HUB_USER}/{repo_name}'
		hf_api.create_repo(UpperCamelCase__ ,			token=UpperCamelCase__ ,			repo_type='dataset' ,			private=UpperCamelCase__	)
		hf_api.upload_file(
		    token=UpperCamelCase__ ,			path_or_fileobj=str(UpperCamelCase__	) ,			path_in_repo='data.zip' ,			repo_id=UpperCamelCase__ ,			repo_type='dataset' ,			)
		yield repo_id
		try:
				hf_api.delete_repo(UpperCamelCase__ ,			token=UpperCamelCase__ ,			repo_type='dataset'	)
		except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
				pass







@pytest.fixture()
def  __snake_case       (   UpperCamelCase__ ,			UpperCamelCase__ ,			UpperCamelCase__	)      ->		List[str]:







		"""simple docstring"""





		return hf_private_dataset_repo_zipped_txt_data_







@pytest.fixture(scope='session'	)
def  __snake_case       (   UpperCamelCase__ ,			UpperCamelCase__ ,			UpperCamelCase__	)      ->		Union[str, Any]:







		"""simple docstring"""





		A					       = f'repo_zipped_img_data-{int(time.time() * 10E3	)}'
		A					       = f'{CI_HUB_USER}/{repo_name}'
		hf_api.create_repo(UpperCamelCase__ ,			token=UpperCamelCase__ ,			repo_type='dataset' ,			private=UpperCamelCase__	)
		hf_api.upload_file(
		    token=UpperCamelCase__ ,			path_or_fileobj=str(UpperCamelCase__	) ,			path_in_repo='data.zip' ,			repo_id=UpperCamelCase__ ,			repo_type='dataset' ,			)
		yield repo_id
		try:
				hf_api.delete_repo(UpperCamelCase__ ,			token=UpperCamelCase__ ,			repo_type='dataset'	)
		except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
				pass







@pytest.fixture()
def  __snake_case       (   UpperCamelCase__ ,			UpperCamelCase__ ,			UpperCamelCase__	)      ->		Tuple:







		"""simple docstring"""





		return hf_private_dataset_repo_zipped_img_data_









"""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.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 650, """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""": 600, """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""": 600, """eval_accuracy""": 0.6, """eval_loss""": 0.7},
        },
    ]			)
class 			lowerCamelCase__						(   unittest.TestCase			):







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







						def __a						(							self   :    Optional[int]			,			_lowercase   :    int		):
								A					       = f'{self.env.base_job_name}-{instance_count}-{"ddp" if "ddp" in self.script else "smd"}'
								# distributed data settings
								A					       = {'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=_lowercase			,			instance_count=_lowercase			,			instance_type=self.instance_type			,			debugger_hook_config=_lowercase			,			hyperparameters={**self.env.distributed_hyperparameters, 'model_name_or_path': self.model_name_or_path}			,			metric_definitions=self.env.metric_definitions			,			distribution=_lowercase			,			py_version='py36'			,			)







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







						@parameterized.expand([(2,)]		)
						def __a						(							self   :    List[Any]			,			_lowercase   :    Union[str, Any]		):
								# create estimator
								A					       = self.create_estimator(_lowercase		)

								# 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'			,			999_999		)
								)

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





