





import time
from contextlib import contextmanager
from pathlib import Path

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


UpperCAmelCase__						:			List[Any]									="__DUMMY_TRANSFORMERS_USER__"
UpperCAmelCase__						:			str									="Dummy User"
UpperCAmelCase__						:			str									="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"

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





@pytest.fixture
def      _lowercase					(						_UpperCAmelCase				)						->  Optional[Any]:
				monkeypatch.setattr(
				    """huggingface_hub.file_download.HUGGINGFACE_CO_URL_TEMPLATE"""			,      lowerCamelCase__				)





@pytest.fixture
def      _lowercase					(						_UpperCAmelCase				)						->  int:
				monkeypatch.setattr("""datasets.config.HF_ENDPOINT"""			,      lowerCamelCase__				)
				monkeypatch.setattr("""datasets.config.HUB_DATASETS_URL"""			,      lowerCamelCase__				)





@pytest.fixture
def      _lowercase					(						_UpperCAmelCase				)						->  Optional[Any]:
				monkeypatch.setattr("""huggingface_hub.hf_api.HfFolder.path_token"""			,      lowerCamelCase__				)





@pytest.fixture
def      _lowercase					(						_UpperCAmelCase			,      _UpperCAmelCase				)						->  Dict:
				HfFolder.save_token(lowerCamelCase__				)
				yield
				HfFolder.delete_token()





@pytest.fixture(scope="""session"""				)
def      _lowercase					(						)						->  Union[str, Any]:
				return HfApi(endpoint=lowerCamelCase__				)





@pytest.fixture(scope="""session"""				)
def      _lowercase					(						_UpperCAmelCase				)						->  Tuple:
				lowerCamelCase	 =HfFolder.get_token()
				HfFolder.save_token(lowerCamelCase__				)
				yield CI_HUB_USER_TOKEN
				if previous_token is not None:
								HfFolder.save_token(lowerCamelCase__				)





@pytest.fixture
def      _lowercase					(						_UpperCAmelCase				)						->  Optional[int]:


				def _cleanup_repo(_UpperCAmelCase				):
								hf_api.delete_repo(lowerCamelCase__			,      token=lowerCamelCase__			,      repo_type="""dataset"""				)

				return _cleanup_repo





@pytest.fixture
def      _lowercase					(						_UpperCAmelCase				)						->  Optional[int]:


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

				return _temporary_repo





@pytest.fixture(scope="""session"""				)
def      _lowercase					(						_UpperCAmelCase			,      _UpperCAmelCase			,      _UpperCAmelCase				)						->  Any:
				lowerCamelCase	 =F"""repo_txt_data-{int(time.time() * 1_0e3				)}"""
				lowerCamelCase	 =F"""{CI_HUB_USER}/{repo_name}"""
				hf_api.create_repo(lowerCamelCase__			,      token=lowerCamelCase__			,      repo_type="""dataset"""			,      private=lowerCamelCase__				)
				hf_api.upload_file(
				    token=lowerCamelCase__			,      path_or_fileobj=str(lowerCamelCase__				)			,      path_in_repo="""data/text_data.txt"""			,      repo_id=lowerCamelCase__			,      repo_type="""dataset"""			,      )
				yield repo_id
				try:
								hf_api.delete_repo(lowerCamelCase__			,      token=lowerCamelCase__			,      repo_type="""dataset"""				)
				except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
								pass





@pytest.fixture()
def      _lowercase					(						_UpperCAmelCase			,      _UpperCAmelCase			,      _UpperCAmelCase				)						->  int:
				return hf_private_dataset_repo_txt_data_





@pytest.fixture(scope="""session"""				)
def      _lowercase					(						_UpperCAmelCase			,      _UpperCAmelCase			,      _UpperCAmelCase				)						->  str:
				lowerCamelCase	 =F"""repo_zipped_txt_data-{int(time.time() * 1_0e3				)}"""
				lowerCamelCase	 =F"""{CI_HUB_USER}/{repo_name}"""
				hf_api.create_repo(lowerCamelCase__			,      token=lowerCamelCase__			,      repo_type="""dataset"""			,      private=lowerCamelCase__				)
				hf_api.upload_file(
				    token=lowerCamelCase__			,      path_or_fileobj=str(lowerCamelCase__				)			,      path_in_repo="""data.zip"""			,      repo_id=lowerCamelCase__			,      repo_type="""dataset"""			,      )
				yield repo_id
				try:
								hf_api.delete_repo(lowerCamelCase__			,      token=lowerCamelCase__			,      repo_type="""dataset"""				)
				except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
								pass





@pytest.fixture()
def      _lowercase					(						_UpperCAmelCase			,      _UpperCAmelCase			,      _UpperCAmelCase				)						->  Any:
				return hf_private_dataset_repo_zipped_txt_data_





@pytest.fixture(scope="""session"""				)
def      _lowercase					(						_UpperCAmelCase			,      _UpperCAmelCase			,      _UpperCAmelCase				)						->  List[str]:
				lowerCamelCase	 =F"""repo_zipped_img_data-{int(time.time() * 1_0e3				)}"""
				lowerCamelCase	 =F"""{CI_HUB_USER}/{repo_name}"""
				hf_api.create_repo(lowerCamelCase__			,      token=lowerCamelCase__			,      repo_type="""dataset"""			,      private=lowerCamelCase__				)
				hf_api.upload_file(
				    token=lowerCamelCase__			,      path_or_fileobj=str(lowerCamelCase__				)			,      path_in_repo="""data.zip"""			,      repo_id=lowerCamelCase__			,      repo_type="""dataset"""			,      )
				yield repo_id
				try:
								hf_api.delete_repo(lowerCamelCase__			,      token=lowerCamelCase__			,      repo_type="""dataset"""				)
				except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
								pass





@pytest.fixture()
def      _lowercase					(						_UpperCAmelCase			,      _UpperCAmelCase			,      _UpperCAmelCase				)						->  List[str]:
				return hf_private_dataset_repo_zipped_img_data_












from tempfile import TemporaryDirectory
from unittest import TestCase
from unittest.mock import MagicMock, patch

from transformers import AutoModel, TFAutoModel
from transformers.onnx import FeaturesManager
from transformers.testing_utils import SMALL_MODEL_IDENTIFIER, require_tf, require_torch

@require_torch
@require_tf
class       __A     (						a    ):


				def _snake_case       (		self							):
								lowerCamelCase	 =SMALL_MODEL_IDENTIFIER
								lowerCamelCase	 ="""pt"""
								lowerCamelCase	 ="""tf"""


				def _snake_case       (		self   , UpperCAmelCase_							):
								lowerCamelCase	 =AutoModel.from_pretrained(self.test_model							)
								model_pt.save_pretrained(UpperCAmelCase_							)


				def _snake_case       (		self   , UpperCAmelCase_							):
								lowerCamelCase	 =TFAutoModel.from_pretrained(self.test_model   , from_pt=UpperCAmelCase_							)
								model_tf.save_pretrained(UpperCAmelCase_							)


				def _snake_case       (		self							):
								lowerCamelCase	 ="""mock_framework"""

								# Framework provided - return whatever the user provides
								lowerCamelCase	 =FeaturesManager.determine_framework(self.test_model   , UpperCAmelCase_							)
								self.assertEqual(UpperCAmelCase_   , UpperCAmelCase_							)

								# Local checkpoint and framework provided - return provided framework
								# PyTorch checkpoint
								with TemporaryDirectory() as local_pt_ckpt:
												self._setup_pt_ckpt(UpperCAmelCase_							)
												lowerCamelCase	 =FeaturesManager.determine_framework(UpperCAmelCase_   , UpperCAmelCase_							)
												self.assertEqual(UpperCAmelCase_   , UpperCAmelCase_							)

								# TensorFlow checkpoint
								with TemporaryDirectory() as local_tf_ckpt:
												self._setup_tf_ckpt(UpperCAmelCase_							)
												lowerCamelCase	 =FeaturesManager.determine_framework(UpperCAmelCase_   , UpperCAmelCase_							)
												self.assertEqual(UpperCAmelCase_   , UpperCAmelCase_							)


				def _snake_case       (		self							):
								# PyTorch checkpoint
								with TemporaryDirectory() as local_pt_ckpt:
												self._setup_pt_ckpt(UpperCAmelCase_							)
												lowerCamelCase	 =FeaturesManager.determine_framework(UpperCAmelCase_							)
												self.assertEqual(UpperCAmelCase_   , self.framework_pt							)

								# TensorFlow checkpoint
								with TemporaryDirectory() as local_tf_ckpt:
												self._setup_tf_ckpt(UpperCAmelCase_							)
												lowerCamelCase	 =FeaturesManager.determine_framework(UpperCAmelCase_							)
												self.assertEqual(UpperCAmelCase_   , self.framework_tf							)

								# Invalid local checkpoint
								with TemporaryDirectory() as local_invalid_ckpt:
												with self.assertRaises(UpperCAmelCase_							):
																lowerCamelCase	 =FeaturesManager.determine_framework(UpperCAmelCase_							)


				def _snake_case       (		self							):
								lowerCamelCase	 =MagicMock(return_value=UpperCAmelCase_							)
								with patch("""transformers.onnx.features.is_tf_available"""   , UpperCAmelCase_							):
												lowerCamelCase	 =FeaturesManager.determine_framework(self.test_model							)
												self.assertEqual(UpperCAmelCase_   , self.framework_pt							)

								# PyTorch not in environment -> use TensorFlow
								lowerCamelCase	 =MagicMock(return_value=UpperCAmelCase_							)
								with patch("""transformers.onnx.features.is_torch_available"""   , UpperCAmelCase_							):
												lowerCamelCase	 =FeaturesManager.determine_framework(self.test_model							)
												self.assertEqual(UpperCAmelCase_   , self.framework_tf							)

								# Both in environment -> use PyTorch
								lowerCamelCase	 =MagicMock(return_value=UpperCAmelCase_							)
								lowerCamelCase	 =MagicMock(return_value=UpperCAmelCase_							)
								with patch("""transformers.onnx.features.is_tf_available"""   , UpperCAmelCase_							), patch(
								    """transformers.onnx.features.is_torch_available"""   , UpperCAmelCase_							):
												lowerCamelCase	 =FeaturesManager.determine_framework(self.test_model							)
												self.assertEqual(UpperCAmelCase_   , self.framework_pt							)

								# Both not in environment -> raise error
								lowerCamelCase	 =MagicMock(return_value=UpperCAmelCase_							)
								lowerCamelCase	 =MagicMock(return_value=UpperCAmelCase_							)
								with patch("""transformers.onnx.features.is_tf_available"""   , UpperCAmelCase_							), patch(
								    """transformers.onnx.features.is_torch_available"""   , UpperCAmelCase_							):
												with self.assertRaises(UpperCAmelCase_							):
																lowerCamelCase	 =FeaturesManager.determine_framework(self.test_model							)





