o d.0@sFddlmZddlZddlmZddlZddlZddlZddlZddl m Z m Z m Z ddl Z ddlmZmZddlmZddlmZddlZddlmZdd lmZdd lmZe rbdd lmZmZd Zd Z ddZ!ddZ"e"dddde efd1ddZ#d2d3ddZ$ d2d4d#d$Z%d2d5d&d'Z&d2d5d(d)Z'd6d7d,d-Z(d2d8d/d0Z)dS)9) 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<cCs(ddl}ddl}t|jjt|jjtj fS)Nr) http.client urllib.errorOSErrorclient HTTPException TimeoutErrorerrorURLErrorsockettimeout)httpurllibr#G/app/.heroku/python/lib/python3.10/site-packages/pandas/_testing/_io.py_get_default_network_errorsIsr%cstfdd}|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) csJfdd} otdkotd}|r#d}d||S|S)Ncs|gRiSNr#)f)args decoratorkwargsr#r$decfsz+optional_args..wrapper..decrr#)lencallable)r(r*r+Z is_decoratingr'r))r(r*r$wrapperdszoptional_args..wrapperr)r)r0r#r/r$ optional_argsXs  r1zhttps://www.google.comFurlstrraise_on_errorboolcheck_before_testc sBddldur td_tfdd}|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. rNTc srstsdz|i|WStys}zNt|dd}|s5t|dr5t|jdd}|vrAd|t|tfddDrXd|t|r_r`d|WYd}~dSd}~ww) Nzsz+network..wrapper..z;Skipping test because exception message is known and error z4Skipping test due to lack of connectivity and error ) can_connectskip Exceptiongetattrhasattrr8r3any isinstance)r(r*errr7_skip_on_messagesr6 error_classespytestr4 skip_errnostr2r<r$r0s6 znetwork..wrapper)rIr%networkr)rKr2r4r6rHrJrGr0r#rFr$rLus\rLreturncCsx|durt}z*t|dd}|jdkr WdWdSWdWdS1s*wYWdS|y;YdSw)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)r2rHresponser#r#r$r>s   r>objrpath"FilePath | ReadPickleBuffer | NoneDataFrame | SeriescCs^|}|durdtdd}t|}t||t|WdS1s(wYdS)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 r pdZ to_pickleZ read_pickle)rRrS_pathZ temp_pathr#r#r$round_trip_pickles  $rZ str | NonecCfddl}|dj}|durd}t|}||||||}Wd|S1s,wY|S)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. rNpathlibZ ___pathlib___)rI importorskipPathr )writerreaderrSrIr_rRr#r#r$round_trip_pathlib7    rbcCr\)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.pathZ___localpath___)rIr^localr )r`rarSrIZ LocalPathrRr#r#r$round_trip_localpathTrcretestdestc Cs|f}d}d}|dkrtj}d}||f}d}nI|dkr7tj}d}tj|d}t|} t||_|| f}d}n)|d kr?t j }n!|d krGt j }n|d krQt d j}n|d krYt}ntd||||d} t| ||WddS1sywYdS)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)nameaddfilegzipbz2ZzstdZ zstandardxzzUnrecognized compression type: )modeN)zipfileZipFiletarfileTarFileTarInfoioBytesIOr-sizerpGzipFilerqBZ2Filer openr ValueErrorrA) compressionrSdatargr(rsmethodZcompress_methodfilebytesr'r#r#r$write_to_compressedqs8    "rNonecCs<ddlm}m}|dur|D]}||qdS||dS)Nr)close get_fignums)Zmatplotlib.pyplotrr)Zfignum_closerr#r#r$rs    r)r2r3r4r5r6r5r&)rMr5)rRrrSrTrMrU)rSr[)rf)rgr3)rMr)* __future__rrq functoolsrrpryrrvtypingrrrrtZpandas._typingrr Z pandas.compatr Zpandas.compat._optionalr ZpandasrXZpandas._testing._randomr Zpandas._testing.contextsr Zpandas.io.commonrrrZ_network_error_messagesZ_network_errno_valsr%r1rLr>rZrbrerrr#r#r#r$sL        #   8