(self,
basecls=None, # type: Optional[Type[Packet]]
external_obj_pipe=None # type: Optional[ObjectPipe[bytes]]
)
| 41 | test_socket_mutex = Lock() |
| 42 | |
| 43 | def __init__(self, |
| 44 | basecls=None, # type: Optional[Type[Packet]] |
| 45 | external_obj_pipe=None # type: Optional[ObjectPipe[bytes]] |
| 46 | ): |
| 47 | # type: (...) -> None |
| 48 | global open_test_sockets |
| 49 | self.basecls = basecls |
| 50 | self.paired_sockets = list() # type: List[TestSocket] |
| 51 | self.ins = external_obj_pipe or ObjectPipe(name="TestSocket") # type: ignore |
| 52 | self._has_external_obj_pip = external_obj_pipe is not None |
| 53 | self.outs = None |
| 54 | open_test_sockets.append(self) |
| 55 | |
| 56 | def __enter__(self): |
| 57 | # type: () -> TestSocket |
no test coverage detected