9 from qemu.qmp import ConnectError, Runstate
10 from qemu.qmp.protocol import AsyncProtocol, StateError
11 from qemu.qmp.util import asyncio_run, create_task
65     async def _do_recv(self) -> None:
69     def _do_send(self, msg: None) -> None:
72     async def send_msg(self) -> None:
75     async def simulate_disconnect(self) -> None:
77         Simulates a bottom-half disconnect.
88         returns bottom half errors as a pre-condition to allowing the
99     async def _do_recv(self) -> str:
105     def _do_send(self, msg: str) -> None:
109     async def send_msg(self, msg: str) -> None:
146         # I don't *fully* understand why, but it takes *two* un-accepted
162     def setUp(self):  member in Smoke
186             'qemu.qmp.protocol'
199             'qemu.qmp.protocol.Steve'
210     def setUp(self):  member in TestBase
228         Decorator; adds SetUp and TearDown to async tests.
367         # result of the task with a nearly-zero timeout.
441     def setUp(self):  member in FakeSession
442         super().setUp()
458         """Test the full state lifecycle (via connect) with a no-op session."""
464         """Test the full state lifecycle (via accept) with a no-op session."""
481             [f"DEBUG:{logname}:<-- None"],
498             [f"DEBUG:{logname}:--> None"],
572     def setUp(self):  member in SimpleSession
573         super().setUp()
574         self.server = LineProtocol(type(self).__name__ + '-server')