o V h$@s`ddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl ZddlmZe jdkrHddlmZnddlZejd*d d Zej d+d,ddZd-ddZddZeeeZ ddZddZejejfddZejeedZejddefd.d"d#Z Gd$d%d%Z!Gd&d'd'ej"ej#Z"Gd(d)d)ej#Z$dS)/) annotationsN)Iterator) )tarfiledirstr | os.PathLikereturnIterator[str | os.PathLike]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)rorigrk/var/www/html/construction_image-detection-poc/venv/lib/python3.10/site-packages/jaraco/context/__init__.pypushds  r target_dirstr | os.PathLike | 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 rbasenamereplacemkdirurllibrequesturlopenropen extractallstrip_first_componentshutilrmtree)urlrreqtfrrrtarball)s"  r'membertarfile.TarInfocCs|jdd\}|_|S)N/)namesplit)r(r_rrrr!Wsr!cGsdd}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) csfdd}t|S)Nc ?st|i|(}| }|VWdn1swYWddSWddS1s3wYdSNr)argskwargssavedresinnerouterrrcomposedtsPz/_compose..compose_two..composed) contextlibcontextmanager)r5r6r7rr4r compose_twoss z_compose..compose_two) functoolsreducereversed)cmgrsr:rrr_compose_sr?cCsR|\}}}|tjtjtjfvr(|jtjkr(t|tjtj Btj B||dS)z> Add support for removing read-only files on Windows. N) r rmdirremoveunlinkerrnoEACCESchmodstatS_IRWXUS_IRWXGS_IRWXO)funcrexc_infor.excrrrremove_readonlys   rMcCs"tdkrtjtjtdStjS)NWindows)onerror)platformsystemr;partialr"r#rMrrrrrobust_removers rSccs*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)removertemp_dirrrrrWs  rW)rVTbranch str | Nonequietboolccsd|vrdnd}|,}|d||g}|d|gt||r#tjnd}tj|||d|VWddS1s;wYdS)aT 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. >>> 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$rXrZdest_ctxexerepo_dircmdstreamrrr repo_contexts "rjc@steZdZdZdZeffddZddZeddZ ed d Z ed d Z d dZ ddZ edddZddZdS) 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 )NNNcCs ||_dSr/) exceptions)selfrlrrr__init__ zExceptionTrap.__init__cC|Sr/rrmrrr __enter__zExceptionTrap.__enter__cC |jdSNrrKrqrrrtype zExceptionTrap.typecCrt)Nr+rvrqrrrvaluerxzExceptionTrap.valuecCrt)NrvrqrrrtbrxzExceptionTrap.tbcGs&|d}|o t||j}|r||_|Sru) issubclassrlrK)rmrKrwmatchesrrr__exit__s zExceptionTrap.__exit__cCs t|jSr/)r[rwrqrrr__bool__rozExceptionTrap.__bool___testcstfdd}|S)a Wrap func and replace the result with the truth value of the trap (True if an exception occurred). First, give the decorator an alias to support Python 3.8 Syntax. >>> raises = ExceptionTrap(ValueError).raises Now decorate a function that always fails. >>> @raises ... def fail(): ... raise ValueError('failed') >>> fail() True csFtj}|i|Wd|S1swY|Sr/)rkrl)r0r1traprrJrmrrwrappers  z%ExceptionTrap.raises..wrapper)r;wraps)rmrJrrrrrraisesszExceptionTrap.raisescCs|j|tjdS)a Wrap func and replace the result with the truth value of the trap (True if no exception). First, give the decorator an alias to support Python 3.8 Syntax. >>> passes = ExceptionTrap(ValueError).passes Now decorate a function that always fails. >>> @passes ... def fail(): ... raise ValueError('failed') >>> fail() False r)roperatornot_)rmrJrrrpasses"szExceptionTrap.passesN)__name__ __module__ __qualname____doc__rK Exceptionrnrrpropertyrwryr{r~rr[rrrrrrrks!    rkc@seZdZdZdS)suppressz A version of contextlib.suppress with decorator support. >>> @suppress(KeyError) ... def key_error(): ... {}[''] >>> key_error() N)rrrrrrrrr8src@s*eZdZdZd ddZddZdd Zd S) 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)() errorr+cCs||_||_dSr/)actioncode)rmrrrrrrnXs zon_interrupt.__init__cCrpr/rrqrrrrr\rszon_interrupt.__enter__cCs6|tus |jdkr dS|jdkrt|j||jdkS)Nignorerr)KeyboardInterruptr SystemExitr)rmexctypeexcinstexctbrrrr~_s    zon_interrupt.__exit__N)rr+)rrrrrnrrr~rrrrrCs   r)rrr r r/)rrr r )r(r)r r))rXrYrZr[)% __future__rr8rCr;rr rPr"rFrbsysrTurllib.requestrtypingr version_info backportsrr9rr'r!r? tarball_cwdrMrSr#rWrRrobust_temp_dirrjrkrContextDecoratorrrrrrsJ     - q