o i(@sUddlmZddlZddlmmZddlZddl Z ddl m Z m Z m Z ddlmZmZddlmZmZmZddlmZmZddlZddlmZddlmZd d lmZd d lm Z m!Z!m"Z"m#Z#d d l$m%Z%d d l&m'Z'e j(dkryddl)m*Z*da+de,d<da-de,d<da.dSddZ/edTdd Z0dUd$d%Z1dVd(d)Z2ej3d*d+dWd0d1Z4ej3d*d2dXd7d8Z5ej3d*d2dYd;d<Z6ej7d=ed>dZd?d@Z8ej7d[dAdBZ9ej7d\dCdDZ:GdEdFdFZ;ej7dGdHd]dIdJZej7d_dQdRZ?dS)`) annotationsN)Callable GeneratorIterator) ExitStackcontextmanager)isasyncgenfunctioniscoroutinefunctionismethod)Anycast) SubRequest)Exit)get_available_backends)current_async_libraryget_async_backendreset_current_async_libraryset_current_async_library)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'Q/home/ubuntu/transcripts/venv/lib/python3.10/site-packages/anyio/pytest_plugin.pyextract_backend_and_options s r) backend_namer"backend_optionsdict[str, Any]Iterator[TestRunner]c cstdur't|}tatdurt|}tt||pi}t| |at d7a zRtVWt d8a t s|d}t|u}|snt d|fdt|fdt vsSt trXt tndt |d}dd|i}tt |d}}tdaadSdSt d8a t sd}t|u}|st d|fdt|fdt vst trt tndt |d}dd|i}tt |d}}tdaaw)Nr)is not)z%(py0)s is not %(py3)sr)py0py3zassert %(py5)spy5)rrrrrrcallbackr enter_contextcreate_test_runner_runner_leases @pytest_ar_call_reprcompare @py_builtinslocals_should_repr_global_name _safereprAssertionError_format_explanationclose)r*r+asynclibtoken @py_assert2 @py_assert1 @py_format4 @py_format6r'r'r( get_runner*s2  x x rEparser pytest.ParserNonecCs|jdddddS)N anyio_modestrictz-AnyIO plugin mode (either "strict" or "auto"))defaulthelp)addini)rFr'r'r(pytest_addoptionHs  rNconfig pytest.ConfigcCsV|dd|ddkr%|jdr'|ddkr)|tdddSdSdSdS) NmarkerszManyio: mark the (coroutine function) test to be run asynchronously via anyio.rIautoasyncio 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)rOr'r'r(pytest_configurePs  r\T) 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 r_r kwargsrc;s|jrtrtjt|jurj|j}n}t|\}}r(||d<r.||d<t||&}t|rB| ||EdHn| ||VWddSWddS1s\wYdS)Nrar_) instancer type__self____func____get__r)rErrun_asyncgen_fixture run_fixture)rar_rb local_funcr*r+runnerfunchas_backend_arghas_request_argr'r(wrapperes(  "z%pytest_fixture_setup..wrapperrar_)rar r_r rbr rr )rmrr fixturenamesargnames)r^r_rporiginal_argnamer'rlr(pytest_fixture_setupcs$ rv)tryfirst collectorpytest.Module | pytest.ClassnameobjcCs|||r?t|dr|jjn|}t|rA|jddk}|d}t|dd}|s5|s5t dd|DrCt j d |dSdSdSdS) N hypothesisrIrRanyio pytestmarkr'css|]}|jdkVqdS)r}N)rz).0markerr'r'r( sz,pytest_pycollect_makeitem..ra) istestfunctionhasattrr| inner_testr rOrVget_closest_markergetattranyrZmark usefixtures)rxrzr{ inner_funcanyio_auto_moder 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) Nrbr rrHcs<t}||WddS1swYdSN)rErun_test)rbrk)r*r+ original_funcr'r(run_with_hypothesiss "z/pytest_pyfunc_call..run_with_hypothesisrar|csi|]}||qSr'r')rarg)funcargsr'r( sz&pytest_pyfunc_call..T)rbr rrH)rgetr)rr{r|r __qualname__r _fixtureinfortrErrrr!rKeyboardInterrupt SystemExit)rrrtestargsrkexcgrpexcr')r*r+rrr(pytest_pyfunc_calls<            rmodule)scopeparamscC|jSr)paramrrr'r'r(raracCst|tr|S|dS)Nrr!r"rqr'r'r(anyio_backend_name rcCst|triS|dS)Nrrrqr'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.SocketKindrrHcCs||_tt|_dSr)_kindsetint _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_INET6rr3rbindOSError getsocknameradd)rrfamiliesportstacksockaddrr'r'r(__call__s4     zFreePortFactory.__call__)rrrrH)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_factory"rrCallable[[], int]rcC|Srr')rr'r'r( free_tcp_port'rrcCrrr')rr'r'r( free_udp_port,rr)rrrr)r*r"r+r,rr-)rFrGrrH)rOrPrrH)r^r r_r rr`)rxryrzr"r{rrrH)rr rr)r_r rr )rar rr")rar rr,)rr)rrrr)rrrr)@ __future__rbuiltinsr8_pytest.assertion.rewrite assertionrewriter6rsyscollections.abcrrr contextlibrrinspectrr r typingr r rZ_pytest.fixturesr _pytest.outcomesrr_core._eventlooprrrr_core._exceptionsrabcr version_infoexceptiongrouprr__annotations__rr5r)rErNr\hookimplrvrrfixturerarrrrrrrr'r'r'r(sZ"                0   $    =