ngP%4ddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl mZejdkrddlmZnddlZej$ddZej$ d ddZ ddZd ZeeeZej$d Zd Zej$ej4fd Zej$dd efdZdZGddZGddej>ej@ZGddej@Z!y)) annotationsN)Iterator) )tarfilec#Ktj}tj| |tj|y#tj|wxYww)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)dirorigs _/var/www/skyplay_api_hub/venv/lib/python3.12/site-packages/setuptools/_vendor/jaraco/context.pypushdrs> 99;DHHSM  s*A!AA!AA!c#K|?tjj|jddjdd}tj| t j j|}tj|d5}|j|tddd|tj|y#1swY#xYw#tj|wxYww)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 rbasenamereplacemkdirurllibrequesturlopenropen extractallstrip_first_componentshutilrmtree)url target_dirreqtfs rtarballr%'s WW%%c*229bAII&RTU HHZ"nn$$S) \\#E 2 Ib MMz2GM H I j!  I I  j!s6AC06CC ( C4C0 CCC--C0cN|jjdd\}|_|S)N/)namesplit)memberr_s rrrGs%[[&&sA.NAv{ McFd}tj|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) c<fd}tj|S)Nc?K|i|5}|5}|ddddddy#1swYxYw#1swYyxYwwN)argskwargssavedresinnerouters rcomposedz/_compose..compose_two..composeddsL'' 5%, #       s( A 7+7 A4 7AA) contextlibcontextmanager)r7r8r9s`` r compose_twoz_compose..compose_twocs ((22r-) functoolsreducereversed)cmgrsr<s r_composerAOs (3   K% 99r-c/Ktjdtd|jdt}t |i|5}||5}|ddddddy#1swYxYw#1swYyxYww)NzBtarball_context is deprecated. Use tarball or tarball_cwd instead. stacklevelr)warningswarnDeprecationWarningpoprr%)r3r4 pushd_ctxtballr s rtarball_contextrLpss MML  7E*I $ !& !UIe,< s5>A7 A+ AA+ A7A( $A++A40A7ctjdtd|dd}tddd }|j |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 replacementrCrDNzjJ)gzbzxz)rFrGrHdictget)r!compression_indicatormappings rinfer_compressionrY|sG MM=  Hccc*G ;;,c 22r-c#hKtj} |||y#||wxYww)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)removertemp_dirs rr^r^s0!Hs2% 2 /2Tc#Kd|vrdnd}|5}|d||g}|r|jd|gttjjd}|r|nd}t j |||dddy#1swYyxYww)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. githgclonez--branchwN)stdout)extendrr rdevnull subprocess check_call) r!branchquietdest_ctxexerepo_dircmdrfrds r repo_contextrosC<%TC xGS(+  JJ F+ ,rww,!tc&1sBA!A<3 B<BBcbtjdtdtjS)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.nullcontextrCrD)rFrGrHr: nullcontextr2r-rnullrrs* MM8  ! ! ##r-cteZdZdZdZeffdZdZedZ edZ edZ dZ d Z ed d Zd Zy ) 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 )NNNc||_yr1) exceptions)selfrvs r__init__zExceptionTrap.__init__s $r-c|Sr1r2rws r __enter__zExceptionTrap.__enter__ r-c |jdSNrexc_inforzs rtypezExceptionTrap.type}}Qr-c |jdS)Nr(rrzs rvaluezExceptionTrap.valuerr-c |jdS)NrCrrzs rtbzExceptionTrap.tbrr-cV|d}|xrt||j}|r||_|Sr~) issubclassrvr)rwrrmatchess r__exit__zExceptionTrap.__exit__s/{<:dDOO< $DMr-c,t|jSr1)boolrrzs r__bool__zExceptionTrap.__bool__ sDIIr-_testcJtjfd}|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 cxtj5}|i|dddS#1swYxYwr1)rtrv)r3r4traprfuncrws rwrapperz%ExceptionTrap.raises..wrapper!s=t/ &4d%f% &;  & &s 09)r=wraps)rwrrrs``` rraiseszExceptionTrap.raisess'&      r-cD|j|tjS)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_)rwrs rpasseszExceptionTrap.passes)s&{{4x}}{55r-N)__name__ __module__ __qualname____doc__r Exceptionrxr{propertyrrrrrrrrr2r-rrtrtssB H#,,%      %)66r-rtceZdZdZy)suppressz A version of contextlib.suppress with decorator support. >>> @suppress(KeyError) ... def key_error(): ... {}[''] >>> key_error() N)rrrrr2r-rrr?sr-rc$eZdZdZddZdZdZy) 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)() c ||_||_yr1)actioncode)rwrrs rrxzon_interrupt.__init__]s  r-c|Sr1r2rzs rr{zon_interrupt.__enter__ar|r-c|tus|jdk(ry|jdk(rt|j||jdk(S)Nignoreerrorr)KeyboardInterruptr SystemExitr)rwexctypeexcinstexctbs rrzon_interrupt.__exit__dsE + +t{{h/F  [[G #TYY'W 4{{j((r-N)rr()rrrrrxr{rr2r-rrrJs$)r-r)r zstr | os.PathLikereturnIterator[str | os.PathLike]r1)r"zstr | os.PathLike | Nonerr)r+tarfile.TarInforr)" __future__rr:r=rr rrgsysr[urllib.requestrrFtypingr version_info backportsrr;rr%rrA tarball_cwdrLrYr r^rorrrtrContextDecoratorrr2r-rrsO"  g!     04"-" ""> :<ug&  32 ]]$ !$$&n6n6bz""J$?$?):..)r-