o h'@sUddlmZddlZddlZddlmZmZmZddlm Z m Z ddl m Z m Z mZddlmZmZddlZddlZddlmZddlmZd d lmZmZd d lmZd d lmZejd krfddl m!Z!da"de#d<da$de#d<da%dRddZ&e dSddZ'dTd#d$Z(dUd'd(Z)ej*d)d*dVd/d0Z+ej*d)d1dWd6d7Z,ej*d)d1dXd:d;Z-ej.dd?Z/ej.dZd@dAZ0ej.d[dBdCZ1GdDdEdEZ2ej.dFdGd\dHdIZ3ej.dFdGd\dJdKZ4ej.d]dNdOZ5ej.d^dPdQZ6dS)_) annotationsN)Callable GeneratorIterator) ExitStackcontextmanager)isasyncgenfunctioniscoroutinefunctionismethod)Anycast) SubRequest)Exit)get_all_backendsget_async_backend)iterate_exceptions) TestRunner) )ExceptionGroupzTestRunner | None_current_runnerzExitStack | None _runner_stackbackendobjectreturntuple[str, dict[str, Any]]cCsft|tr |ifSt|tr/t|dkr/t|dtr/t|dtr/ttttttff|Std)Nrrz@anyio_backend must be either a string or tuple of (string, dict)) isinstancestrtuplelendictr r TypeError)rr$U/var/www/html/pro-man-master/venv/lib/python3.10/site-packages/anyio/pytest_plugin.pyextract_backend_and_optionss r& backend_namerbackend_optionsdict[str, Any]Iterator[TestRunner]ccstdur.t|}tatjddur"tj|}ttjj ||p%i}t | |at d7a ztVWt d8a t sMtdusCJt daadSdSt d8a t sctdus[Jt daawNr)rrrrsniffiocurrent_async_library_cvargetsetcallbackreset enter_contextcreate_test_runner_runner_leasesclose)r'r(asynclibtokenr$r$r% get_runner%s2     r8parser pytest.ParserNonecCs|jddddddS)N anyio_modestrictz-AnyIO plugin mode (either "strict" or "auto")string)defaulthelptype)addini)r9r$r$r%pytest_addoptionCs  rCconfig pytest.ConfigcCsV|dd|ddkr%|jdr'|ddkr)|tdddSdSdSdS) NmarkerszManyio: mark the (coroutine function) test to be run asynchronously via anyio.r<autoasyncio asyncio_modezlAnyIO auto mode has been enabled together with pytest-asyncio auto mode. This may cause unexpected behavior.r)addinivalue_linegetini pluginmanager has_pluginissue_config_time_warningpytestPytestConfigWarning)rDr$r$r%pytest_configureLs  rQT) hookwrapper fixturedefr requestGenerator[Any]c#sd fdd }|jtstrNd|jvrN||_|j}d|jvs.|jd 7_d|jvs<|jd 7_z dVW|_||_S|_||_wdVS) N anyio_backendr rTr kwargsrc;s|jrtrtjt|jurj|j}n}t|\}}r(||d<r.||d<t||&}t|rB| ||EdHn| ||VWddSWddS1s\wYdS)NrVrT) instancer rA__self____func____get__r&r8rrun_asyncgen_fixture run_fixture)rVrTrW local_funcr'r(runnerfunchas_backend_arghas_request_argr$r%wrapperas(  "z%pytest_fixture_setup..wrapperrVrT)rVr rTr rWr rr )rarr fixturenamesargnames)rSrTrdoriginal_argnamer$r`r%pytest_fixture_setup_s$ rj)tryfirst collectorpytest.Module | pytest.ClassnameobjcCs|||r?t|dr|jjn|}t|rA|jddk}|d}t|dd}|s5|s5t dd|DrCt j d |dSdSdSdS) N hypothesisr<rGanyio pytestmarkr$css|]}|jdkVqdS)rqN)rn).0markerr$r$r% sz,pytest_pycollect_makeitem..rV) istestfunctionhasattrrp inner_testr rDrKget_closest_markergetattranyrOmark usefixtures)rlrnro inner_funcanyio_auto_modert own_markersr$r$r%pytest_pycollect_makeitems   r pyfuncitem bool | Nonec sd fdd }|jd}|rt|\t|jdr3|jjjj|jkr1tr1||jj_dSt|jr|jfd d |j j D}t 2}z | |j|Wnt yt}zt|D] }t|tttfrn||qad}~wwWdd S1swYd SdS) NrWr rr;cs<t}||WddS1swYdSN)r8run_test)rWr_)r'r( original_funcr$r%run_with_hypothesiss "z/pytest_pyfunc_call..run_with_hypothesisrVrpcsi|]}||qSr$r$)rsarg)funcargsr$r% sz&pytest_pyfunc_call..T)rWr rr;)rr.r&rwrorprx __qualname__r _fixtureinforhr8rrrrrKeyboardInterrupt SystemExit)rrrtestargsr_excgrpexcr$)r'r(rrr%pytest_pyfunc_calls<            rmodule)scopeparamscC|jSr)paramrfr$r$r%rVrVcCst|tr|S|dS)Nrrrrer$r$r%anyio_backend_name rcCst|triS|dSr+rrer$r$r%anyio_backend_optionsrrc@s4eZdZdZdddZeddd ZddddZd S)FreePortFactoryaO Manages port generation based on specified socket kind, ensuring no duplicate ports are generated. This class provides functionality for generating available free ports on the system. It is initialized with a specific socket kind and can generate ports for given address families while avoiding reuse of previously generated ports. Users should not instantiate this class directly, but use the ``free_tcp_port_factory`` and ``free_udp_port_factory`` fixtures instead. For simple uses cases, ``free_tcp_port`` and ``free_udp_port`` can be used instead. kindsocket.SocketKindrr;cCs||_tt|_dSr)_kindr/int _generated)selfrr$r$r%__init__szFreePortFactory.__init__cCr)z The type of socket connection (e.g., :data:`~socket.SOCK_STREAM` or :data:`~socket.SOCK_DGRAM`) used to bind for checking port availability )r)rr$r$r%rszFreePortFactory.kindNfamilysocket.AddressFamily | Nonerc Cs|dur|g}n tjg}tjr|tj d}tP}|D]1}|t||j}|tjkr1dnd}z |||fWn t yFYnw|sO| d}q||j vrd|j ||WdSWdn1snwYq)z Return an unbound port for the given address family. :param family: if omitted, both IPv4 and IPv6 addresses will be tried :return: a port number NTrz::1z 127.0.0.1r) socketAF_INEThas_ipv6appendAF_INET6rr2rbindOSError getsocknameradd)rrfamiliesportstacksockaddrr$r$r%__call__s4     zFreePortFactory.__call__)rrrr;)rrr)rrrr)__name__ __module__r__doc__rpropertyrrr$r$r$r%rs   rsession)rcC ttjSr)rr SOCK_STREAMr$r$r$r%free_tcp_port_factory rcCrr)rr SOCK_DGRAMr$r$r$r%free_udp_port_factoryrrCallable[[], int]rcC|Srr$)rr$r$r% free_tcp_port#rrcCrrr$)rr$r$r% free_udp_port(rr)rrrr)r'rr(r)rr*)r9r:rr;)rDrErr;)rSr rTr rrU)rlrmrnrrorrr;)rr rr)rTr rr )rVr rr)rVr rr))rr)rrrr)rrrr)7 __future__rrsyscollections.abcrrr contextlibrrinspectrr r typingr r rOr,_pytest.fixturesr _pytest.outcomesr_core._eventlooprr_core._exceptionsrabcr version_infoexceptiongrouprr__annotations__rr4r&r8rCrQhookimplrjrrfixturerVrrrrrrrr$r$r$r%sZ              0   $    =