a Sic/@sXddlmZddlZddlmZddlZddlZddlZddlZddl m Z m Z m Z ddl Z ddlmZmZddlmZddlmZddlZddlmZdd lmZdd lmZe rdd lmZmZd Zd Z ddZ!ddZ"e"dddde efddZ#d,ddddZ$d-ddddddZ%d.d d!d"d#Z&d/d d!d$d%Z'd0d'd(Z(d1d)dd*d+Z)dS)2) 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#O/var/www/html/django/DPS/env/lib/python3.9/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) csNfdd} o,tdko,td}|rFd}d||S|SdS)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.comFc sBddldurtd_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 sr"s"ts"dz|i|WSty}zt|dd}|sjt|drjt|jdd}|vrd|t|tfddDrd|t|rr‚nd|WYd}~n d}~00dS) Nzz+network..wrapper..z;Skipping test because exception message is known and error z4Skipping test due to lack of connectivity and error ) can_connectskip Exceptiongetattrhasattrr3strany isinstance)r(r*errr2_skip_on_messagescheck_before_test error_classespytestraise_on_error skip_errnosturlr7r$r0s6 znetwork..wrapper)rGr%networkr)rJrKrHrErFrIrDr0r#rCr$rLts\"rLbool)returncCsx|durt}zLt|dd,}|jdkr:WdWdSWdn1sN0YWn|ynYdS0dSdS)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)rKrFresponser#r#r$r:s 4 r:rz"FilePath | ReadPickleBuffer | NonezDataFrame | Series)objpathrNcCs^|}|durdtdd}t|&}t||t|WdS1sP0YdS)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 pd to_pickle read_pickle)rSrT_pathZ temp_pathr#r#r$round_trip_pickles   r[z str | None)rTcCsdddl}|dj}|dur d}t|(}||||||}Wdn1sV0Y|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___)rG importorskipPathr )writerreaderrTrGr^rSr#r#r$round_trip_pathlib9s   *racCsdddl}|dj}|dur d}t|(}||||||}Wdn1sV0Y|S)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___)rGr]localr )r_r`rTrGZ LocalPathrSr#r#r$round_trip_localpathVs   *rctestc Cs|f}d}d}|dkr.tj}d}||f}d}n|dkrntj}d}tj|d}t|} t||_|| f}d}nR|d kr~t j }nB|d krt j }n2|d krt d j}n|d krt}ntd||||d} t| ||Wdn1s0YdS)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)zipfileZipFiletarfileTarFileTarInfoioBytesIOr-sizermGzipFilernBZ2Filer openr ValueErrorr=) compressionrTdatadestr(rrmethodZcompress_methodfilebytesr'r#r#r$write_to_compressedss6    rNonecCs:ddlm}m}|dur.|D] }||qn||dS)Nr)close get_fignums)matplotlib.pyplotrr)Zfignum_closerr#r#r$rs   r)N)N)N)N)rd)N)* __future__rrn functoolsrrmrxrrutypingrrrrspandas._typingrr pandas.compatr pandas.compat._optionalr pandasrWpandas._testing._randomr pandas._testing.contextsr pandas.io.commonrrrZ_network_error_messagesZ_network_errno_valsr%r1rLr:r[rarcrrr#r#r#r$sJ         # 8