o ;i@sddlZddlmZddlmZmZmZmZddlm Z ddl m Z ddl m Z ddlmZmZer8dd lmZGd d d ejZGd d d ejZGdddZdS)N) TracebackType) TYPE_CHECKING CollectionOptionalType)Channel)Server) H2Protocol) CodecBaseStatusDetailsCodecBase) IServablec@sReZdZdejfddZdefddZdddZdd d Z dd d Z dd dZ dS)_ServerreturncCtNNotImplementedErrorselfrC/home/ubuntu/.local/lib/python3.10/site-packages/grpclib/testing.pyget_loopz_Server.get_loopcCrrrrrrr is_servingrz_Server.is_servingNctrrrrrr start_servingz_Server.start_servingcrrrrrrr serve_foreverrz_Server.serve_forevercCdSrrrrrrcloserz _Server.closecsdSrrrrrr wait_closed rz_Server.wait_closedrN) __name__ __module__ __qualname__asyncioAbstractEventLooprboolrrrr r!rrrrrs   rcsbeZdZdeddffdd ZdeddfddZdeddfd d Zdefd d Z dd dZ Z S)_InMemoryTransportprotocolrNcstt|_||_dSr)super__init__r&get_event_loop_loop _protocol)rr* __class__rrr,&s   z_InMemoryTransport.__init__datacCs |jjs|j|dSdSr)r/ connection is_closing data_receivedrr2rrr _write_soon.s z_InMemoryTransport._write_sooncCs|r |j|j|dSdSr)r. call_soonr7r6rrrwrite2sz_InMemoryTransport.writecCsdS)NFrrrrrr46rz_InMemoryTransport.is_closingcCrrrrrrrr 9rz_InMemoryTransport.closer") r#r$r%r r,bytesr7r9r(r4r __classcell__rrr0rr)$sr)c @steZdZdZ  ddeddeedeeddfdd Zde fd d Z d ee e d ee dee ddfddZdS) ChannelForaManages specially initialised :py:class:`~grpclib.client.Channel` with an in-memory transport to a :py:class:`~grpclib.server.Server` Example: .. code-block:: python3 class Greeter(GreeterBase): ... greeter = Greeter() async with ChannelFor([greeter]) as channel: stub = GreeterStub(channel) response = await stub.SayHello(HelloRequest(name='Dr. Strange')) assert response.message == 'Hello, Dr. Strange!' Nservicesr codecstatus_details_codecrcCs||_||_||_dS)a :param services: list of services you want to test :param codec: instance of a codec to encode and decode messages, if omitted ``ProtoCodec`` is used by default :param status_details_codec: instance of a status details codec to encode and decode error details in a trailing metadata, if omitted ``ProtoStatusDetailsCodec`` is used by default N) _services_codec_status_details_codec)rr=r>r?rrrr,Os zChannelFor.__init__cst|j|j|jd|_t|j_|jj|j_|j |_ t |j|jd|_ |j |j _ |j j t|j |j t|j j |j S)z> :return: :py:class:`~grpclib.client.Channel` )r>r?)r r@rArB_serverrr. create_future_server_closed_fut_protocol_factory_server_protocolr_channelr/connection_mader)rrrr __aenter__cs*   zChannelFor.__aenter__exc_typeexc_valexc_tbcsT|jjdus J|jjd|j|jd|j|jIdHdSr)rHr/connection_lostr rGrCr!)rrKrLrMrrr __aexit__~s   zChannelFor.__aexit__)NN)r#r$r%__doc__rrr r r,rrJr BaseExceptionrrOrrrrr<=s.  r<)r&typesrtypingrrrrclientrserverr r*r encoding.baser r _typingr AbstractServerr Transportr)r<rrrrs