




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 		_A(       UpperCamelCase__							:			Any				,       UpperCamelCase__							:			Optional[int]				,       UpperCamelCase__							:			List[str]				,       UpperCamelCase__							:			int				,       UpperCamelCase__							:			Tuple				,       UpperCamelCase__							:			Optional[int]				,       UpperCamelCase__							:			int				,       UpperCamelCase__							:			Union[str, Any]				,       UpperCamelCase__							:			str				,       UpperCamelCase__							:			Optional[int]				,       UpperCamelCase__							:			Tuple				,       UpperCamelCase__							:			Dict				,       )    -> Union[str, Any]:
							'''simple docstring'''


							__lowercase      =    {
							    '''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),
							}
							__lowercase     ,     __lowercase      =    input_paths_and_base_extractors[compression_format]
							if input_path is None:
														__lowercase      =    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__				)
							__lowercase      =    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
																					__lowercase      =    file_path.read_text(encoding='''utf-8'''				)
							else:
														__lowercase      =    output_path.read_text(encoding='''utf-8'''				)
							__lowercase      =    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 		_A(       UpperCamelCase__							:			Tuple				,       UpperCamelCase__							:			List[str]				,       UpperCamelCase__							:			List[str]				,       UpperCamelCase__							:			List[Any]				,       UpperCamelCase__							:			Tuple				,       UpperCamelCase__							:			Any				,       UpperCamelCase__							:			Optional[int]				,       UpperCamelCase__							:			int				,       UpperCamelCase__							:			Optional[int]				,       UpperCamelCase__							:			str				,       UpperCamelCase__							:			Tuple				,       UpperCamelCase__							:			Tuple				,       )    -> Tuple:
							'''simple docstring'''


							__lowercase      =    {
							    '''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,
							}
							__lowercase      =    input_paths[compression_format]
							if input_path is None:
														__lowercase      =    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__				)
							__lowercase      =    Extractor.infer_extractor_format(UpperCamelCase__				)
							assert extractor_format is not None
							__lowercase      =    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
																					__lowercase      =    file_path.read_text(encoding='''utf-8'''				)
							else:
														__lowercase      =    output_path.read_text(encoding='''utf-8'''				)
							__lowercase      =    text_file.read_text(encoding='''utf-8'''				)
							assert extracted_file_content == expected_file_content







@pytest.fixture
def 		_A(       UpperCamelCase__							:			List[str]				,       UpperCamelCase__							:			int				)    -> Tuple:
							'''simple docstring'''


							import tarfile

							__lowercase      =    tmp_path / '''data_dot_dot'''
							directory.mkdir()
							__lowercase      =    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 		_A(       UpperCamelCase__							:			Tuple				)    -> Union[str, Any]:
							'''simple docstring'''


							import tarfile

							__lowercase      =    tmp_path / '''data_sym_link'''
							directory.mkdir()
							__lowercase      =    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 		_A(       UpperCamelCase__							:			str				,       UpperCamelCase__							:			Dict				,       UpperCamelCase__							:			Dict				,       UpperCamelCase__							:			List[Any]				,       UpperCamelCase__							:			Optional[Any]				,       UpperCamelCase__							:			Dict				)    -> Optional[Any]:
							'''simple docstring'''


							__lowercase      =    {
							    '''tar_file_with_dot_dot''': tar_file_with_dot_dot,
							    '''tar_file_with_sym_link''': tar_file_with_sym_link,
							}
							__lowercase      =    insecure_tar_files[insecure_tar_file]
							__lowercase      =    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 		_A(       UpperCamelCase__							:			List[str]				)    -> str:
							'''simple docstring'''


							__lowercase      =    tmpdir / '''not_a_zip_file'''
							# From: https://github.com/python/cpython/pull/5053
							__lowercase      =    (
							    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












# This code is adapted from OpenAI's release
# https://github.com/openai/human-eval/blob/master/human_eval/execution.py

import contextlib
import faulthandler
import io
import multiprocessing
import os
import platform
import signal
import tempfile







def 		_A(       UpperCamelCase__							:			Any				,       UpperCamelCase__							:			int				,       UpperCamelCase__							:			List[Any]				,       UpperCamelCase__							:			int				)    -> List[Any]:
							'''simple docstring'''


							__lowercase      =    multiprocessing.Manager()
							__lowercase      =    manager.list()

							__lowercase      =    multiprocessing.Process(target=UpperCamelCase__				,       args=(check_program, result, timeout)				)
							p.start()
							p.join(timeout=timeout + 1				)
							if p.is_alive():
														p.kill()

							if not result:
														result.append('''timed out'''				)

							return {
							    "task_id": task_id,
							    "passed": result[0] == "passed",
							    "result": result[0],
							    "completion_id": completion_id,
							}







def 		_A(       UpperCamelCase__							:			int				,       UpperCamelCase__							:			List[Any]				,       UpperCamelCase__							:			Optional[int]				)    -> List[Any]:
							'''simple docstring'''


							with create_tempdir():
														# These system calls are needed when cleaning up tempdir.
														import os
														import shutil

														__lowercase      =    shutil.rmtree
														__lowercase      =    os.rmdir
														__lowercase      =    os.chdir

														# Disable functionalities that can make destructive changes to the test.
														reliability_guard()

														# Run program.
														try:
																					__lowercase      =    {}
																					with swallow_io():
																												with time_limit(UpperCamelCase__				):
																																			exec(UpperCamelCase__				,       UpperCamelCase__				)
																					result.append('''passed'''				)
														except TimeoutException:
																					result.append('''timed out'''				)
														except BaseException as e:
																					result.append(F'failed: {e}'				)

														# Needed for cleaning up.
														__lowercase      =    rmtree
														__lowercase      =    rmdir
														__lowercase      =    chdir







@contextlib.contextmanager
def 		_A(       UpperCamelCase__							:			Union[str, Any]				)    -> Optional[int]:
							'''simple docstring'''


							def signal_handler(UpperCamelCase__							:			List[str]				,       UpperCamelCase__							:			Optional[Any]				):
														raise TimeoutException('''Timed out!'''				)

							signal.setitimer(signal.ITIMER_REAL				,       UpperCamelCase__				)
							signal.signal(signal.SIGALRM				,       UpperCamelCase__				)
							try:
														yield
							finally:
														signal.setitimer(signal.ITIMER_REAL				,       0				)







@contextlib.contextmanager
def 		_A(       )    -> Any:
							'''simple docstring'''


							__lowercase      =    WriteOnlyStringIO()
							with contextlib.redirect_stdout(UpperCamelCase__				):
														with contextlib.redirect_stderr(UpperCamelCase__				):
																					with redirect_stdin(UpperCamelCase__				):
																												yield







@contextlib.contextmanager
def 		_A(       )    -> Tuple:
							'''simple docstring'''


							with tempfile.TemporaryDirectory() as dirname:
														with chdir(UpperCamelCase__				):
																					yield dirname







class a (     __SCREAMING_SNAKE_CASE  ):
						"""simple docstring"""

						pass







class a (     io.StringIO  ):
						"""simple docstring"""

						def     UpperCAmelCase_      ( self :     List[str]				,							*lowerCamelCase__ :     str				,							**lowerCamelCase__ :     str				)    ->						Tuple:
													"""simple docstring"""


													raise OSError






						def     UpperCAmelCase_      ( self :     List[str]				,							*lowerCamelCase__ :     List[str]				,							**lowerCamelCase__ :     Dict				)    ->						List[Any]:
													"""simple docstring"""


													raise OSError






						def     UpperCAmelCase_      ( self :     List[str]				,							*lowerCamelCase__ :     Dict				,							**lowerCamelCase__ :     Dict				)    ->						Optional[int]:
													"""simple docstring"""


													raise OSError






						def     UpperCAmelCase_      ( self :     Tuple				,							*lowerCamelCase__ :     Optional[Any]				,							**lowerCamelCase__ :     Tuple				)    ->						Optional[Any]:
													"""simple docstring"""


													return False







class a (     contextlib._RedirectStream  ):  # type: ignore
						"""simple docstring"""

						UpperCamelCase_	:       str       	=		'stdin'







@contextlib.contextmanager
def 		_A(       UpperCamelCase__							:			int				)    -> Any:
							'''simple docstring'''


							if root == ".":
														yield
														return
							__lowercase      =    os.getcwd()
							os.chdir(UpperCamelCase__				)
							try:
														yield
							except BaseException as exc:
														raise exc
							finally:
														os.chdir(UpperCamelCase__				)







def 		_A(       UpperCamelCase__							:			Optional[Any]=None				)    -> Any:
							'''simple docstring'''


							if maximum_memory_bytes is not None:
														import resource

														resource.setrlimit(resource.RLIMIT_AS				,       (maximum_memory_bytes, maximum_memory_bytes)				)
														resource.setrlimit(resource.RLIMIT_DATA				,       (maximum_memory_bytes, maximum_memory_bytes)				)
														if not platform.uname().system == "Darwin":
																					resource.setrlimit(resource.RLIMIT_STACK				,       (maximum_memory_bytes, maximum_memory_bytes)				)

							faulthandler.disable()

							import builtins

							__lowercase      =    None
							__lowercase      =    None

							import os

							__lowercase      =    '''1'''

							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None

							import shutil

							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None

							import subprocess

							__lowercase      =    None  # type: ignore

							__lowercase      =    None

							import sys

							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None
							__lowercase      =    None






