Function core

python/tests/test_transport.py:104–117

()

    t, proc = _make_transport()

    async def core() -> None:
        req = await _read_request(proc)
        assert req.kind == pb.Envelope.KIND_REQUEST
        assert req.WhichOneof("payload") == "list_machines_req"
        _feed(
            proc,
            pb.Envelope(
                id=req.id,
                kind=pb.Envelope.KIND_RESPONSE,
                machine_list=m_pb.MachineList(
                    items=[m_pb.MachineSummary(id="m1", hostname="work")]
                ),
            ),
        )

    core_task = asyncio.create_task(core())
    resp = await t.call_unary(pb.Envelope(list_machines_req=m_pb.ListMachinesRequest(presence="any")))