U X7hP%@sZddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl mZejdkrddlmZnddlZejdddd d Zejd&d dd d dZdddddZddZeeeZejddZddZejejfddZejddefddZddZGd d!d!ZGd"d#d#ejej ZGd$d%d%ej Z!dS)') annotationsN)Iterator) )tarfilezstr | os.PathLikezIterator[str | os.PathLike])dirreturnc cs.t}t|z |VW5t|XdS)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 [/var/www/formularioweb/env/lib/python3.8/site-packages/setuptools/_vendor/jaraco/context.pypushds   rzstr | os.PathLike | None) target_dirrc cs~|dkr$tj|dddd}t|z>tj |}t j |dd}|j |t dW5QRX|VW5t|XdS)a Get a tarball, extract it, yield, then clean up. >>> 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) Nz.tar.gzz.tgzzr|*)fileobjmode)pathfilter)r rbasenamereplacemkdirshutilrmtreeurllibZrequestZurlopenropen extractallstrip_first_component)urlrreqZtfr r rtarball's   r!ztarfile.TarInfo)memberrcCs|jdd\}|_|S)N/)namesplit)r"r_r r rrGsrcGsdd}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 ?s4|| }| }|VW5QRXW5QRXdSNr )argskwargsZsavedresinnerouterr rcomposeddsz/_compose..compose_two..composed) contextlibcontextmanager)r-r.r/r r,r compose_twocsz_compose..compose_two) functoolsreducereversed)Zcmgrsr2r r r_composeOsr6c osPtjdtdd|dt}t|| }|| }|VW5QRXW5QRXdS)NzBtarball_context is deprecated. Use tarball or tarball_cwd instead. stacklevelr)warningswarnDeprecationWarningpoprr!)r)r*Z pushd_ctxZtballrr r rtarball_contextps r>cCs6tjdtdd|dd}tdddd }||dS) a Given a URL or filename, infer the compression code for tar. >>> infer_compression('http://foo/bar.tar.gz') 'z' >>> infer_compression('http://foo/bar.tgz') 'z' >>> infer_compression('file.bz') 'j' >>> infer_compression('file.xz') 'J' z3infer_compression is deprecated with no replacementr7r8NzjJ)gzZbzxz)r:r;r<dictget)rZcompression_indicatormappingr r rinfer_compression|s  rHccs"t}z |VW5||XdS)a` 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) ... _ = pathlib.Path(the_dir).joinpath('somefile').write_text('contents', encoding='utf-8') >>> assert not os.path.exists(the_dir) N)tempfilemkdtemp)Zremovertemp_dirr r rrKs  rKTc csrd|kr dnd}|R}|d||g}|r6|d|gttjjd}|rL|nd}tj||d|VW5QRXdS)z 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. gitZhgZclonez--branchwN)stdout)extendrr rdevnull subprocess check_call) rbranchquietZdest_ctxZexeZrepo_dircmdrPrNr r r repo_contexts  rVcCstjdtddtS)aD A null context suitable to stand in for a meaningful context. >>> with null() as value: ... assert value is None This context is most useful when dealing with two or more code branches but only some need a context. Wrap the others in a null context to provide symmetry across all options. z.null is deprecated. Use contextlib.nullcontextr7r8)r:r;r<r0 nullcontextr r r rnulls rXc@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)selfrZr r r__init__szExceptionTrap.__init__cCs|Sr(r r[r r r __enter__szExceptionTrap.__enter__cCs |jdSNrexc_infor]r r rtypeszExceptionTrap.typecCs |jdS)Nr$r`r]r r rvalueszExceptionTrap.valuecCs |jdS)Nr7r`r]r r rtbszExceptionTrap.tbcGs&|d}|ot||j}|r"||_|Sr_) issubclassrZra)r[rarbmatchesr r r__exit__s zExceptionTrap.__exit__cCs t|jSr()boolrbr]r r r__bool__ szExceptionTrap.__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 c s(tj}||W5QRX|Sr()rYrZ)r)r*traprkfuncr[r rwrapper!s z%ExceptionTrap.raises..wrapper)r3wraps)r[rnrkror rmrraisesszExceptionTrap.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 rj)rqoperatornot_)r[rnr r rpasses)szExceptionTrap.passesN)__name__ __module__ __qualname____doc__ra Exceptionr\r^propertyrbrcrdrgrirhrqrtr r r rrYs!   rYc@seZdZdZdS)suppressz A version of contextlib.suppress with decorator support. >>> @suppress(KeyError) ... def key_error(): ... {}[''] >>> key_error() N)rurvrwrxr r r rr{?sr{c@s*eZdZdZd ddZddZdd Zd S) on_interrupta Replace a KeyboardInterrupt with SystemExit(1) >>> 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)r[r~rr r rr\]szon_interrupt.__init__cCs|Sr(r r]r r rr^aszon_interrupt.__enter__cCs6|tk s|jdkrdS|jdkr,t|j||jdkS)Nignorer}r{)KeyboardInterruptr~ SystemExitr)r[exctypeexcinstexctbr r rrgds   zon_interrupt.__exit__N)r}r$)rurvrwrxr\r^rgr r r rr|Js r|)N)" __future__rr0r3rrr rrQsysrIZurllib.requestrr:typingr version_info backportsrr1rr!rr6Z tarball_cwdr>rHrrKrVrXrYr{ContextDecoratorr|r r r rsB     q