



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


a_      :      str    =     get_tests_dir('fixtures')
class    __UpperCamelCase			(       unittest.TestCase				):

 """simple docstring"""



 def 						_UpperCAmelCase       (	self      ) ->			Optional[Any]:
  # A mock response for an HTTP head request to emulate server down
  a__												= mock.Mock()
  a__												= 5_0_0
  a__												= {}
  a__												= HTTPError
  a__												= {}

  # Download this model to make sure it's in the cache.
  a__												= 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:
   a__												= ViTImageProcessor.from_pretrained('''hf-internal-testing/tiny-random-vit'''      )
   # This check we did call the fake head request
   mock_head.assert_called()

 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  # This test is for deprecated behavior and can be removed in v5
  a__												= ViTImageProcessor.from_pretrained(
      '''https://huggingface.co/hf-internal-testing/tiny-random-vit/resolve/main/preprocessor_config.json'''      )



 def 						_UpperCAmelCase       (	self      ) ->			int:
  with self.assertRaises(SCREAMING_SNAKE_CASE      ):
   # config is in subfolder, the following should not work without specifying the subfolder
   a__												= AutoImageProcessor.from_pretrained('''hf-internal-testing/stable-diffusion-all-variants'''      )

  a__												= AutoImageProcessor.from_pretrained(
      '''hf-internal-testing/stable-diffusion-all-variants'''		,				subfolder='''feature_extractor'''      )

  self.assertIsNotNone(SCREAMING_SNAKE_CASE      )





@is_staging_test
class    __UpperCamelCase			(       unittest.TestCase				):

 """simple docstring"""



 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			Tuple:
  a__												= TOKEN
  HfFolder.save_token(SCREAMING_SNAKE_CASE      )

 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			Union[str, Any]:
  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 						_UpperCAmelCase       (	self      ) ->			Optional[Any]:
  a__												= ViTImageProcessor.from_pretrained(SCREAMING_SNAKE_CASE      )
  image_processor.push_to_hub('''test-image-processor'''		,				use_auth_token=self._token      )

  a__												= 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      )

  a__												= 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 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  a__												= ViTImageProcessor.from_pretrained(SCREAMING_SNAKE_CASE      )
  image_processor.push_to_hub('''valid_org/test-image-processor'''		,				use_auth_token=self._token      )

  a__												= 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      )

  a__												= 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 						_UpperCAmelCase       (	self      ) ->			int:
  CustomImageProcessor.register_for_auto_class()
  a__												= 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'''}		,				)

  a__												= 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 os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


a_      :      int    =     '\\n    Text data.\n    Second line of data.'

a_      :      Optional[Any]    =     'file'






@pytest.fixture(scope='''session'''       )
def     __a					(		__UpperCAmelCase       ):
 a__												= tmp_path_factory.mktemp('''data'''       ) / (FILE_PATH + '''.zstd''')
 a__												= bytes(__UpperCAmelCase							,  '''utf-8'''       )
 with zstd.open(__UpperCAmelCase							,  '''wb'''       ) as f:
  f.write(__UpperCAmelCase       )
 return path






@pytest.fixture
def     __a					(		__UpperCAmelCase       ):
 with open(os.path.join(tmpfs.local_root_dir							,  __UpperCAmelCase       )							,  '''w'''       ) as f:
  f.write(__UpperCAmelCase       )
 return FILE_PATH






@pytest.mark.parametrize('''compression_format'''							,  ['''gzip''', '''xz''', '''zstd''']       )
def     __a					(		__UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase       ):
 a__												= {'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_path}
 a__												= input_paths[compression_format]
 a__												= tmp_path / '''cache'''
 a__												= DownloadConfig(cache_dir=__UpperCAmelCase							,  extract_compressed_file=__UpperCAmelCase       )
 a__												= cached_path(__UpperCAmelCase							,  download_config=__UpperCAmelCase       )
 with open(__UpperCAmelCase       ) as f:
  a__												= f.read()
 with open(__UpperCAmelCase       ) as f:
  a__												= f.read()
 assert extracted_file_content == expected_file_content






@pytest.mark.parametrize('''default_extracted'''							,  [True, False]       )
@pytest.mark.parametrize('''default_cache_dir'''							,  [True, False]       )
def     __a					(		__UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase       ):
 a__												= '''custom_cache'''
 a__												= '''custom_extracted_dir'''
 a__												= tmp_path / '''custom_extracted_path'''
 if default_extracted:
  a__												= ('''downloads''' if default_cache_dir else custom_cache_dir, '''extracted''')
 else:
  monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_DIR'''							,  __UpperCAmelCase       )
  monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_PATH'''							,  str(__UpperCAmelCase       )       )
  a__												= custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

 a__												= xz_file
 a__												= (
     DownloadConfig(extract_compressed_file=__UpperCAmelCase       )
     if default_cache_dir
     else DownloadConfig(cache_dir=tmp_path / custom_cache_dir							,  extract_compressed_file=__UpperCAmelCase       )
 )
 a__												= cached_path(__UpperCAmelCase							,  download_config=__UpperCAmelCase       )
 assert Path(__UpperCAmelCase       ).parent.parts[-2:] == expected






def     __a					(		__UpperCAmelCase       ):
 # absolute path
 a__												= str(Path(__UpperCAmelCase       ).resolve()       )
 assert cached_path(__UpperCAmelCase       ) == text_file
 # relative path
 a__												= str(Path(__UpperCAmelCase       ).resolve().relative_to(Path(os.getcwd()       )       )       )
 assert cached_path(__UpperCAmelCase       ) == text_file






def     __a					(		__UpperCAmelCase       ):
 # absolute path
 a__												= str(tmp_path.resolve() / '''__missing_file__.txt'''       )
 with pytest.raises(__UpperCAmelCase       ):
  cached_path(__UpperCAmelCase       )
 # relative path
 a__												= '''./__missing_file__.txt'''
 with pytest.raises(__UpperCAmelCase       ):
  cached_path(__UpperCAmelCase       )






def     __a					(		__UpperCAmelCase       ):
 a__												= get_from_cache(f"tmp://{tmpfs_file}"       )
 with open(__UpperCAmelCase       ) as f:
  a__												= f.read()
 assert output_file_content == FILE_CONTENT






@patch('''datasets.config.HF_DATASETS_OFFLINE'''							,  __UpperCAmelCase       )
def     __a					(		):
 with pytest.raises(__UpperCAmelCase       ):
  cached_path('''https://huggingface.co'''       )






@patch('''datasets.config.HF_DATASETS_OFFLINE'''							,  __UpperCAmelCase       )
def     __a					(		__UpperCAmelCase       ):
 a__												= tmp_path_factory.mktemp('''data'''       ) / '''file.html'''
 with pytest.raises(__UpperCAmelCase       ):
  http_get('''https://huggingface.co'''							,  temp_file=__UpperCAmelCase       )
 with pytest.raises(__UpperCAmelCase       ):
  http_head('''https://huggingface.co'''       )






@patch('''datasets.config.HF_DATASETS_OFFLINE'''							,  __UpperCAmelCase       )
def     __a					(		__UpperCAmelCase       ):
 a__												= tmp_path_factory.mktemp('''data'''       ) / '''file.html'''
 with pytest.raises(__UpperCAmelCase       ):
  ftp_get('''ftp://huggingface.co'''							,  temp_file=__UpperCAmelCase       )
 with pytest.raises(__UpperCAmelCase       ):
  ftp_head('''ftp://huggingface.co'''       )






@patch('''datasets.config.HF_DATASETS_OFFLINE'''							,  __UpperCAmelCase       )
def     __a					(		__UpperCAmelCase       ):
 a__												= tmp_path_factory.mktemp('''data'''       ) / '''file.html'''
 with pytest.raises(__UpperCAmelCase       ):
  fsspec_get('''s3://huggingface.co'''							,  temp_file=__UpperCAmelCase       )
 with pytest.raises(__UpperCAmelCase       ):
  fsspec_head('''s3://huggingface.co'''       )





