
"""simple docstring"""







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 					__lowerCAmelCase       (_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						):
	__lowerCAmelCase    :		Union[str, Any]		    =   {
	    '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),
	}
	__lowerCAmelCase    ,     __lowerCAmelCase    :		Tuple		    =   input_paths_and_base_extractors[compression_format]
	if input_path is None:
		__lowerCAmelCase    :		List[str]		    =   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(_UpperCamelCase	)
	assert base_extractor.is_extractable(_UpperCamelCase	)
	__lowerCAmelCase    :		Optional[int]		    =   tmp_path / ('extracted' if is_archive else 'extracted.txt')
	base_extractor.extract(_UpperCamelCase ,						_UpperCamelCase	)
	if is_archive:
		assert output_path.is_dir()
		for file_path in output_path.iterdir():
			assert file_path.name == text_file.name
			__lowerCAmelCase    :		Dict		    =   file_path.read_text(encoding='utf-8'	)
	else:
		__lowerCAmelCase    :		Tuple		    =   output_path.read_text(encoding='utf-8'	)
	__lowerCAmelCase    :		Any		    =   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 					__lowerCAmelCase       (_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						):
	__lowerCAmelCase    :		str		    =   {
	    '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,
	}
	__lowerCAmelCase    :		Any		    =   input_paths[compression_format]
	if input_path is None:
		__lowerCAmelCase    :		List[Any]		    =   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(_UpperCamelCase	)
	__lowerCAmelCase    :		Dict		    =   Extractor.infer_extractor_format(_UpperCamelCase	)
	assert extractor_format is not None
	__lowerCAmelCase    :		Any		    =   tmp_path / ('extracted' if is_archive else 'extracted.txt')
	Extractor.extract(_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase	)
	if is_archive:
		assert output_path.is_dir()
		for file_path in output_path.iterdir():
			assert file_path.name == text_file.name
			__lowerCAmelCase    :		Dict		    =   file_path.read_text(encoding='utf-8'	)
	else:
		__lowerCAmelCase    :		Optional[Any]		    =   output_path.read_text(encoding='utf-8'	)
	__lowerCAmelCase    :		Optional[int]		    =   text_file.read_text(encoding='utf-8'	)
	assert extracted_file_content == expected_file_content
@pytest.fixture
def 					__lowerCAmelCase       (_UpperCamelCase ,						_UpperCamelCase	):
	import tarfile

	__lowerCAmelCase    :		Tuple		    =   tmp_path / 'data_dot_dot'
	directory.mkdir()
	__lowerCAmelCase    :		str		    =   directory / 'tar_file_with_dot_dot.tar'
	with tarfile.TarFile(_UpperCamelCase ,						'w'	) as f:
		f.add(_UpperCamelCase ,						arcname=os.path.join('..' ,						text_file.name	)	)
	return path
@pytest.fixture
def 					__lowerCAmelCase       (_UpperCamelCase	):
	import tarfile

	__lowerCAmelCase    :		List[str]		    =   tmp_path / 'data_sym_link'
	directory.mkdir()
	__lowerCAmelCase    :		List[str]		    =   directory / 'tar_file_with_sym_link.tar'
	os.symlink('..' ,						directory / 'subdir' ,						target_is_directory=_UpperCamelCase	)
	with tarfile.TarFile(_UpperCamelCase ,						'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 					__lowerCAmelCase       (_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase	):
	__lowerCAmelCase    :		Tuple		    =   {
	    'tar_file_with_dot_dot': tar_file_with_dot_dot,
	    'tar_file_with_sym_link': tar_file_with_sym_link,
	}
	__lowerCAmelCase    :		List[str]		    =   insecure_tar_files[insecure_tar_file]
	__lowerCAmelCase    :		Optional[Any]		    =   tmp_path / 'extracted'
	TarExtractor.extract(_UpperCamelCase ,						_UpperCamelCase	)
	assert caplog.text
	for record in caplog.records:
		assert record.levelname == "ERROR"
		assert error_log in record.msg
def 					__lowerCAmelCase       (_UpperCamelCase	):
	# We should have less false positives than zipfile.is_zipfile
	# We do that by checking only the magic number
	__lowerCAmelCase    :		int		    =   tmpdir / 'not_a_zip_file'
	# From: https://github.com/python/cpython/pull/5053
	__lowerCAmelCase    :		str		    =   (
	    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(_UpperCamelCase	)
	assert zipfile.is_zipfile(str(_UpperCamelCase	)	)  # is a false positive for `zipfile`
	assert not ZipExtractor.is_extractable(_UpperCamelCase	)  # but we're right

"""simple docstring"""







def 					__lowerCAmelCase       (_UpperCamelCase ,						_UpperCamelCase = False	):

	if not isinstance(_UpperCamelCase ,						_UpperCamelCase	):
		__lowerCAmelCase    :		str		    =   F"Expected string as input, found {type(_UpperCamelCase	)}"
		raise ValueError(_UpperCamelCase	)
	if not isinstance(_UpperCamelCase ,						_UpperCamelCase	):
		__lowerCAmelCase    :		Union[str, Any]		    =   F"Expected boolean as use_pascal parameter, found {type(_UpperCamelCase	)}"
		raise ValueError(_UpperCamelCase	)

	__lowerCAmelCase    :		Tuple		    =   input_str.split('_'	)

	__lowerCAmelCase    :		int		    =   0 if use_pascal else 1

	__lowerCAmelCase    :		Any		    =   words[start_index:]

	__lowerCAmelCase    :		Any		    =   [word[0].upper() + word[1:] for word in words_to_capitalize]

	__lowerCAmelCase    :		Any		    =   '' if use_pascal else words[0]

	return "".join([initial_word, *capitalized_words]	)


if __name__ == "__main__":
							from doctest import testmod

							testmod()