



import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoImageProcessor, ViTImageProcessor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / '''utils'''))

from test_module.custom_image_processing import CustomImageProcessor  # noqa E402


lowerCAmelCase__					  =  get_tests_dir('''fixtures''')

class 				snake_case__(unittest.TestCase      ):

					"""simple docstring"""






					def 			snake_case					(							self :     List[str]					):
									# A mock response for an HTTP head request to emulate server down
									lowercase__					:   Optional[int]													=					mock.Mock()
									lowercase__					:   int													=					500
									lowercase__					:   str													=					{}
									lowercase__					:   Dict													=					HTTPError
									lowercase__					:   Any													=					{}

									# Download this model to make sure it's in the cache.
									lowercase__					:   List[Any]													=					ViTImageProcessor.from_pretrained("hf-internal-testing/tiny-random-vit"					)
									# Under the mock environment we get a 500 error when trying to reach the model.
									with mock.patch("requests.Session.request"				,       return_value=SCREAMING_SNAKE_CASE					) as mock_head:
													lowercase__					:   List[str]													=					ViTImageProcessor.from_pretrained("hf-internal-testing/tiny-random-vit"					)
													# This check we did call the fake head request
													mock_head.assert_called()







					def 			snake_case					(							self :     str					):
									# This test is for deprecated behavior and can be removed in v5
									lowercase__					:   Optional[int]													=					ViTImageProcessor.from_pretrained(
									    "https://huggingface.co/hf-internal-testing/tiny-random-vit/resolve/main/preprocessor_config.json"					)







					def 			snake_case					(							self :     Any					):
									with self.assertRaises(SCREAMING_SNAKE_CASE					):
													# config is in subfolder, the following should not work without specifying the subfolder
													lowercase__					:   Union[str, Any]													=					AutoImageProcessor.from_pretrained("hf-internal-testing/stable-diffusion-all-variants"					)

									lowercase__					:   Optional[int]													=					AutoImageProcessor.from_pretrained(
									    "hf-internal-testing/stable-diffusion-all-variants"				,       subfolder="feature_extractor"					)

									self.assertIsNotNone(SCREAMING_SNAKE_CASE					)






@is_staging_test
class 				snake_case__(unittest.TestCase      ):

					"""simple docstring"""






					@classmethod
					def 			snake_case					(							cls :     Tuple					):
									lowercase__					:   int													=					TOKEN
									HfFolder.save_token(SCREAMING_SNAKE_CASE					)







					@classmethod
					def 			snake_case					(							cls :     Optional[int]					):
									try:
													delete_repo(token=cls._token				,       repo_id="test-image-processor"					)
									except HTTPError:
													pass

									try:
													delete_repo(token=cls._token				,       repo_id="valid_org/test-image-processor-org"					)
									except HTTPError:
													pass

									try:
													delete_repo(token=cls._token				,       repo_id="test-dynamic-image-processor"					)
									except HTTPError:
													pass







					def 			snake_case					(							self :     int					):
									lowercase__					:   Dict													=					ViTImageProcessor.from_pretrained(SCREAMING_SNAKE_CASE					)
									image_processor.push_to_hub("test-image-processor"				,       use_auth_token=self._token					)

									lowercase__					:   Optional[int]													=					ViTImageProcessor.from_pretrained(f"""{USER}/test-image-processor"""					)
									for k, v in image_processor.__dict__.items():
													self.assertEqual(SCREAMING_SNAKE_CASE				,       getattr(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)					)

									# Reset repo
									delete_repo(token=self._token				,       repo_id="test-image-processor"					)

									# Push to hub via save_pretrained
									with tempfile.TemporaryDirectory() as tmp_dir:
													image_processor.save_pretrained(
													    SCREAMING_SNAKE_CASE				,       repo_id="test-image-processor"				,       push_to_hub=SCREAMING_SNAKE_CASE				,       use_auth_token=self._token					)

									lowercase__					:   int													=					ViTImageProcessor.from_pretrained(f"""{USER}/test-image-processor"""					)
									for k, v in image_processor.__dict__.items():
													self.assertEqual(SCREAMING_SNAKE_CASE				,       getattr(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)					)







					def 			snake_case					(							self :     Union[str, Any]					):
									lowercase__					:   Dict													=					ViTImageProcessor.from_pretrained(SCREAMING_SNAKE_CASE					)
									image_processor.push_to_hub("valid_org/test-image-processor"				,       use_auth_token=self._token					)

									lowercase__					:   Dict													=					ViTImageProcessor.from_pretrained("valid_org/test-image-processor"					)
									for k, v in image_processor.__dict__.items():
													self.assertEqual(SCREAMING_SNAKE_CASE				,       getattr(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)					)

									# Reset repo
									delete_repo(token=self._token				,       repo_id="valid_org/test-image-processor"					)

									# Push to hub via save_pretrained
									with tempfile.TemporaryDirectory() as tmp_dir:
													image_processor.save_pretrained(
													    SCREAMING_SNAKE_CASE				,       repo_id="valid_org/test-image-processor-org"				,       push_to_hub=SCREAMING_SNAKE_CASE				,       use_auth_token=self._token					)

									lowercase__					:   str													=					ViTImageProcessor.from_pretrained("valid_org/test-image-processor-org"					)
									for k, v in image_processor.__dict__.items():
													self.assertEqual(SCREAMING_SNAKE_CASE				,       getattr(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)					)







					def 			snake_case					(							self :     Dict					):
									CustomImageProcessor.register_for_auto_class()
									lowercase__					:   Optional[Any]													=					CustomImageProcessor.from_pretrained(SCREAMING_SNAKE_CASE					)

									image_processor.push_to_hub("test-dynamic-image-processor"				,       use_auth_token=self._token					)

									# This has added the proper auto_map field to the config
									self.assertDictEqual(
									    image_processor.auto_map				,       {"AutoImageProcessor": "custom_image_processing.CustomImageProcessor"}				,       )

									lowercase__					:   int													=					AutoImageProcessor.from_pretrained(
									    f"""{USER}/test-dynamic-image-processor"""				,       trust_remote_code=SCREAMING_SNAKE_CASE					)
									# Can't make an isinstance check because the new_image_processor is from the CustomImageProcessor class of a dynamic module
									self.assertEqual(new_image_processor.__class__.__name__				,       "CustomImageProcessor"					)









import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import 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.g4dn.xlarge""",
            """results""": {"""train_runtime""": 6_5_0, """eval_accuracy""": 0.6, """eval_loss""": 0.9},
        },
        {
            """framework""": """tensorflow""",
            """script""": """run_tf.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.g4dn.xlarge""",
            """results""": {"""train_runtime""": 6_0_0, """eval_accuracy""": 0.3, """eval_loss""": 0.9},
        },
    ]      )
class 				snake_case__(unittest.TestCase      ):

					"""simple docstring"""






					def 			snake_case					(							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=SCREAMING_SNAKE_CASE				,       )
									assert hasattr(self				,       "env"					)







					def 			snake_case					(							self :     Tuple				,       SCREAMING_SNAKE_CASE :     str=1					):
									# 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}-single"""				,       instance_count=SCREAMING_SNAKE_CASE				,       instance_type=self.instance_type				,       debugger_hook_config=SCREAMING_SNAKE_CASE				,       hyperparameters={**self.env.hyperparameters, "model_name_or_path": self.model_name_or_path}				,       metric_definitions=self.env.metric_definitions				,       py_version="py36"				,       )







					def 			snake_case					(							self :     Union[str, Any]				,       SCREAMING_SNAKE_CASE :     str					):
									TrainingJobAnalytics(SCREAMING_SNAKE_CASE					).export_csv(f"""{self.env.test_path}/{job_name}_metrics.csv"""					)







					def 			snake_case					(							self :     str					):
									# create estimator
									lowercase__					:   Optional[int]													=					self.create_estimator()

									# run training
									estimator.fit()

									# result dataframe
									lowercase__					:   Union[str, Any]													=					TrainingJobAnalytics(estimator.latest_training_job.name					).dataframe()

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




