Method __init__

python/tests/test_transport.py:58–64

(self)

def _feed(process: _Process, envelope: pb.Envelope) -> None:
    data = envelope.SerializeToString()
    process.stdout.feed_data(_VarintBytes(len(data)) + data)


async def _request(process: _Process) -> pb.Envelope:
    return pb.Envelope.FromString(await _read_delimited(process.requests))


@pytest.mark.asyncio
async def test_unary_round_trip_and_error_mapping() -> None: