Method kill_worker

tests/conftest.py:261–270

(self, worker_id: int, signal: int = signal.SIGTERM, wait=True)

        self.kill_process("server")

    def kill_worker(self, worker_id: int, signal: int = signal.SIGTERM, wait=True):
        table = self.command(["worker", "info", str(worker_id)], as_table=True)
        pid = table.get_row_value("Process pid")
        process = self.find_process_by_pid(int(pid))
        if process is None:
            raise Exception(f"Worker {worker_id} not found")

        process = self.kill_process(process.name, signal=signal)
        if wait:
            wait_until(lambda: process.poll() is not None)

    def find_process_by_pid(self, pid: int) -> Optional[ManagedProcess]:
        for p in self.processes: