a oEb$.@s8ddlmZddlZddlmZddlZddlmZmZm Z ddl Z ddl m Z m Z ddlmZddlmZddlZddlmZdd lmZdd lmZerdd lmZmZd Zd ZddZddZeddddeefddZ e Z!d)ddZ"d*ddddddZ#d+ddd d!Z$d,ddd"d#Z%d-d%d&Z&d.d'd(Z'dS)/) 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|jjfS)Nr) http.client urllib.errorOSErrorclient HTTPException TimeoutErrorerrorURLError)httpurllibr!S/home/ja/django-apps/lartica_env/lib/python3.9/site-packages/pandas/_testing/_io.py_get_default_network_errorsFsr#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"dec]sz+optional_args..wrapper..decrr!)lencallable)r&r(r)Z is_decoratingr%r')r&r(r"wrapper[szoptional_args..wrapperr)r'r.r!r-r" optional_argsOs  r/zhttps://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 Exceptiongetattrhasattrr1strany isinstance)r&r(errr0_skip_on_messagescheck_before_test error_classespytestraise_on_error skip_errnosturlr5r"r.s6 znetwork..wrapper)rEr#networkr)rHrIrFrCrDrGrBr.r!rAr"rJks\"rJcCsX|durt}z,t|Wdn1s.0YWn|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 NFT)r#r)rIrDr!r!r"r8s " r8rz"FilePath | ReadPickleBuffer | NonezDataFrame | Series)objpathreturncCs^|}|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 pdZ to_pickleZ read_pickle)rKrL_pathZ temp_pathr!r!r"round_trip_pickles   rRz str | None)rLcCsdddl}|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___)rE importorskipPathr )writerreaderrLrErUrKr!r!r"round_trip_pathlib1s   *rXcCsdddl}|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___)rErTlocalr )rVrWrLrEZ LocalPathrKr!r!r"round_trip_localpathNs   *rZtestc Cs|f}d}d}|dkr.tj}d}||f}d}nR|dkr>tj}nB|dkrNtj}n2|dkrbtd j}n|d krrt}nt d ||||d }t |||Wd n1s0Yd S)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. wbwritezipwwritestrgzipbz2ZzstdZ zstandardxzzUnrecognized compression type: )modeN) zipfileZipFileraGzipFilerbBZ2Filer openr ValueErrorr;) compressionrLdatadestr&rdmethodZcompress_methodr%r!r!r"write_to_compressedks& rocCs:ddlm}m}|dur.|D] }||qn||dS)Nr)close get_fignums)Zmatplotlib.pyplotrprq)Zfignum_closerqr!r!r"rps   rp)N)N)N)N)r[)N)( __future__rrb functoolsrratypingrrrreZpandas._typingrr Z pandas.compatr Zpandas.compat._optionalr ZpandasrPZpandas._testing._randomr Zpandas._testing.contextsr Zpandas.io.commonrrrZ_network_error_messagesZ_network_errno_valsr#r/rJZwith_connectivity_checkr8rRrXrZrorpr!r!r!r"sF           ! 0