import json
import os
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from requests.exceptions import HTTPError

from transformers.utils import (
    CONFIG_NAME,
    FLAX_WEIGHTS_NAME,
    TF2_WEIGHTS_NAME,
    TRANSFORMERS_CACHE,
    WEIGHTS_NAME,
    cached_file,
    get_file_from_repo,
    has_file,
)


__UpperCamelCase		:       str									=     'hf-internal-testing/tiny-random-bert'
__UpperCamelCase		:       int									=     os.path.join(TRANSFORMERS_CACHE, 'models--hf-internal-testing--tiny-random-bert')
__UpperCamelCase		:       Optional[Any]									=     '9b8c223d42b2188cb49d29af482996f9d0f3e5a6'





class 					lowercase__						(   unittest.TestCase):


						def        __A       (       self	:				Optional[Any]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    str							=							cached_file(UpperCamelCase__    ,		UpperCamelCase__						)
												# Should have downloaded the file in here
												self.assertTrue(os.path.isdir(UpperCamelCase__						)						)
												# Cache should contain at least those three subfolders:
												for subfolder in ["blobs", "refs", "snapshots"]:
																		self.assertTrue(os.path.isdir(os.path.join(UpperCamelCase__    ,		UpperCamelCase__						)						)						)
												with open(os.path.join(UpperCamelCase__    ,		'''refs'''    ,		'''main'''						)						) as f:
																		SCREAMING_SNAKE_CASE							:    Optional[Any]							=							f.read()
												self.assertEqual(UpperCamelCase__    ,		os.path.join(UpperCamelCase__    ,		'''snapshots'''    ,		UpperCamelCase__    ,		UpperCamelCase__						)						)
												self.assertTrue(os.path.isfile(UpperCamelCase__						)						)

												# File is cached at the same place the second time.
												SCREAMING_SNAKE_CASE							:    int							=							cached_file(UpperCamelCase__    ,		UpperCamelCase__						)
												self.assertEqual(UpperCamelCase__    ,		UpperCamelCase__						)

												# Using a specific revision to test the full commit hash.
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							cached_file(UpperCamelCase__    ,		UpperCamelCase__    ,		revision='''9b8c223'''						)
												self.assertEqual(UpperCamelCase__    ,		os.path.join(UpperCamelCase__    ,		'''snapshots'''    ,		UpperCamelCase__    ,		UpperCamelCase__						)						)


						def        __A       (       self	:				Any						):





												'''simple docstring'''





												with self.assertRaisesRegex(UpperCamelCase__    ,		'''is not a valid model identifier'''						):
																		SCREAMING_SNAKE_CASE							:    Dict							=							cached_file('''tiny-random-bert'''    ,		UpperCamelCase__						)

												with self.assertRaisesRegex(UpperCamelCase__    ,		'''is not a valid git identifier'''						):
																		SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							cached_file(UpperCamelCase__    ,		UpperCamelCase__    ,		revision='''aaaa'''						)

												with self.assertRaisesRegex(UpperCamelCase__    ,		'''does not appear to have a file named'''						):
																		SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							cached_file(UpperCamelCase__    ,		'''conf'''						)


						def        __A       (       self	:				Dict						):





												'''simple docstring'''





												with self.assertRaisesRegex(UpperCamelCase__    ,		'''does not appear to have a file named'''						):
																		SCREAMING_SNAKE_CASE							:    int							=							cached_file(UpperCamelCase__    ,		'''conf'''						)

												with open(os.path.join(UpperCamelCase__    ,		'''refs'''    ,		'''main'''						)						) as f:
																		SCREAMING_SNAKE_CASE							:    str							=							f.read()
												self.assertTrue(os.path.isfile(os.path.join(UpperCamelCase__    ,		'''.no_exist'''    ,		UpperCamelCase__    ,		'''conf'''						)						)						)

												SCREAMING_SNAKE_CASE							:    int							=							cached_file(UpperCamelCase__    ,		'''conf'''    ,		_raise_exceptions_for_missing_entries=UpperCamelCase__						)
												self.assertIsNone(UpperCamelCase__						)

												SCREAMING_SNAKE_CASE							:    List[Any]							=							cached_file(UpperCamelCase__    ,		'''conf'''    ,		local_files_only=UpperCamelCase__    ,		_raise_exceptions_for_missing_entries=UpperCamelCase__						)
												self.assertIsNone(UpperCamelCase__						)

												SCREAMING_SNAKE_CASE							:    Tuple							=							mock.Mock()
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							500
												SCREAMING_SNAKE_CASE							:    List[str]							=							{}
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							HTTPError
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							{}

												# Under the mock environment we get a 500 error when trying to reach the tokenizer.
												with mock.patch('''requests.Session.request'''    ,		return_value=UpperCamelCase__						) as mock_head:
																		SCREAMING_SNAKE_CASE							:    List[Any]							=							cached_file(UpperCamelCase__    ,		'''conf'''    ,		_raise_exceptions_for_connection_errors=UpperCamelCase__						)
																		self.assertIsNone(UpperCamelCase__						)
																		# This check we did call the fake head request
																		mock_head.assert_called()


						def        __A       (       self	:				List[Any]						):





												'''simple docstring'''





												self.assertTrue(has_file('''hf-internal-testing/tiny-bert-pt-only'''    ,		UpperCamelCase__						)						)
												self.assertFalse(has_file('''hf-internal-testing/tiny-bert-pt-only'''    ,		UpperCamelCase__						)						)
												self.assertFalse(has_file('''hf-internal-testing/tiny-bert-pt-only'''    ,		UpperCamelCase__						)						)


						def        __A       (       self	:				Tuple						):





												'''simple docstring'''





												self.assertIsNone(get_file_from_repo('''bert-base-cased'''    ,		'''ahah.txt'''						)						)

												# The function raises if the repository does not exist.
												with self.assertRaisesRegex(UpperCamelCase__    ,		'''is not a valid model identifier'''						):
																		get_file_from_repo('''bert-base-case'''    ,		UpperCamelCase__						)

												# The function raises if the revision does not exist.
												with self.assertRaisesRegex(UpperCamelCase__    ,		'''is not a valid git identifier'''						):
																		get_file_from_repo('''bert-base-cased'''    ,		UpperCamelCase__    ,		revision='''ahaha'''						)

												SCREAMING_SNAKE_CASE							:    Optional[int]							=							get_file_from_repo('''bert-base-cased'''    ,		UpperCamelCase__						)
												# The name is the cached name which is not very easy to test, so instead we load the content.
												SCREAMING_SNAKE_CASE							:    Tuple							=							json.loads(open(UpperCamelCase__    ,		'''r'''						).read()						)
												self.assertEqual(config['''hidden_size''']    ,		768						)



						def        __A       (       self	:				List[str]						):





												'''simple docstring'''





												with tempfile.TemporaryDirectory() as tmp_dir:
																		SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							Path(UpperCamelCase__						) / '''a.txt'''
																		filename.touch()
																		self.assertEqual(get_file_from_repo(UpperCamelCase__    ,		'''a.txt'''						)    ,		str(UpperCamelCase__						)						)

																		self.assertIsNone(get_file_from_repo(UpperCamelCase__    ,		'''b.txt'''						)						)






import os
import tempfile
import unittest
from pathlib import Path

from transformers import AutoConfig, is_tf_available
from transformers.testing_utils import require_tf


if is_tf_available():
					import tensorflow as tf

					from transformers import TensorFlowBenchmark, TensorFlowBenchmarkArguments





@require_tf
class 					lowercase__						(   unittest.TestCase):


						def        __A       (       self	:				str    ,		UpperCamelCase__	:				int						):





												'''simple docstring'''





												for model_result in results.values():
																		for batch_size, sequence_length in zip(model_result['''bs''']    ,		model_result['''ss''']						):
																								SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							model_result['''result'''][batch_size][sequence_length]
																								self.assertIsNotNone(UpperCamelCase__						)


						def        __A       (       self	:				Optional[Any]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    List[Any]							=							'''sshleifer/tiny-gpt2'''
												SCREAMING_SNAKE_CASE							:    List[Any]							=							TensorFlowBenchmarkArguments(
												    models=[MODEL_ID]    ,		training=UpperCamelCase__    ,		inference=UpperCamelCase__    ,		sequence_lengths=[8]    ,		batch_sizes=[1]    ,		eager_mode=UpperCamelCase__    ,		multi_process=UpperCamelCase__    ,		)
												SCREAMING_SNAKE_CASE							:    str							=							TensorFlowBenchmark(UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    List[Any]							=							benchmark.run()
												self.check_results_dict_not_empty(results.time_inference_result						)
												self.check_results_dict_not_empty(results.memory_inference_result						)


						def        __A       (       self	:				List[str]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    str							=							'''sgugger/tiny-distilbert-classification'''
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							TensorFlowBenchmarkArguments(
												    models=[MODEL_ID]    ,		training=UpperCamelCase__    ,		inference=UpperCamelCase__    ,		sequence_lengths=[8]    ,		batch_sizes=[1]    ,		multi_process=UpperCamelCase__    ,		only_pretrain_model=UpperCamelCase__    ,		)
												SCREAMING_SNAKE_CASE							:    List[str]							=							TensorFlowBenchmark(UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							benchmark.run()
												self.check_results_dict_not_empty(results.time_inference_result						)
												self.check_results_dict_not_empty(results.memory_inference_result						)


						def        __A       (       self	:				Any						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							'''sshleifer/tiny-gpt2'''
												SCREAMING_SNAKE_CASE							:    int							=							TensorFlowBenchmarkArguments(
												    models=[MODEL_ID]    ,		training=UpperCamelCase__    ,		inference=UpperCamelCase__    ,		sequence_lengths=[8]    ,		batch_sizes=[1]    ,		multi_process=UpperCamelCase__    ,		)
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							TensorFlowBenchmark(UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    int							=							benchmark.run()
												self.check_results_dict_not_empty(results.time_inference_result						)
												self.check_results_dict_not_empty(results.memory_inference_result						)


						def        __A       (       self	:				Optional[Any]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Optional[int]							=							'''sshleifer/tiny-gpt2'''
												SCREAMING_SNAKE_CASE							:    int							=							AutoConfig.from_pretrained(UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							TensorFlowBenchmarkArguments(
												    models=[MODEL_ID]    ,		training=UpperCamelCase__    ,		inference=UpperCamelCase__    ,		sequence_lengths=[8]    ,		batch_sizes=[1]    ,		eager_mode=UpperCamelCase__    ,		multi_process=UpperCamelCase__    ,		)
												SCREAMING_SNAKE_CASE							:    str							=							TensorFlowBenchmark(UpperCamelCase__    ,		[config]						)
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							benchmark.run()
												self.check_results_dict_not_empty(results.time_inference_result						)
												self.check_results_dict_not_empty(results.memory_inference_result						)


						def        __A       (       self	:				int						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Dict							=							'''sshleifer/tiny-gpt2'''
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							AutoConfig.from_pretrained(UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							TensorFlowBenchmarkArguments(
												    models=[MODEL_ID]    ,		training=UpperCamelCase__    ,		inference=UpperCamelCase__    ,		sequence_lengths=[8]    ,		batch_sizes=[1]    ,		multi_process=UpperCamelCase__    ,		)
												SCREAMING_SNAKE_CASE							:    List[str]							=							TensorFlowBenchmark(UpperCamelCase__    ,		[config]						)
												SCREAMING_SNAKE_CASE							:    int							=							benchmark.run()
												self.check_results_dict_not_empty(results.time_inference_result						)
												self.check_results_dict_not_empty(results.memory_inference_result						)


						def        __A       (       self	:				List[Any]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    str							=							'''sshleifer/tiny-gpt2'''
												SCREAMING_SNAKE_CASE							:    int							=							TensorFlowBenchmarkArguments(
												    models=[MODEL_ID]    ,		training=UpperCamelCase__    ,		inference=UpperCamelCase__    ,		sequence_lengths=[8]    ,		batch_sizes=[1]    ,		multi_process=UpperCamelCase__    ,		)
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							TensorFlowBenchmark(UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    Tuple							=							benchmark.run()
												self.check_results_dict_not_empty(results.time_train_result						)
												self.check_results_dict_not_empty(results.memory_train_result						)


						def        __A       (       self	:				Optional[Any]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    List[str]							=							'''sshleifer/tiny-gpt2'''
												SCREAMING_SNAKE_CASE							:    int							=							AutoConfig.from_pretrained(UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    Tuple							=							TensorFlowBenchmarkArguments(
												    models=[MODEL_ID]    ,		training=UpperCamelCase__    ,		inference=UpperCamelCase__    ,		sequence_lengths=[8]    ,		batch_sizes=[1]    ,		multi_process=UpperCamelCase__    ,		)
												SCREAMING_SNAKE_CASE							:    Dict							=							TensorFlowBenchmark(UpperCamelCase__    ,		[config]						)
												SCREAMING_SNAKE_CASE							:    str							=							benchmark.run()
												self.check_results_dict_not_empty(results.time_train_result						)
												self.check_results_dict_not_empty(results.memory_train_result						)


						def        __A       (       self	:				Optional[int]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    List[str]							=							'''patrickvonplaten/t5-tiny-random'''
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							AutoConfig.from_pretrained(UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							TensorFlowBenchmarkArguments(
												    models=[MODEL_ID]    ,		training=UpperCamelCase__    ,		inference=UpperCamelCase__    ,		sequence_lengths=[8]    ,		batch_sizes=[1]    ,		multi_process=UpperCamelCase__    ,		)
												SCREAMING_SNAKE_CASE							:    Dict							=							TensorFlowBenchmark(UpperCamelCase__    ,		configs=[config]						)
												SCREAMING_SNAKE_CASE							:    Dict							=							benchmark.run()
												self.check_results_dict_not_empty(results.time_inference_result						)
												self.check_results_dict_not_empty(results.memory_inference_result						)


						@unittest.skipIf(is_tf_available() and len(tf.config.list_physical_devices('''GPU'''						)						) == 0    ,		'''Cannot do xla on CPU.'''						)
						def        __A       (       self	:				Optional[int]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							'''sshleifer/tiny-gpt2'''
												SCREAMING_SNAKE_CASE							:    Tuple							=							TensorFlowBenchmarkArguments(
												    models=[MODEL_ID]    ,		training=UpperCamelCase__    ,		inference=UpperCamelCase__    ,		sequence_lengths=[8]    ,		batch_sizes=[1]    ,		use_xla=UpperCamelCase__    ,		multi_process=UpperCamelCase__    ,		)
												SCREAMING_SNAKE_CASE							:    Any							=							TensorFlowBenchmark(UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    List[Any]							=							benchmark.run()
												self.check_results_dict_not_empty(results.time_inference_result						)
												self.check_results_dict_not_empty(results.memory_inference_result						)


						def        __A       (       self	:				List[str]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Optional[int]							=							'''sshleifer/tiny-gpt2'''
												with tempfile.TemporaryDirectory() as tmp_dir:
																		SCREAMING_SNAKE_CASE							:    Tuple							=							TensorFlowBenchmarkArguments(
																		    models=[MODEL_ID]    ,		inference=UpperCamelCase__    ,		save_to_csv=UpperCamelCase__    ,		sequence_lengths=[8]    ,		batch_sizes=[1]    ,		inference_time_csv_file=os.path.join(UpperCamelCase__    ,		'''inf_time.csv'''						)    ,		inference_memory_csv_file=os.path.join(UpperCamelCase__    ,		'''inf_mem.csv'''						)    ,		env_info_csv_file=os.path.join(UpperCamelCase__    ,		'''env.csv'''						)    ,		multi_process=UpperCamelCase__    ,		)
																		SCREAMING_SNAKE_CASE							:    Optional[Any]							=							TensorFlowBenchmark(UpperCamelCase__						)
																		benchmark.run()
																		self.assertTrue(Path(os.path.join(UpperCamelCase__    ,		'''inf_time.csv'''						)						).exists()						)
																		self.assertTrue(Path(os.path.join(UpperCamelCase__    ,		'''inf_mem.csv'''						)						).exists()						)
																		self.assertTrue(Path(os.path.join(UpperCamelCase__    ,		'''env.csv'''						)						).exists()						)



						def        __A       (       self	:				int						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Any							=							'''sshleifer/tiny-gpt2'''

												def _check_summary_is_not_empty(UpperCamelCase__	:				Dict						):
																		self.assertTrue(hasattr(UpperCamelCase__    ,		'''sequential'''						)						)
																		self.assertTrue(hasattr(UpperCamelCase__    ,		'''cumulative'''						)						)
																		self.assertTrue(hasattr(UpperCamelCase__    ,		'''current'''						)						)
																		self.assertTrue(hasattr(UpperCamelCase__    ,		'''total'''						)						)

												with tempfile.TemporaryDirectory() as tmp_dir:
																		SCREAMING_SNAKE_CASE							:    List[str]							=							TensorFlowBenchmarkArguments(
																		    models=[MODEL_ID]    ,		inference=UpperCamelCase__    ,		sequence_lengths=[8]    ,		batch_sizes=[1]    ,		log_filename=os.path.join(UpperCamelCase__    ,		'''log.txt'''						)    ,		log_print=UpperCamelCase__    ,		trace_memory_line_by_line=UpperCamelCase__    ,		eager_mode=UpperCamelCase__    ,		multi_process=UpperCamelCase__    ,		)
																		SCREAMING_SNAKE_CASE							:    List[Any]							=							TensorFlowBenchmark(UpperCamelCase__						)
																		SCREAMING_SNAKE_CASE							:    Dict							=							benchmark.run()
																		_check_summary_is_not_empty(result.inference_summary						)
																		self.assertTrue(Path(os.path.join(UpperCamelCase__    ,		'''log.txt'''						)						).exists()						)





