Ki0UddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl Z ddlZddlmZmZmZddlmZddlmZmZmZmZese j6dk\rddlZnddlmZer,ddlmZdd lm Z m!Z!m"Z"dd l#m$Z$m%Z%m&Z&ed e Z'e$d Z(e)e*e+dze+dzedzfZ,de-d<edZ.eddZ/eddZ0ejbd%dZ2ejbd&d'dZ3dZ4 d(dZ5e4ejle5Z7 d)dZ8e8e2e3Z9 d*dZ:d+dZ;ejbe jxfd,dZ=ej|e=e;Z?ejbdde?f d-dZ@Gdd ZAGd!d"ejejZBGd#d$ejZDy).) annotationsN)Callable GeneratorIterator) TracebackType) TYPE_CHECKINGLiteralTypeVarcast) )tarfile) TypeAlias)FileDescriptorOrPath OptExcInfoStrPath) ParamSpecSelfUnpack_FileDescriptorOrPathT)bound_Pr_UnpackableOptExcInfo_R_T1_coT) covariant_T2_coc#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 c/home/jay/workspace/scripts/.codegraph-venv/lib/python3.12/site-packages/jaraco/context/__init__.pypushdr%4s> 99;DHHSM  s*A!AA!AA!c#K|?tjj|jddjdd}tj| t j j|}tj|d5}|j|tddd|tj|y#1swY#xYw#tj|wxYww)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)rr*basenamereplacemkdirurllibrequesturlopenropen extractall_default_filtershutilrmtree)url target_dirreqtfs r$tarballr;Es@WW%%c*229bAII&RTU HHZ"nn$$S) \\#E 2 Cb MMz/M B C j!  C C  j!s6AC06CC ( C4C0 CCC--C0c8d}tj||dS)NcfdS)Nc$|||SN)memberr*f1f2s r$z?_compose_tarfile_filters..compose_two..ssBr&$'7$>r@)rBrCs``r$ compose_twoz-_compose_tarfile_filters..compose_twors >>rEc|Sr?r@)rAr*s r$rDz*_compose_tarfile_filters..usvrE) functoolsreduce)filtersrFs r$_compose_tarfile_filtersrKqs?   K2M NNrEcN|jjdd\}|_|S)N/)namesplit)rAr*_s r$strip_first_componentrRxs%[[&&sA.NAv{ MrEcT dd}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>dfd }tj|S)Nc?K|i|5}|5}|ddddddy#1swYxYw#1swYyxYwwr?r@)argskwargssavedresinnerouters r$composedz/_compose..compose_two..composedsL'' 5%, #       s( A 7+7 A4 7AA)rV_P.argsrW _P.kwargsreturnzGenerator[_T2_co]) contextlibcontextmanager)rZr[r\s`` r$rFz_compose..compose_twos ((22rE)rZz7Callable[_P, contextlib.AbstractContextManager[_T1_co]]r[z=Callable[[_T1_co], contextlib.AbstractContextManager[_T2_co]]r_9Callable[_P, contextlib._GeneratorContextManager[_T2_co]])rHrIreversed)cmgrsrFs r$_composeres;83F3L3 C3   K% 99rEcJ|\}}}|tjtjtjfvrl|jtj k(rOtj |tjtjztjz||y)z> Add support for removing read-only files on Windows. N) rrmdirremoveunlinkerrnoEACCESchmodstatS_IRWXUS_IRWXGS_IRWXO)funcr*exc_inforQexcs r$remove_readonlyrtsgIAsA "))RYY//CII4M t||dll2T\\AB T  rEctjdk(r3tjt dt j tSt j S)NWindowsCallable[..., None])onerror)platformsystemrHpartialr r5r6rtr@rEr$robust_removerr|sH ??  )  & 6# ]]rEc#hKtj} |||y#||wxYww)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 r$rrs0!Hs2% 2 /2)rc#Kd|vrdnd}|5}|d||g}|jd|gt|z|rtjnd}tj||||dddy#1swYyxYww)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)extendbool subprocessDEVNULL check_call)r7branchquietdest_ctxexerepo_dircmdstreams r$ repo_contextrs~*C<%TC xGS(+ J'$v,67',##$c&@ sA=AA1( A=1A:6A=ceZdZUdZdZded<effddZddZe ddZ e ddZ e dd Z dd Z dd Zed  dd ZddZy) 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 )NNNrrrc||_yr?) exceptions)selfrs r$__init__zExceptionTrap.__init__+s $rEc|Sr?r@rs r$ __enter__zExceptionTrap.__enter__. rEc |jdSNrrrrs r$typezExceptionTrap.type1}}QrEc |jdS)NrNrrs r$valuezExceptionTrap.value5rrEc |jdS)Nrrs r$tbzExceptionTrap.tb9rrEcV|d}|xrt||j}|r||_|Sr) issubclassrrr)rrrexc_typematchess r$__exit__zExceptionTrap.__exit__=s1A;Dz(DOOD $DMrEc,t|jSr?)rrrs r$__bool__zExceptionTrap.__bool__GsDIIrE_testcLtjdfd }|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 cxtj5}|i|dddS#1swYxYwr?)rr)rVrWtraprrqrs r$wrapperz%ExceptionTrap.raises..wrapperZs=t/ &4d%f% &;  & &s 09)rVr]rWr^r_r)rHwraps)rrqrrs``` r$raiseszExceptionTrap.raisesJs'      rEcD|j|tjS)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_)rrqs r$passeszExceptionTrap.passesbs{{4x}}{55rEN)rztuple[type[BaseException], ...]r_r)r_type[BaseException] | None)r_BaseException | None)r_TracebackType | None)rrzUnpack[_UnpackableOptExcInfo]r_z*builtins.type[BaseException] | None | bool)r_r)rqCallable[_P, _R]rzCallable[[ExceptionTrap], bool]r_$functools._Wrapped[_P, _R, _P, bool])rqrr_r)__name__ __module__ __qualname____doc__rr__annotations__ Exceptionrrpropertyrrrrrrrrr@rEr$rrsB,Hj+FO\%      0 4SW$0O -06rErceZdZdZy)suppressz A version of contextlib.suppress with decorator support. >>> @suppress(KeyError) ... def key_error(): ... {}[''] >>> key_error() N)rrrrr@rEr$rrssrErcBeZdZdZ d ddZddZ d dZy) 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)() c ||_||_yr?)actioncode)rrrs r$rzon_interrupt.__init__s  rEc|Sr?r@rs r$rzon_interrupt.__enter__rrEc|tus|jdk(ry|jdk(rt|j||jdk(S)Nignoreerrorr)KeyboardInterruptr SystemExitr)rexctypeexcinstexctbs r$rzon_interrupt.__exit__sG + +t{{h/F [[G #TYY'W 4{{j((rEN)rrN)rintrz&Literal['ignore', 'suppress', 'error']r)rrrrrrr_z None | bool)rrrrrrrr@rEr$rr~sS*XYQT<  )+ )& )$ )  )rEr)r"rr_Iterator[StrPath]r?)r7strr8zStrPath | Noner_r)rAtarfile.TarInfor*objectr_r)rdzUnpack[tuple[Callable[[_T1_co], contextlib.AbstractContextManager[_T2_co]], Callable[_P, contextlib.AbstractContextManager[_T1_co]]]]r_rb)rqz*Callable[[_FileDescriptorOrPathT], object]r*rrrztuple[object, OSError, object]r_None)r_rw)rzCallable[[str], object]r_Generator[str]) r7rrz str | Nonerrrz4Callable[[], contextlib.AbstractContextManager[str]]r_r)E __future__rbuiltinsr`rjrHrrryr5rmrsysr~urllib.requestr/collections.abcrrrtypesrtypingrr r r version_infor backportsr _typeshedrrrtyping_extensionsrrrrrtupler BaseExceptionrrrrrrar%r;rKrR data_filterr4re tarball_cwdrtr|r6rr{robust_temp_dirrrrContextDecoratorrr@rEr$rsp"    99C$$/! CC99$ (< 4B#($DD$y  T] T * T *     ("("VO  +7+>+>@UV*: *:?*:Zug&   4 - $  06 "$)##Hn6FG ET    C   :i6i6Xz""J$?$?():..()rE