Unit-test a Sume job poll loop with a fake clock and no network

Inject the status reader and the sleep function to test a Sume poll loop in milliseconds: next_poll_after_seconds, backoff fallback and the client deadline.

4 min readSume
All posts

Make the poll loop a function that takes two callables, get_status and sleep, and your tests can drive it with a scripted list of status envelopes and a recorder instead of a clock. The test then runs in milliseconds, uses no network, and can prove the three behaviors the docs ask for.

Those behaviors come from Jobs and results: stop on terminal, obey next_poll_after_seconds when it is present and back off when it is not, and put the overall deadline in the client, because a client-side timeout does not cancel the job.

What the loop must do

Each line of the contract maps to one assertion in the sample below.

Poll contract and the assertion that covers it (Jobs and results docs, read 2026-10-10)
ContractSource fieldAssertion
Stop when finishedterminal: trueThe loop returns the envelope.
Wait as toldnext_poll_after_secondsThe recorded sleeps equal the scripted hints.
Back off without a hintfield absent or nullSleeps grow 1, 2, 4 and so on, capped at 60.
Own the deadlineclient-sideA TimeoutError is raised; nothing cancels the job.
Fetch only when readyresult_ready: trueThe test reads it from the final envelope.

The loop and its test

The fallback is exponential and capped at 60 seconds. The deadline check runs before the sleep, so a loop with 5 seconds left never starts a 16-second nap. The script ends by showing the backoff that was used when the server gives no hint; it prints 1, 2, 4, 8 and then times out.

def wait_for_job(get_status, sleep, deadline_s=1200):
    waited, n = 0, 0
    while True:
        s = get_status()
        if s["terminal"]:
            return s
        delay = s.get("next_poll_after_seconds") or min(60, 2 ** n)
        n += 1
        if waited + delay > deadline_s:
            raise TimeoutError("client deadline reached; the job keeps running")
        sleep(delay)
        waited += delay

steps = iter([
    {"terminal": False, "sume_status": "queued", "next_poll_after_seconds": 3},
    {"terminal": False, "sume_status": "processing", "next_poll_after_seconds": 6},
    {"terminal": True, "sume_status": "completed", "result_ready": True}])
slept = []
final = wait_for_job(lambda: next(steps), slept.append)
assert slept == [3, 6] and final["result_ready"]

slept = []
try:
    wait_for_job(lambda: {"terminal": False}, slept.append, deadline_s=20)
except TimeoutError as e:
    print("ok:", e, "| backoff used:", slept)

What the test does not prove

It proves your control flow, not the live service. Statuses are queued, processing, completed, failed and canceled, and the envelope also carries a queue-shaped status that maps one to one onto sume_status; the docs say not to mix the two. The fake above uses only sume_status and the booleans.

Keep one slow test against the real API for the shape of the envelope, and keep the fake-clock tests for everything else. If the deadline expires, store the job id and read status_url later, or cancel explicitly; the job keeps running and billing otherwise.

  • Never call time.sleep inside the loop body; inject it.
  • Script a failed and a canceled ending too, and assert you do not fetch the result.
  • Do not resubmit on a timeout; reuse the same Idempotency-Key if you must retry the submit.

Sources

Related posts

More in Developers

All Developers posts

Written by Sume