| 55 | |
| 56 | |
| 57 | def _feed(process: _Process, envelope: pb.Envelope) -> None: |
| 58 | data = envelope.SerializeToString() |
| 59 | process.stdout.feed_data(_VarintBytes(len(data)) + data) |
| 60 | |
| 61 | |
| 62 | async def _request(process: _Process) -> pb.Envelope: |
| 63 | return pb.Envelope.FromString(await _read_delimited(process.requests)) |
| 64 | |
| 65 | |
| 66 | @pytest.mark.asyncio |
| 67 | async def test_unary_round_trip_and_error_mapping() -> None: |
| 68 | transport, process = _transport() |
| 69 | call = asyncio.create_task( |
| 70 | transport.call_unary(pb.Envelope(system_bootstrap_req=pb.SystemBootstrapRequest())) |
| 71 | ) |
| 72 | request = await _request(process) |
| 73 | _feed( |
no outgoing calls