


"""simple docstring"""






import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

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

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    PROCESSOR_MAPPING,
    TOKENIZER_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    AutoProcessor,
    AutoTokenizer,
    BertTokenizer,
    ProcessorMixin,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
)
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test
from transformers.tokenization_utils import TOKENIZER_CONFIG_FILE
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_tokenizers_available


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

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402
from test_module.custom_processing import CustomProcessor  # noqa E402
from test_module.custom_tokenization import CustomTokenizer  # noqa E402


__snake_case				=				get_tests_dir('''fixtures/dummy_feature_extractor_config.json''')
__snake_case				=				get_tests_dir('''fixtures/vocab.json''')
__snake_case				=				get_tests_dir('''fixtures''')


class 			__lowerCamelCase     (    unittest.TestCase ):

     '''simple docstring'''
     A_				:		str			=						["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]


     def 							_UpperCAmelCase					(					self							)				->							List[Any]:
          _a					= 0


     def 							_UpperCAmelCase					(					self							)				->							List[Any]:
          _a					= AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''							)
          self.assertIsInstance(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)


     def 							_UpperCAmelCase					(					self							)				->							Union[str, Any]:
          with tempfile.TemporaryDirectory() as tmpdirname:
               _a					= WavaVecaConfig()
               _a					= AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''							)

               # save in new folder
               model_config.save_pretrained(SCREAMING_SNAKE_CASE__							)
               processor.save_pretrained(SCREAMING_SNAKE_CASE__							)

               _a					= AutoProcessor.from_pretrained(SCREAMING_SNAKE_CASE__							)

          self.assertIsInstance(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)


     def 							_UpperCAmelCase					(					self							)				->							Any:
          with tempfile.TemporaryDirectory() as tmpdirname:
               # copy relevant files
               copyfile(SCREAMING_SNAKE_CASE__			, os.path.join(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)							)
               copyfile(SCREAMING_SNAKE_CASE__			, os.path.join(SCREAMING_SNAKE_CASE__			, '''vocab.json'''							)							)

               _a					= AutoProcessor.from_pretrained(SCREAMING_SNAKE_CASE__							)

          self.assertIsInstance(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)


     def 							_UpperCAmelCase					(					self							)				->							Optional[Any]:
          with tempfile.TemporaryDirectory() as tmpdirname:
               _a					= WavaVecaFeatureExtractor()
               _a					= AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''							)

               _a					= WavaVecaProcessor(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)

               # save in new folder
               processor.save_pretrained(SCREAMING_SNAKE_CASE__							)

               # drop `processor_class` in tokenizer
               with open(os.path.join(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)			, '''r'''							) as f:
                    _a					= json.load(SCREAMING_SNAKE_CASE__							)
                    config_dict.pop('''processor_class'''							)

               with open(os.path.join(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)			, '''w'''							) as f:
                    f.write(json.dumps(SCREAMING_SNAKE_CASE__							)							)

               _a					= AutoProcessor.from_pretrained(SCREAMING_SNAKE_CASE__							)

          self.assertIsInstance(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)


     def 							_UpperCAmelCase					(					self							)				->							str:
          with tempfile.TemporaryDirectory() as tmpdirname:
               _a					= WavaVecaFeatureExtractor()
               _a					= AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''							)

               _a					= WavaVecaProcessor(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)

               # save in new folder
               processor.save_pretrained(SCREAMING_SNAKE_CASE__							)

               # drop `processor_class` in feature extractor
               with open(os.path.join(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)			, '''r'''							) as f:
                    _a					= json.load(SCREAMING_SNAKE_CASE__							)
                    config_dict.pop('''processor_class'''							)

               with open(os.path.join(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)			, '''w'''							) as f:
                    f.write(json.dumps(SCREAMING_SNAKE_CASE__							)							)

               _a					= AutoProcessor.from_pretrained(SCREAMING_SNAKE_CASE__							)

          self.assertIsInstance(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)


     def 							_UpperCAmelCase					(					self							)				->							Dict:
          with tempfile.TemporaryDirectory() as tmpdirname:
               _a					= WavaVecaConfig(processor_class='''Wav2Vec2Processor'''							)
               model_config.save_pretrained(SCREAMING_SNAKE_CASE__							)
               # copy relevant files
               copyfile(SCREAMING_SNAKE_CASE__			, os.path.join(SCREAMING_SNAKE_CASE__			, '''vocab.json'''							)							)
               # create emtpy sample processor
               with open(os.path.join(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)			, '''w'''							) as f:
                    f.write('''{}'''							)

               _a					= AutoProcessor.from_pretrained(SCREAMING_SNAKE_CASE__							)

          self.assertIsInstance(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)


     def 							_UpperCAmelCase					(					self							)				->							Any:
          # If remote code is not set, we will time out when asking whether to load the model.
          with self.assertRaises(SCREAMING_SNAKE_CASE__							):
               _a					= AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor'''							)
          # If remote code is disabled, we can't load this config.
          with self.assertRaises(SCREAMING_SNAKE_CASE__							):
               _a					= AutoProcessor.from_pretrained(
                   '''hf-internal-testing/test_dynamic_processor'''			, trust_remote_code=SCREAMING_SNAKE_CASE__							)

          _a					= AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor'''			, trust_remote_code=SCREAMING_SNAKE_CASE__							)
          self.assertTrue(processor.special_attribute_present							)
          self.assertEqual(processor.__class__.__name__			, '''NewProcessor'''							)

          _a					= processor.feature_extractor
          self.assertTrue(feature_extractor.special_attribute_present							)
          self.assertEqual(feature_extractor.__class__.__name__			, '''NewFeatureExtractor'''							)

          _a					= processor.tokenizer
          self.assertTrue(tokenizer.special_attribute_present							)
          if is_tokenizers_available():
               self.assertEqual(tokenizer.__class__.__name__			, '''NewTokenizerFast'''							)

               # Test we can also load the slow version
               _a					= AutoProcessor.from_pretrained(
                   '''hf-internal-testing/test_dynamic_processor'''			, trust_remote_code=SCREAMING_SNAKE_CASE__			, use_fast=SCREAMING_SNAKE_CASE__							)
               _a					= new_processor.tokenizer
               self.assertTrue(new_tokenizer.special_attribute_present							)
               self.assertEqual(new_tokenizer.__class__.__name__			, '''NewTokenizer'''							)
          else:
               self.assertEqual(tokenizer.__class__.__name__			, '''NewTokenizer'''							)


     def 							_UpperCAmelCase					(					self							)				->							Tuple:
          try:
               AutoConfig.register('''custom'''			, SCREAMING_SNAKE_CASE__							)
               AutoFeatureExtractor.register(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)
               AutoTokenizer.register(SCREAMING_SNAKE_CASE__			, slow_tokenizer_class=SCREAMING_SNAKE_CASE__							)
               AutoProcessor.register(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)
               # Trying to register something existing in the Transformers library will raise an error
               with self.assertRaises(SCREAMING_SNAKE_CASE__							):
                    AutoProcessor.register(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)

               # Now that the config is registered, it can be used as any other config with the auto-API
               _a					= CustomFeatureExtractor.from_pretrained(SCREAMING_SNAKE_CASE__							)

               with tempfile.TemporaryDirectory() as tmp_dir:
                    _a					= os.path.join(SCREAMING_SNAKE_CASE__			, '''vocab.txt'''							)
                    with open(SCREAMING_SNAKE_CASE__			, '''w'''			, encoding='''utf-8'''							) as vocab_writer:
                         vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]							)							)
                    _a					= CustomTokenizer(SCREAMING_SNAKE_CASE__							)

               _a					= CustomProcessor(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)

               with tempfile.TemporaryDirectory() as tmp_dir:
                    processor.save_pretrained(SCREAMING_SNAKE_CASE__							)
                    _a					= AutoProcessor.from_pretrained(SCREAMING_SNAKE_CASE__							)
                    self.assertIsInstance(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)

          finally:
               if "custom" in CONFIG_MAPPING._extra_content:
                    del CONFIG_MAPPING._extra_content["custom"]
               if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
                    del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]
               if CustomConfig in TOKENIZER_MAPPING._extra_content:
                    del TOKENIZER_MAPPING._extra_content[CustomConfig]
               if CustomConfig in PROCESSOR_MAPPING._extra_content:
                    del PROCESSOR_MAPPING._extra_content[CustomConfig]


     def 							_UpperCAmelCase					(					self							)				->							Dict:


          class 			__lowerCamelCase     (    __lowercase ):

               '''simple docstring'''
               A_				:		List[Any]			=						False







          class 			__lowerCamelCase     (    __lowercase ):

               '''simple docstring'''
               A_				:		List[str]			=						False







          class 			__lowerCamelCase     (    __lowercase ):

               '''simple docstring'''
               A_				:		Union[str, Any]			=						"""AutoFeatureExtractor"""
               A_				:		int			=						"""AutoTokenizer"""
               A_				:		int			=						False

          try:
               AutoConfig.register('''custom'''			, SCREAMING_SNAKE_CASE__							)
               AutoFeatureExtractor.register(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)
               AutoTokenizer.register(SCREAMING_SNAKE_CASE__			, slow_tokenizer_class=SCREAMING_SNAKE_CASE__							)
               AutoProcessor.register(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)
               # If remote code is not set, the default is to use local classes.
               _a					= AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor'''							)
               self.assertEqual(processor.__class__.__name__			, '''NewProcessor'''							)
               self.assertFalse(processor.special_attribute_present							)
               self.assertFalse(processor.feature_extractor.special_attribute_present							)
               self.assertFalse(processor.tokenizer.special_attribute_present							)

               # If remote code is disabled, we load the local ones.
               _a					= AutoProcessor.from_pretrained(
                   '''hf-internal-testing/test_dynamic_processor'''			, trust_remote_code=SCREAMING_SNAKE_CASE__							)
               self.assertEqual(processor.__class__.__name__			, '''NewProcessor'''							)
               self.assertFalse(processor.special_attribute_present							)
               self.assertFalse(processor.feature_extractor.special_attribute_present							)
               self.assertFalse(processor.tokenizer.special_attribute_present							)

               # If remote is enabled, we load from the Hub.
               _a					= AutoProcessor.from_pretrained(
                   '''hf-internal-testing/test_dynamic_processor'''			, trust_remote_code=SCREAMING_SNAKE_CASE__							)
               self.assertEqual(processor.__class__.__name__			, '''NewProcessor'''							)
               self.assertTrue(processor.special_attribute_present							)
               self.assertTrue(processor.feature_extractor.special_attribute_present							)
               self.assertTrue(processor.tokenizer.special_attribute_present							)

          finally:
               if "custom" in CONFIG_MAPPING._extra_content:
                    del CONFIG_MAPPING._extra_content["custom"]
               if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
                    del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]
               if CustomConfig in TOKENIZER_MAPPING._extra_content:
                    del TOKENIZER_MAPPING._extra_content[CustomConfig]
               if CustomConfig in PROCESSOR_MAPPING._extra_content:
                    del PROCESSOR_MAPPING._extra_content[CustomConfig]


     def 							_UpperCAmelCase					(					self							)				->							Any:
          _a					= AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-bert'''							)
          self.assertEqual(processor.__class__.__name__			, '''BertTokenizerFast'''							)





     def 							_UpperCAmelCase					(					self							)				->							Dict:
          _a					= AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-convnext'''							)
          self.assertEqual(processor.__class__.__name__			, '''ConvNextImageProcessor'''							)







@is_staging_test
class 			__lowerCamelCase     (    unittest.TestCase ):

     '''simple docstring'''
     A_				:		Dict			=						["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]


     @classmethod
     def 							_UpperCAmelCase					(					cls							)				->							Dict:
          _a					= TOKEN
          HfFolder.save_token(SCREAMING_SNAKE_CASE__							)


     @classmethod
     def 							_UpperCAmelCase					(					cls							)				->							Union[str, Any]:
          try:
               delete_repo(token=cls._token			, repo_id='''test-processor'''							)
          except HTTPError:
               pass

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

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


     def 							_UpperCAmelCase					(					self							)				->							Union[str, Any]:
          _a					= WavaVecaProcessor.from_pretrained(SCREAMING_SNAKE_CASE__							)
          with tempfile.TemporaryDirectory() as tmp_dir:
               processor.save_pretrained(
                   os.path.join(SCREAMING_SNAKE_CASE__			, '''test-processor'''							)			, push_to_hub=SCREAMING_SNAKE_CASE__			, use_auth_token=self._token							)

               _a					= WavaVecaProcessor.from_pretrained(F'{USER}/test-processor'							)
               for k, v in processor.feature_extractor.__dict__.items():
                    self.assertEqual(SCREAMING_SNAKE_CASE__			, getattr(new_processor.feature_extractor			, SCREAMING_SNAKE_CASE__							)							)
               self.assertDictEqual(new_processor.tokenizer.get_vocab()			, processor.tokenizer.get_vocab()							)


     def 							_UpperCAmelCase					(					self							)				->							Dict:
          _a					= WavaVecaProcessor.from_pretrained(SCREAMING_SNAKE_CASE__							)

          with tempfile.TemporaryDirectory() as tmp_dir:
               processor.save_pretrained(
                   os.path.join(SCREAMING_SNAKE_CASE__			, '''test-processor-org'''							)			, push_to_hub=SCREAMING_SNAKE_CASE__			, use_auth_token=self._token			, organization='''valid_org'''			, )

               _a					= WavaVecaProcessor.from_pretrained('''valid_org/test-processor-org'''							)
               for k, v in processor.feature_extractor.__dict__.items():
                    self.assertEqual(SCREAMING_SNAKE_CASE__			, getattr(new_processor.feature_extractor			, SCREAMING_SNAKE_CASE__							)							)
               self.assertDictEqual(new_processor.tokenizer.get_vocab()			, processor.tokenizer.get_vocab()							)





     def 							_UpperCAmelCase					(					self							)				->							str:
          CustomFeatureExtractor.register_for_auto_class()
          CustomTokenizer.register_for_auto_class()
          CustomProcessor.register_for_auto_class()

          _a					= CustomFeatureExtractor.from_pretrained(SCREAMING_SNAKE_CASE__							)

          with tempfile.TemporaryDirectory() as tmp_dir:
               _a					= os.path.join(SCREAMING_SNAKE_CASE__			, '''vocab.txt'''							)
               with open(SCREAMING_SNAKE_CASE__			, '''w'''			, encoding='''utf-8'''							) as vocab_writer:
                    vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]							)							)
               _a					= CustomTokenizer(SCREAMING_SNAKE_CASE__							)

          _a					= CustomProcessor(SCREAMING_SNAKE_CASE__			, SCREAMING_SNAKE_CASE__							)

          with tempfile.TemporaryDirectory() as tmp_dir:
               create_repo(F'{USER}/test-dynamic-processor'			, token=self._token							)
               _a					= Repository(SCREAMING_SNAKE_CASE__			, clone_from=F'{USER}/test-dynamic-processor'			, token=self._token							)
               processor.save_pretrained(SCREAMING_SNAKE_CASE__							)

               # This has added the proper auto_map field to the feature extractor config
               self.assertDictEqual(
                   processor.feature_extractor.auto_map			, {
                       '''AutoFeatureExtractor''': '''custom_feature_extraction.CustomFeatureExtractor''',
                       '''AutoProcessor''': '''custom_processing.CustomProcessor''',
                   }			, )

               # This has added the proper auto_map field to the tokenizer config
               with open(os.path.join(SCREAMING_SNAKE_CASE__			, '''tokenizer_config.json'''							)							) as f:
                    _a					= json.load(SCREAMING_SNAKE_CASE__							)
               self.assertDictEqual(
                   tokenizer_config['''auto_map''']			, {
                       '''AutoTokenizer''': ['''custom_tokenization.CustomTokenizer''', None],
                       '''AutoProcessor''': '''custom_processing.CustomProcessor''',
                   }			, )

               # The code has been copied from fixtures
               self.assertTrue(os.path.isfile(os.path.join(SCREAMING_SNAKE_CASE__			, '''custom_feature_extraction.py'''							)							)							)
               self.assertTrue(os.path.isfile(os.path.join(SCREAMING_SNAKE_CASE__			, '''custom_tokenization.py'''							)							)							)
               self.assertTrue(os.path.isfile(os.path.join(SCREAMING_SNAKE_CASE__			, '''custom_processing.py'''							)							)							)

               repo.push_to_hub()

          _a					= AutoProcessor.from_pretrained(F'{USER}/test-dynamic-processor'			, trust_remote_code=SCREAMING_SNAKE_CASE__							)
          # Can't make an isinstance check because the new_processor is from the CustomProcessor class of a dynamic module
          self.assertEqual(new_processor.__class__.__name__			, '''CustomProcessor'''							)

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






		def  __A       (		self		:							Optional[int]  )     ->							List[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  __A       (		self		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							List[Any]  )     ->							Optional[Any]:
						# configuration for running training on smdistributed Model Parallel
						__lowerCamelCase    					=	{
						    '''enabled''': True,
						    '''processes_per_host''': 8,
						}
						__lowerCamelCase    					=	{
						    '''enabled''': True,
						    '''parameters''': {
						        '''microbatches''': 4,
						        '''placement_strategy''': '''spread''',
						        '''pipeline''': '''interleaved''',
						        '''optimize''': '''speed''',
						        '''partitions''': 4,
						        '''ddp''': True,
						    },
						}

						__lowerCamelCase    					=	{'''smdistributed''': {'''modelparallel''': smp_options}, '''mpi''': mpi_options}

						__lowerCamelCase    					=	'''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=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,
						        '''max_steps''': 5_00,
						    }							,			metric_definitions=self.env.metric_definitions							,			distribution=SCREAMING_SNAKE_CASE__							,			py_version='''py36'''							,			)






		def  __A       (		self		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							List[Any]  )     ->							List[str]:
						TrainingJobAnalytics(SCREAMING_SNAKE_CASE__  ).export_csv(f'''{self.env.test_path}/{job_name}_metrics.csv'''  )







		@parameterized.expand([(1,)]  )
		def  __A       (		self		:							Optional[Any]							,			SCREAMING_SNAKE_CASE__		:							Optional[Any]  )     ->							Optional[int]:
						# create estimator
						__lowerCamelCase    					=	self.create_estimator(SCREAMING_SNAKE_CASE__  )

						# run training
						estimator.fit()

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

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



