.Ph$$ddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl ZddlmZe jdkrddlmZnddlZejd"d Zej d#d$d Zd%dZdZeeeZ dZdZejejfdZejeeZejddefd&dZ GddZ!Gddej"ej#Z"Gd d!ej#Z$dS)') annotationsN)Iterator) )tarfiledirstr | os.PathLikereturnIterator[str | os.PathLike]c#Ktj}tj| |Vtj|dS#tj|wxYw)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)rorigs W/var/www/html/test/jupyter/venv/lib/python3.11/site-packages/jaraco/context/__init__.pypushdrsQ 9;;DHSMMM  s AA target_dirstr | os.PathLike | Nonec#K|Gtj|dddd}tj| t j|}tj |d5}| |tdddn #1swxYwY|Vtj |dS#tj |wxYw)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)urlrreqtfs rtarballr))s5DW%%c**229bAAII&RTUU HZ"n$$S)) \#E 2 2 2 Ib MMz2GM H H H I I I I I I I I I I I I I I I j!!!!! j!!!!s0!5C$B?3 C$?CC$CC$$C:membertarfile.TarInfocL|jdd\}|_|S)N/)namesplit)r*r_s rr#r#Ws&[&&sA..NAv{ McLd}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}|Vdddn #1swxYwYddddS#1swxYwYdSN)argskwargssavedresinnerouters rcomposedz/_compose..compose_two..composedts''' 5%%,, #                                  s, A) A- A- AAA) contextlibcontextmanager)r<r=r>s`` r compose_twoz_compose..compose_twoss5      (222r2) functoolsreducereversed)cmgrsrAs r_composerF_s,(333  K% 9 99r2c|\}}}|tjtjtjfvr[|jtjkrFtj|tjtj ztj z||dS)z> Add support for removing read-only files on Windows. N) r rmdirremoveunlinkerrnoEACCESchmodstatS_IRWXUS_IRWXGS_IRWXO)funcrexc_infor1excs rremove_readonlyrUskIAsA ")RY///CI4M4M t|dl2T\ABBB T  r2ctjdkr%tjtjt n tjS)NWindows)onerror)platformsystemrBpartialr$r%rUr7r2rrobust_removerr\s; ?   ) ) &-AAAA ]r2c#pKtj} |V||dS#||wxYw)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_dirs rrarasP!!Hs( 5)r`Tbranch str | Nonequietboolc#Kd|vrdnd}|5}|d||g}|d|gt|z|r tjnd}tj||||VddddS#1swxYwYdS)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)extendre subprocessDEVNULL check_call)r&rbrddest_ctxexerepo_dircmdstreams r repo_contextrus C<<%%TC xGS(+ J'$v,,6777',6##$c&@@@@ sAA;;A?A?ceZdZdZdZeffdZdZedZ edZ edZ dZ d Z ed d Zd Zd S) 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||_dSr6) exceptions)selfrys r__init__zExceptionTrap.__init__s $r2c|Sr6r7rzs r __enter__zExceptionTrap.__enter__ r2c|jdSNrrSr}s rtypezExceptionTrap.type}Qr2c|jdS)Nr.rr}s rvaluezExceptionTrap.valuerr2c|jdS)Nrr}s rtbzExceptionTrap.tbrr2cV|d}|ot||j}|r||_|Sr) issubclassryrS)rzrSrmatchess r__exit__zExceptionTrap.__exit__s5{<:dDO<<  %$DMr2c*t|jSr6)rerr}s r__bool__zExceptionTrap.__bool__sDIr2_testcNtjfd}|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 ctj5}|i|dddn #1swxYwY|Sr6)rwry)r8r9traprrRrzs rwrapperz%ExceptionTrap.raises..wrapperst// &4d%f%%% & & & & & & & & & & & & & & &5;; s +//)rBwraps)rzrRrrs``` rraiseszExceptionTrap.raisessF&             r2cD||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_)rzrRs rpasseszExceptionTrap.passes"s&{{4x}{555r2N)__name__ __module__ __qualname____doc__rS Exceptionr{r~propertyrrrrrrerrr7r2rrwrwsB H#,,%%%%  X   X   X %)666666r2rwceZdZdZdS)suppressz A version of contextlib.suppress with decorator support. >>> @suppress(KeyError) ... def key_error(): ... {}[''] >>> key_error() N)rrrrr7r2rrr8sr2rc&eZdZdZddZdZdZdS) 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.c"||_||_dSr6)actioncode)rzrrs rr{zon_interrupt.__init__Xs  r2c|Sr6r7r}s rr~zon_interrupt.__enter__\rr2c|tus |jdkrdS|jdkrt|j||jdkS)Nignorerr)KeyboardInterruptr SystemExitr)rzexctypeexcinstexctbs rrzon_interrupt.__exit___sK + + +t{h/F/F F [G # #TY''W 4{j((r2N)rr.)rrrrr{r~rr7r2rrrCsP()))))r2r)rr r r r6)rrr r )r*r+r r+)rbrcrdre)% __future__rr?rKrBrr rYr$rNrmsysr^urllib.requestrtypingr version_info backportsrr@rr)r#rF tarball_cwdrUr\r%rar[robust_temp_dirrurwrContextDecoratorrr7r2rrs""""""    g!!!!!!!NNN       04*"*"*"*"*"Z:::<hug&&     ]"$)#Hnn6F6FGGG "$0n6n6n6n6n6n6n6n6bz"J$?!)!)!)!)!):.!)!)!)!)!)r2