`Vg/@ddlmZddlZddlmZddlZddlZddlZddlZddl m Z m Z m Z ddl Z ddlmZmZddlmZddlmZddlZddlmZdd lmZdd lmZe rdd lmZmZd Zd Z dZ!dZ"e"dddde efdZ#dddZ$ d ddZ%dddZ&dddZ'ddZ(dddZ)y)) annotationsNwraps) TYPE_CHECKINGAnyCallable)FilePathReadPickleBuffer) get_lzma_file)import_optional_dependency)rands) ensure_clean)urlopen) DataFrameSeries) z timed outz Server Hangupz#HTTP Error 503: Service Unavailablez502: Proxy ErrorzHTTP Error 502: internal errorzHTTP Error 502zHTTP Error 503zHTTP Error 403zHTTP Error 400z$Temporary failure in name resolutionzName or service not knownzConnection refusedzcertificate verify)eonh6<cddl}ddl}t|jjt |j jtjfS)Nr) http.client urllib.errorOSErrorclient HTTPException TimeoutErrorerrorURLErrorsockettimeout)httpurllibs T/var/www/html/hubwallet-dev/venv/lib/python3.12/site-packages/pandas/_testing/_io.py_get_default_network_errorsr&Is=  !!   c.tfd}|S)aB allows a decorator to take optional positional and keyword arguments. Assumes that taking a single, callable, positional argument means that it is decorating a function, i.e. something like this:: @my_decorator def function(): pass Calls decorator with decorator(f, *args, **kwargs) cfd} xrtdk(xrtd}|rd}d||S|S)Nc|giSN)fargs decoratorkwargss r%decz+optional_args..wrapper..decfsQ000 0r'rr,)lencallable)r.r0r1 is_decoratingr-r/s`` r%wrapperzoptional_args..wrapperdsK 1# Ks4yA~K(47:K QADq6MJr'r)r/r6s` r% optional_argsr7Xs# 9   Nr'zhttps://www.google.comFc xddl td_tfd}|S)a Label a test as requiring network connection and, if an error is encountered, only raise if it does not find a network connection. In comparison to ``network``, this assumes an added contract to your test: you must assert that, under normal conditions, your test will ONLY fail if it does not have network connectivity. You can call this in 3 ways: as a standard decorator, with keyword arguments, or with a positional argument that is the url to check. Parameters ---------- t : callable The test requiring network connectivity. url : path The url to test via ``pandas.io.common.urlopen`` to check for connectivity. Defaults to 'https://www.google.com'. raise_on_error : bool If True, never catches errors. check_before_test : bool If True, checks connectivity before running the test case. error_classes : tuple or Exception error classes to ignore. If not in ``error_classes``, raises the error. defaults to OSError. Be careful about changing the error classes here. skip_errnos : iterable of int Any exception that has .errno or .reason.erno set to one of these values will be skipped with an appropriate message. _skip_on_messages: iterable of string any exception e for which one of the strings is a substring of str(e) will be skipped with an appropriate message. Intended to suppress errors where an errno isn't available. Notes ----- * ``raise_on_error`` supersedes ``check_before_test`` Returns ------- t : callable The decorated test ``t``, with checks for connectivity errors. Example ------- Tests decorated with @network will fail if it's possible to make a network connection to another URL (defaults to google.com):: >>> from pandas import _testing as tm >>> @tm.network ... def test_network(): ... with pd.io.common.urlopen("rabbit://bonanza.com"): ... pass >>> test_network() # doctest: +SKIP Traceback ... URLError: You can specify alternative URLs:: >>> @tm.network("https://www.yahoo.com") ... def test_something_with_yahoo(): ... raise OSError("Failure Message") >>> test_something_with_yahoo() # doctest: +SKIP Traceback (most recent call last): ... OSError: Failure Message If you set check_before_test, it will check the url first and not run the test on failure:: >>> @tm.network("failing://url.blaher", check_before_test=True) ... def test_something(): ... print("I ran!") ... raise ValueError("Failure") >>> test_something() # doctest: +SKIP Traceback (most recent call last): ... Errors not related to networking will always be raised. rNTcr" s t sjd   |i|S#t$r}t|dd}|s#t |drt|j dd}| vrjd|t |tfdDrjd|t|r rjd|Yd}~yd}~wwxYw)Nzz+network..wrapper..s"I!1779 -Is*-z;Skipping test because exception message is known and error z4Skipping test due to lack of connectivity and error ) can_connectskip Exceptiongetattrhasattrr;strany isinstance) r.r0errr:r@_skip_on_messagescheck_before_test error_classespytestraise_on_error skip_errnosturls @r%r6znetwork..wrappers "]3 KKNseT  d%f% % C$/EWUH5 GT: # I#OPHEI7HII QRUQVWc=1^ J3%P' s0 C#B CC#)rNr&networkr) rQrRrOrLrMrPrKr6rNs ``````` @r%rSrStsAx35 AI 1X   D Nr'c| t} t|d5}|jdk7r dddy dddy#1swYyxYw#|$rYywxYw)a@ Try to connect to the given url. True if succeeds, False if OSError raised Parameters ---------- url : basestring The URL to try to connect to Returns ------- connectable : bool Return True if no OSError (unable to connect) or URLError (bad url) was raised N)r"FT)r&rstatus)rRrMresponses r%rBrBsl 35  S" % #%  %   s0 A ?A A AA A AAc|}|dtdd}t|5}tj||tj|cdddS#1swYyxYw)a Pickle an object and then read it again. Parameters ---------- obj : any object The object to pickle and then re-read. path : str, path object or file-like object, default None The path where the pickled object is written and then read. Returns ------- pandas object The original object that was pickled and then re-read. N__ z __.pickle)r rpd to_pickle read_pickle)objpath_path temp_paths r%round_trip_picklercsY$ E }U2YKy) e )  S)$~~i()))s +AAcddl}|jdj}|d}t|5}||||||}ddd|S#1swYSxYw)a Write an object to file specified by a pathlib.Path and read it back Parameters ---------- writer : callable bound to pandas object IO writing function (e.g. DataFrame.to_csv ) reader : callable IO reading function (e.g. pd.read_csv ) path : str, default None The path where the object is written and then read. Returns ------- pandas object The original object that was serialized and then re-read. rNpathlib ___pathlib___)rN importorskipPathr)writerreaderr`rNrhr_s r%round_trip_pathlibrk9si$   y ) . .D | d !ttDzT$Z ! J! J AA cddl}|jdj}|d}t|5}||||||}ddd|S#1swYSxYw)a Write an object to file specified by a py.path LocalPath and read it back. Parameters ---------- writer : callable bound to pandas object IO writing function (e.g. DataFrame.to_csv ) reader : callable IO reading function (e.g. pd.read_csv ) path : str, default None The path where the object is written and then read. Returns ------- pandas object The original object that was serialized and then re-read. rNzpy.path___localpath___)rNrglocalr)rirjr`rN LocalPathr_s r%round_trip_localpathrqVsi$##I.44I |  d &tyYt_%& J& Jrlc&|f}d}d}|dk(rtj}d}||f}d}n|dk(rTtj}d}tj|}t j |} t||_|| f}d}ne|d k(rtj}nO|d k(rtj}n9|d k(rtd j}n|d k(r t}nt!d||||5} t#| ||dddy#1swYyxYw)a Write data to a compressed file. Parameters ---------- compression : {'gzip', 'bz2', 'zip', 'xz', 'zstd'} The compression type to use. path : str The file path to write the data. data : str The data to write. dest : str, default "test" The destination file (for ZIP only) Raises ------ ValueError : An invalid compression value was passed in. wbwritezipwwritestrtar)nameaddfilegzipbz2zstd zstandardxzzUnrecognized compression type: )modeN)zipfileZipFiletarfileTarFileTarInfoioBytesIOr3sizer{GzipFiler|BZ2Filer openr ValueErrorrE) compressionr`datadestr.rmethodcompress_methodfilebytesr-s r%write_to_compressedrss&"GD D Fe!//d|  !//D) 4 I e}  --  ++  4[AFF  '/:;-HII D )"Q6D!"""s .DDcRddlm}m}||D] }|| y||y)Nr)close get_fignums)matplotlib.pyplotrr)fignum_closers r%rrs/ ~!m F 6N  vr'r+)returnbool)r_rr`z"FilePath | ReadPickleBuffer | NonerzDataFrame | Series)r`z str | None)test)rNone)* __future__rr| functoolsrr{rr!rtypingrrrrpandas._typingr r pandas.compatr pandas.compat._optionalr pandasr\pandas._testing._randomr pandas._testing.contextsrpandas.io.commonrrr_network_error_messages_network_errno_valsr&r7rSrBrcrkrqrrr,r'r%rs"   (>)1$ (" 8 !#-EEPF:>) )6))4::1"p r'