(self, family)
| 40 | |
| 41 | class TestTCPServer(TCPServer): |
| 42 | def __init__(self, family): |
| 43 | super().__init__() |
| 44 | self.streams = [] # type: List[IOStream] |
| 45 | self.queue = Queue() # type: Queue[IOStream] |
| 46 | sockets = bind_sockets(0, "localhost", family) |
| 47 | self.add_sockets(sockets) |
| 48 | self.port = sockets[0].getsockname()[1] |
| 49 | |
| 50 | def handle_stream(self, stream, address): |
| 51 | self.streams.append(stream) |
nothing calls this directly
no test coverage detected