Class FakeWork

tests/lib/environments/test_poller_method.py:42–67

Sync resource fake. ``poll_script`` is consumed in order; values are either ``BetaSelfHostedWork``-shaped stubs, ``None`` (no work available), or ``Exception`` instances (raised from poll).

class FakeWork:
    """Sync resource fake. ``poll_script`` is consumed in order; values are
    either ``BetaSelfHostedWork``-shaped stubs, ``None`` (no work available),
    or ``Exception`` instances (raised from poll).
    """

    def __init__(self, poll_script: list[Any]) -> None:
        self._poll_script = list(poll_script)
        self.poll_calls: list[dict[str, Any]] = []
        self.ack_calls: list[tuple[str, dict[str, Any]]] = []
        self.stop_calls: list[tuple[str, dict[str, Any]]] = []

    def poll(self, environment_id: str, **kwargs: Any) -> Any:
        self.poll_calls.append({"environment_id": environment_id, **kwargs})
        if not self._poll_script:
            raise _StopTest("poll script exhausted")
        nxt = self._poll_script.pop(0)
        if isinstance(nxt, BaseException):
            raise nxt
        return nxt

    def ack(self, work_id: str, **kwargs: Any) -> None:
        self.ack_calls.append((work_id, kwargs))

    def stop(self, work_id: str, **kwargs: Any) -> None:
        self.stop_calls.append((work_id, kwargs))


class FakeAsyncWork: