






'''simple docstring'''



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,
)


SCREAMING_SNAKE_CASE						=	'\\n    Text data.\n    Second line of data.'

SCREAMING_SNAKE_CASE						=	'file'


@pytest.fixture(scope='''session'''			)
def 							lowercase_				(    __A :       Optional[int]			)       ->      Dict:



 """simple docstring"""




 lowercase					:   Union[str, Any]				  =tmp_path_factory.mktemp('''data'''			) / (FILE_PATH + '''.zstd''')
 lowercase					:   Dict				  =bytes(__A					,      '''utf-8'''			)
 with zstd.open(__A					,      '''wb'''			) as f:
  f.write(__A			)
 return path


@pytest.fixture
def 							lowercase_				(    __A :       Any			)       ->      int:



 """simple docstring"""




 with open(os.path.join(tmpfs.local_root_dir					,      __A			)					,      '''w'''			) as f:
  f.write(__A			)
 return FILE_PATH


@pytest.mark.parametrize('''compression_format'''					,      ['''gzip''', '''xz''', '''zstd''']			)
def 							lowercase_				(    __A :       int					,      __A :       int					,      __A :       Any					,      __A :       List[Any]					,      __A :       Dict					,      __A :       Tuple			)       ->      Dict:



 """simple docstring"""




 lowercase					:   Optional[Any]				  ={'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_path}
 lowercase					:   List[Any]				  =input_paths[compression_format]
 lowercase					:   Optional[Any]				  =tmp_path / '''cache'''
 lowercase					:   List[Any]				  =DownloadConfig(cache_dir=__A					,      extract_compressed_file=__A			)
 lowercase					:   List[str]				  =cached_path(__A					,      download_config=__A			)
 with open(__A			) as f:
  lowercase					:   Any				  =f.read()
 with open(__A			) as f:
  lowercase					:   int				  =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 							lowercase_				(    __A :       Union[str, Any]					,      __A :       List[Any]					,      __A :       List[Any]					,      __A :       Optional[Any]					,      __A :       Any			)       ->      Optional[Any]:



 """simple docstring"""




 lowercase					:   Dict				  ='''custom_cache'''
 lowercase					:   int				  ='''custom_extracted_dir'''
 lowercase					:   Optional[Any]				  =tmp_path / '''custom_extracted_path'''
 if default_extracted:
  lowercase					:   Tuple				  =('''downloads''' if default_cache_dir else custom_cache_dir, '''extracted''')
 else:
  monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_DIR'''					,      __A			)
  monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_PATH'''					,      str(__A			)			)
  lowercase					:   List[Any]				  =custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

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


def 							lowercase_				(    __A :       int			)       ->      Any:



 """simple docstring"""




 lowercase					:   Tuple				  =str(Path(__A			).resolve()			)
 assert cached_path(__A			) == text_file
 # relative path
 lowercase					:   Optional[Any]				  =str(Path(__A			).resolve().relative_to(Path(os.getcwd()			)			)			)
 assert cached_path(__A			) == text_file


def 							lowercase_				(    __A :       Union[str, Any]			)       ->      Optional[Any]:



 """simple docstring"""




 lowercase					:   Optional[int]				  =str(tmp_path.resolve() / '''__missing_file__.txt'''			)
 with pytest.raises(__A			):
  cached_path(__A			)
 # relative path
 lowercase					:   List[str]				  ='''./__missing_file__.txt'''
 with pytest.raises(__A			):
  cached_path(__A			)


def 							lowercase_				(    __A :       Optional[Any]			)       ->      Tuple:



 """simple docstring"""




 lowercase					:   Union[str, Any]				  =get_from_cache(F'tmp://{tmpfs_file}'			)
 with open(__A			) as f:
  lowercase					:   Optional[int]				  =f.read()
 assert output_file_content == FILE_CONTENT


@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,      __A			)
def 							lowercase_				(    )       ->      Dict:



 """simple docstring"""




 with pytest.raises(__A			):
  cached_path('''https://huggingface.co'''			)


@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,      __A			)
def 							lowercase_				(    __A :       Tuple			)       ->      Optional[Any]:



 """simple docstring"""




 lowercase					:   Optional[int]				  =tmp_path_factory.mktemp('''data'''			) / '''file.html'''
 with pytest.raises(__A			):
  http_get('''https://huggingface.co'''					,      temp_file=__A			)
 with pytest.raises(__A			):
  http_head('''https://huggingface.co'''			)


@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,      __A			)
def 							lowercase_				(    __A :       Optional[int]			)       ->      Optional[Any]:



 """simple docstring"""




 lowercase					:   str				  =tmp_path_factory.mktemp('''data'''			) / '''file.html'''
 with pytest.raises(__A			):
  ftp_get('''ftp://huggingface.co'''					,      temp_file=__A			)
 with pytest.raises(__A			):
  ftp_head('''ftp://huggingface.co'''			)





@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,      __A			)
def 							lowercase_				(    __A :       List[str]			)       ->      Any:



 """simple docstring"""




 lowercase					:   Tuple				  =tmp_path_factory.mktemp('''data'''			) / '''file.html'''
 with pytest.raises(__A			):
  fsspec_get('''s3://huggingface.co'''					,      temp_file=__A			)
 with pytest.raises(__A			):
  fsspec_head('''s3://huggingface.co'''			)











'''simple docstring'''



import importlib.metadata
from typing import Union

from packaging.version import Version, parse

from .constants import STR_OPERATION_TO_FUNC


SCREAMING_SNAKE_CASE						=	parse(importlib.metadata.version('torch'))


def 							lowercase_				(    __A :       Union[str, Version]					,      __A :       str					,      __A :       str			)       ->      Union[str, Any]:



 """simple docstring"""




 if operation not in STR_OPERATION_TO_FUNC.keys():
  raise ValueError(F'`operation` must be one of {list(STR_OPERATION_TO_FUNC.keys()			)}, received {operation}'			)
 lowercase					:   Any				  =STR_OPERATION_TO_FUNC[operation]
 if isinstance(__A					,      __A			):
  lowercase					:   List[Any]				  =parse(importlib.metadata.version(__A			)			)
 return operation(__A					,      parse(__A			)			)





def 							lowercase_				(    __A :       str					,      __A :       str			)       ->      Tuple:



 """simple docstring"""




 return compare_versions(__A					,      __A					,      __A			)



