o l>j0@sFUddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl Z ddlZddlmZmZmZddlmZddlmZmZmZmZes^e jdkrcddlZnddlmZerddlmZdd lm Z m!Z!m"Z"dd l#m$Z$m%Z%m&Z&ed e d Z'e$d Z(e)e*e+dBe+dBedBfZ,de-d<edZ.edddZ/edddZ0ej1dLddZ2ej1dMdNdd Z3d!d"Z4dOd'd(Z5e4ej6e5Z7dPd,d-Z8e8e2e3Z9 dQd3d4Z:dRd6d7Z;ej1e je=e;d=Z?ej1dde?fdTdDdEZ@GdFdGdGZAGdHdIdIejBejCZBGdJdKdKejCZDdS)U) annotationsN)Callable GeneratorIterator) TracebackType) TYPE_CHECKINGLiteralTypeVarcast) )tarfile) TypeAlias)FileDescriptorOrPath OptExcInfoStrPath) ParamSpecSelfUnpack_FileDescriptorOrPathT)bound_Pr_UnpackableOptExcInfo_R_T1_coT) covariant_T2_codirrreturnIterator[StrPath]c cs8t}t|z |VWt|dSt|w)z >>> tmp_path = getfixture('tmp_path') >>> with pushd(tmp_path): ... assert os.getcwd() == os.fspath(tmp_path) >>> assert os.getcwd() != os.fspath(tmp_path) N)osgetcwdchdir)rorigr$L/home/thesage/.local/lib/python3.10/site-packages/jaraco/context/__init__.pypushd4s  r&urlstr target_dirStrPath | Nonec cs|durtj|dddd}t|z/tj|}tj |dd}|j |t dWdn1s8wY|VWt |dSt |w)au Get a URL to a tarball, download, extract, yield, then clean up. Assumes everything in the tarball is prefixed with a common directory. That common path is stripped and the contents are extracted to ``target_dir``, similar to passing ``-C {target} --strip-components 1`` to the ``tar`` command. Uses the streaming protocol to extract the contents from a stream in a single pass without loading the whole file into memory. >>> import urllib.request >>> url = getfixture('tarfile_served') >>> target = getfixture('tmp_path') / 'out' >>> tb = tarball(url, target_dir=target) >>> import pathlib >>> with tb as extracted: ... contents = pathlib.Path(extracted, 'contents.txt').read_text(encoding='utf-8') >>> assert not os.path.exists(extracted) If the target is not specified, contents are extracted to a directory relative to the current working directory named after the name of the file as extracted from the URL. >>> target = getfixture('tmp_path') >>> with pushd(target), tarball(url): ... target.joinpath('served').is_dir() True Nz.tar.gzz.tgzzr|*)fileobjmode)pathfilter)r r.basenamereplacemkdirurllibrequesturlopenr open extractall_default_filtershutilrmtree)r'r)reqtfr$r$r%tarballEs   r=cGsdd}t||ddS)NcsfddS)Ncs|||SNr$memberr.f1f2r$r%ssz?_compose_tarfile_filters..compose_two..r$rAr$rAr% compose_tworsz-_compose_tarfile_filters..compose_twocS|Sr>r$r?r$r$r%rDusz*_compose_tarfile_filters..) functoolsreduce)filtersrEr$r$r%_compose_tarfile_filtersqsrJr@tarfile.TarInfor.objectcCs|jdd\}|_|S)N/)namesplit)r@r._r$r$r%strip_first_componentxsrRcmgrsUnpack[tuple[Callable[[_T1_co], contextlib.AbstractContextManager[_T2_co]], Callable[_P, contextlib.AbstractContextManager[_T1_co]]]]9Callable[_P, contextlib._GeneratorContextManager[_T2_co]]cGsd dd}t|t|S) a Compose any number of dependent context managers into a single one. The last, innermost context manager may take arbitrary arguments, but each successive context manager should accept the result from the previous as a single parameter. Like :func:`jaraco.functools.compose`, behavior works from right to left, so the context manager should be indicated from outermost to innermost. Example, to create a context manager to change to a temporary directory: >>> temp_dir_as_cwd = _compose(pushd, temp_dir) >>> with temp_dir_as_cwd() as dir: ... assert os.path.samefile(os.getcwd(), dir) inner7Callable[_P, contextlib.AbstractContextManager[_T1_co]]outer=Callable[[_T1_co], contextlib.AbstractContextManager[_T2_co]]rrUcsd fdd }t|S) Nargs_P.argskwargs _P.kwargsrGenerator[_T2_co]c ?st|i|(}| }|VWdn1swYWddSWddS1s3wYdSr>r$)rZr\savedresrVrXr$r%composedsPz/_compose..compose_two..composed)rZr[r\r]rr^) contextlibcontextmanager)rVrXrbr$rar%rEs z_compose..compose_twoN)rVrWrXrYrrU)rGrHreversed)rSrEr$r$r%_composes rffunc*Callable[[_FileDescriptorOrPathT], object]exc_infotuple[object, OSError, object]NonecCsR|\}}}|tjtjtjfvr(|jtjkr(t|tjtj Btj B||dS)z> Add support for removing read-only files on Windows. N) r rmdirremoveunlinkerrnoEACCESchmodstatS_IRWXUS_IRWXGS_IRWXO)rgr.rirQexcr$r$r%remove_readonlys   rwCallable[..., None]cCs(tdkrtjtdtjtdStjS)NWindowsrx)onerror)platformsystemrGpartialr r9r:rwr$r$r$r%robust_removers  r~removerCallable[[str], object]Generator[str]ccs*t}z |VW||dS||w)z Create a temporary directory context. Pass a custom remover to override the removal behavior. >>> import pathlib >>> with temp_dir() as the_dir: ... assert os.path.isdir(the_dir) >>> assert not os.path.exists(the_dir) N)tempfilemkdtemp)rtemp_dirr$r$r%rs  r)rbranch str | Nonequietbooldest_ctx4Callable[[], contextlib.AbstractContextManager[str]]ccsd|vrdnd}|,}|d||g}|d|gt||r#tjnd}tj|||d|VWddS1s;wYdS)a Check out the repo indicated by url. If dest_ctx is supplied, it should be a context manager to yield the target directory for the check out. >>> getfixture('ensure_git') >>> getfixture('needs_internet') >>> repo = repo_context('https://github.com/jaraco/jaraco.context') >>> with repo as dest: ... listing = os.listdir(dest) >>> 'README.rst' in listing True githgclonez--branchN)stdoutstderr)extendr subprocessDEVNULL check_call)r'rrrexerepo_dircmdstreamr$r$r% repo_contexts "rc@seZdZUdZdZded<effd(ddZd)d d Ze d*ddZ e d+ddZ e d,ddZ d-ddZ d.ddZedd/d#d$Zd0d%d&Zd'S)1 ExceptionTrapa A context manager that will catch certain exceptions and provide an indication they occurred. >>> with ExceptionTrap() as trap: ... raise Exception() >>> bool(trap) True >>> with ExceptionTrap() as trap: ... pass >>> bool(trap) False >>> with ExceptionTrap(ValueError) as trap: ... raise ValueError("1 + 1 is not 3") >>> bool(trap) True >>> trap.value ValueError('1 + 1 is not 3') >>> trap.tb >>> with ExceptionTrap(ValueError) as trap: ... raise Exception() Traceback (most recent call last): ... Exception >>> bool(trap) False )NNNrri exceptionstuple[type[BaseException], ...]cCs ||_dSr>)r)selfrr$r$r%__init__+ zExceptionTrap.__init__rrcCrFr>r$rr$r$r% __enter__.zExceptionTrap.__enter__type[BaseException] | NonecC |jdSNrrirr$r$r%type1 zExceptionTrap.typeBaseException | NonecCr)NrNrrr$r$r%value5rzExceptionTrap.valueTracebackType | NonecCr)Nrrr$r$r%tb9rzExceptionTrap.tbUnpack[_UnpackableOptExcInfo]*builtins.type[BaseException] | None | boolcGs&|d}|o t||j}|r||_|Sr) issubclassrri)rriexc_typematchesr$r$r%__exit__=s zExceptionTrap.__exit__rcCs t|jSr>)rrrr$r$r%__bool__GrzExceptionTrap.__bool___testrgCallable[_P, _R]rCallable[[ExceptionTrap], bool]$functools._Wrapped[_P, _R, _P, bool]cs td fdd }|S) a> Wrap func and replace the result with the truth value of the trap (True if an exception occurred). Decorate a function that always fails. >>> @ExceptionTrap(ValueError).raises ... def fail(): ... raise ValueError('failed') >>> fail() True rZr[r\r]rrcsFtj}|i|Wd|S1swY|Sr>)rr)rZr\traprrgrr$r%wrapperZs  z%ExceptionTrap.raises..wrapperN)rZr[r\r]rr)rGwraps)rrgrrr$rr%raisesJszExceptionTrap.raisescCs|j|tjdS)a7 Wrap func and replace the result with the truth value of the trap (True if no exception). Decorate a function that always fails. >>> @ExceptionTrap(ValueError).passes ... def fail(): ... raise ValueError('failed') >>> fail() False r)roperatornot_)rrgr$r$r%passesbszExceptionTrap.passesN)rrrr)rr)rr)rr)rirrr)rr)rgrrrrr)rgrrr)__name__ __module__ __qualname____doc__ri__annotations__ Exceptionrrpropertyrrrrrrrrr$r$r$r%rs  !      rc@seZdZdZdS)suppressz A version of contextlib.suppress with decorator support. >>> @suppress(KeyError) ... def key_error(): ... {}[''] >>> key_error() N)rrrrr$r$r$r%rssrc@s2eZdZdZ dddd Zdd d ZdddZdS) on_interruptaF Replace a KeyboardInterrupt with SystemExit(1). Useful in conjunction with console entry point functions. >>> def do_interrupt(): ... raise KeyboardInterrupt() >>> on_interrupt('error')(do_interrupt)() Traceback (most recent call last): ... SystemExit: 1 >>> on_interrupt('error', code=255)(do_interrupt)() Traceback (most recent call last): ... SystemExit: 255 >>> on_interrupt('suppress')(do_interrupt)() >>> with __import__('pytest').raises(KeyboardInterrupt): ... on_interrupt('ignore')(do_interrupt)() errorrNcodeintaction&Literal['ignore', 'suppress', 'error']cCs||_||_dSr>)rr)rrrr$r$r%rs zon_interrupt.__init__rrcCrFr>r$rr$r$r%rrzon_interrupt.__enter__exctyperexcinstrexctbr None | boolcCs6|tus |jdkr dS|jdkrt|j||jdkS)Nignorerr)KeyboardInterruptr SystemExitr)rrrrr$r$r%rs    zon_interrupt.__exit__N)rrN)rrrrr)rrrrrrrr)rrrrrrrr$r$r$r%r~s  r)rrrrr>)r'r(r)r*rr)r@rKr.rLrrK)rSrTrrU)rgrhr.rrirjrrk)rrx)rrrr) r'r(rrrrrrrr)E __future__rbuiltinsrcrorGrr r{r9rrrsysrurllib.requestr3collections.abcrrrtypesrtypingrrr r version_infor backportsr _typeshedrrrtyping_extensionsrrrrrtupler BaseExceptionrrrrrrdr&r=rJrR data_filterr8rf tarball_cwdrwr~r:rr}robust_temp_dirrrrContextDecoratorrr$r$r$r%sx         +    -   l