(self, stream_id: int)
| 1060 | raise AssertionError(f"Not a command: {event}") |
| 1061 | |
| 1062 | def make_stream(self, stream_id: int) -> layer.CommandGenerator[None]: |
| 1063 | ctx = self.context.fork() |
| 1064 | self.streams[stream_id] = HttpStream(ctx, stream_id) |
| 1065 | yield from self.event_to_child(self.streams[stream_id], events.Start()) |
| 1066 | |
| 1067 | def get_connection( |
| 1068 | self, event: GetHttpConnection, *, reuse: bool = True |
no test coverage detected