(self)
| 7 | self.writer = None |
| 8 | |
| 9 | async def connect(self): |
| 10 | try: |
| 11 | self.reader, self.writer = await asyncio.open_unix_connection(self.socket_path) |
| 12 | print(f"Connected to {self.socket_path}") |
| 13 | except Exception as e: |
| 14 | print(f"Connection failed: {e}") |
| 15 | |
| 16 | async def send(self, message: str): |
| 17 | if self.writer is not None: |
no outgoing calls
no test coverage detected