



import os
import zipfile

import pytest

from datasets.utils.extract import (
    BzipaExtractor,
    Extractor,
    GzipExtractor,
    LzaExtractor,
    SevenZipExtractor,
    TarExtractor,
    XzExtractor,
    ZipExtractor,
    ZstdExtractor,
)

from .utils import require_lza, require_pyazr, require_zstandard





@pytest.mark.parametrize(
    '''compression_format, is_archive'''   , [
        ('''7z''', True),
        ('''bz2''', False),
        ('''gzip''', False),
        ('''lz4''', False),
        ('''tar''', True),
        ('''xz''', False),
        ('''zip''', True),
        ('''zstd''', False),
    ]   , )
def      __snake_case	(			_UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , ):
    __a						       =				{
        "7z": (seven_zip_file, SevenZipExtractor),
        "bz2": (bza_file, BzipaExtractor),
        "gzip": (gz_file, GzipExtractor),
        "lz4": (lza_file, LzaExtractor),
        "tar": (tar_file, TarExtractor),
        "xz": (xz_file, XzExtractor),
        "zip": (zip_file, ZipExtractor),
        "zstd": (zstd_file, ZstdExtractor),
    }
    __a						       =				input_paths_and_base_extractors[compression_format]
    if input_path is None:
        __a						       =				f'for \'{compression_format}\' compression_format, '
        if compression_format == "7z":
            reason += require_pyazr.kwargs["reason"]
        elif compression_format == "lz4":
            reason += require_lza.kwargs["reason"]
        elif compression_format == "zstd":
            reason += require_zstandard.kwargs["reason"]
        pytest.skip(__lowerCAmelCase				)
    assert base_extractor.is_extractable(__lowerCAmelCase				)
    __a						       =				tmp_path / ("extracted" if is_archive else "extracted.txt")
    base_extractor.extract(__lowerCAmelCase   , __lowerCAmelCase				)
    if is_archive:
        assert output_path.is_dir()
        for file_path in output_path.iterdir():
            assert file_path.name == text_file.name
            __a						       =				file_path.read_text(encoding='''utf-8'''				)
    else:
        __a						       =				output_path.read_text(encoding='''utf-8'''				)
    __a						       =				text_file.read_text(encoding='''utf-8'''				)
    assert extracted_file_content == expected_file_content





@pytest.mark.parametrize(
    '''compression_format, is_archive'''   , [
        ('''7z''', True),
        ('''bz2''', False),
        ('''gzip''', False),
        ('''lz4''', False),
        ('''tar''', True),
        ('''xz''', False),
        ('''zip''', True),
        ('''zstd''', False),
    ]   , )
def      __snake_case	(			_UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , ):
    __a						       =				{
        "7z": seven_zip_file,
        "bz2": bza_file,
        "gzip": gz_file,
        "lz4": lza_file,
        "tar": tar_file,
        "xz": xz_file,
        "zip": zip_file,
        "zstd": zstd_file,
    }
    __a						       =				input_paths[compression_format]
    if input_path is None:
        __a						       =				f'for \'{compression_format}\' compression_format, '
        if compression_format == "7z":
            reason += require_pyazr.kwargs["reason"]
        elif compression_format == "lz4":
            reason += require_lza.kwargs["reason"]
        elif compression_format == "zstd":
            reason += require_zstandard.kwargs["reason"]
        pytest.skip(__lowerCAmelCase				)
    __a						       =				Extractor.infer_extractor_format(__lowerCAmelCase				)
    assert extractor_format is not None
    __a						       =				tmp_path / ("extracted" if is_archive else "extracted.txt")
    Extractor.extract(__lowerCAmelCase   , __lowerCAmelCase   , __lowerCAmelCase				)
    if is_archive:
        assert output_path.is_dir()
        for file_path in output_path.iterdir():
            assert file_path.name == text_file.name
            __a						       =				file_path.read_text(encoding='''utf-8'''				)
    else:
        __a						       =				output_path.read_text(encoding='''utf-8'''				)
    __a						       =				text_file.read_text(encoding='''utf-8'''				)
    assert extracted_file_content == expected_file_content





@pytest.fixture
def      __snake_case	(			_UpperCAmelCase   , _UpperCAmelCase				):
    import tarfile

    __a						       =				tmp_path / "data_dot_dot"
    directory.mkdir()
    __a						       =				directory / "tar_file_with_dot_dot.tar"
    with tarfile.TarFile(__lowerCAmelCase   , '''w'''				) as f:
        f.add(__lowerCAmelCase   , arcname=os.path.join('''..'''   , text_file.name				)				)
    return path





@pytest.fixture
def      __snake_case	(			_UpperCAmelCase				):
    import tarfile

    __a						       =				tmp_path / "data_sym_link"
    directory.mkdir()
    __a						       =				directory / "tar_file_with_sym_link.tar"
    os.symlink('''..'''   , directory / '''subdir'''   , target_is_directory=__lowerCAmelCase				)
    with tarfile.TarFile(__lowerCAmelCase   , '''w'''				) as f:
        f.add(str(directory / '''subdir'''				)   , arcname='''subdir'''				)  # str required by os.readlink on Windows and Python < 3.8
    return path





@pytest.mark.parametrize(
    '''insecure_tar_file, error_log'''   , [('''tar_file_with_dot_dot''', '''illegal path'''), ('''tar_file_with_sym_link''', '''Symlink''')]   , )
def      __snake_case	(			_UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase   , _UpperCAmelCase				):
    __a						       =				{
        "tar_file_with_dot_dot": tar_file_with_dot_dot,
        "tar_file_with_sym_link": tar_file_with_sym_link,
    }
    __a						       =				insecure_tar_files[insecure_tar_file]
    __a						       =				tmp_path / "extracted"
    TarExtractor.extract(__lowerCAmelCase   , __lowerCAmelCase				)
    assert caplog.text
    for record in caplog.records:
        assert record.levelname == "ERROR"
        assert error_log in record.msg





def      __snake_case	(			_UpperCAmelCase				):
    # We should have less false positives than zipfile.is_zipfile
    # We do that by checking only the magic number
    __a						       =				tmpdir / "not_a_zip_file"
    # From: https://github.com/python/cpython/pull/5053
    __a						       =				(
        B"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00"
        B"\x00\x02\x08\x06\x00\x00\x00\x99\x81\xb6'\x00\x00\x00\x15I"
        B"DATx\x01\x01\n\x00\xf5\xff\x00PK\x05\x06\x00PK\x06\x06\x07"
        B"\xac\x01N\xc6|a\r\x00\x00\x00\x00IEND\xaeB`\x82"
    )
    with not_a_zip_file.open('''wb'''				) as f:
        f.write(__lowerCAmelCase				)
    assert zipfile.is_zipfile(str(__lowerCAmelCase				)				)  # is a false positive for `zipfile`
    assert not ZipExtractor.is_extractable(__lowerCAmelCase				)  # but we're right









'''simple docstring'''


def        __lowerCAmelCase       (__lowerCAmelCase					,       __lowerCAmelCase      ):

		if density <= 0:
				raise ValueError("Impossible fluid density"      )
		if bulk_modulus <= 0:
				raise ValueError("Impossible bulk modulus"      )

		return (bulk_modulus / density) ** 0.5


if __name__ == "__main__":
						import doctest

						doctest.testmod()



