(self, environment_id: str, **kwargs: Any)
| 52 | self.stop_calls: list[tuple[str, dict[str, Any]]] = [] |
| 53 | |
| 54 | def poll(self, environment_id: str, **kwargs: Any) -> Any: |
| 55 | self.poll_calls.append({"environment_id": environment_id, **kwargs}) |
| 56 | if not self._poll_script: |
| 57 | raise _StopTest("poll script exhausted") |
| 58 | nxt = self._poll_script.pop(0) |
| 59 | if isinstance(nxt, BaseException): |
| 60 | raise nxt |
| 61 | return nxt |
| 62 | |
| 63 | def ack(self, work_id: str, **kwargs: Any) -> None: |
| 64 | self.ack_calls.append((work_id, kwargs)) |
no test coverage detected