python/tests/protocol.py

35     def __init__(self, name=None):  argument
38         super().__init__(name)
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.
81         state without fully quiescing it back to IDLE. This is normally
88         returns bottom half errors as a pre-condition to allowing the
89         loop to return back to IDLE.
95     def __init__(self, name=None):  argument
96         super().__init__(name)
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
168             "<NullProtocol runstate=IDLE>"
174             Runstate.IDLE
179             self.proto.name,
185             self.proto.logger.name,
193             self.proto.name,
198             self.proto.logger.name,
204             "<NullProtocol name='Steve' runstate=IDLE>"
212         self.assertEqual(self.proto.runstate, Runstate.IDLE)
216         self.assertEqual(self.proto.runstate, Runstate.IDLE)
246         Runstate.IDLE,
254         Runstate.IDLE,
266             for state in states:
270                     state,
271                     msg=f"Expected state '{state.name}'",
279 class State(TestBase):  class
288             Runstate.IDLE,
345         state = await self.proto.runstate_changed()
346         self.assertEqual(state, Runstate.CONNECTING)
367         # result of the task with a nearly-zero timeout.
369         state = await self.proto.runstate_changed()
370         self.assertEqual(state, Runstate.CONNECTING)
384         state = await self.proto.runstate_changed()
385         self.assertEqual(state, Runstate.CONNECTING)
394         self.assertEqual(context.exception.state, Runstate.CONNECTING)
395         self.assertEqual(context.exception.required, Runstate.IDLE)
410         # Kick the loop to coerce the state change
415         await self._watch_runstates(Runstate.DISCONNECTING, Runstate.IDLE)
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."""
473         logname = self.proto.logger.name
481             [f"DEBUG:{logname}:<-- None"],
489         logname = self.proto.logger.name
498             [f"DEBUG:{logname}:--> None"],
514         self.assertEqual(context.exception.state, current_state)
515         self.assertEqual(context.exception.required, Runstate.IDLE)
550              " Call disconnect() to return to IDLE state."),
565              " Call disconnect() to return to IDLE state."),
574         self.server = LineProtocol(type(self).__name__ + '-server')