Test a video poll loop with a fake clock, no real waits (Python)

To test a poll loop for long video renders without waiting, pass the loop its sleep and clock functions as arguments, then give the test a fake clock that jumps forward instead of sleeping. A 20-minute client deadline then takes microseconds to prove.

A 30-second render on Seedance 2.5 or Wan 3.0 can run for minutes, so the loop's edge cases are all about time: the deadline, the next_poll_after_seconds hint, and the moment a job turns terminal. Real sleeps make those tests slow and flaky. Injected time makes them exact.

What the loop must get right

Sume's docs recommend polling GET /v1/jobs/:id/status with exponential backoff and stopping on completed, failed, or canceled (Jobs and results).

• Stop on terminal: true, whatever the status is.
• Use next_poll_after_seconds when it is present, and your own backoff when it is not.
• Give up at your own deadline and report a timeout, without canceling the job. A client timeout does not cancel the job, and the job still bills.
• Never submit again from inside the loop.

A poller you can test

The function takes get (returns the status dict), sleep, and clock. The test builds a scripted sequence of statuses and a clock that advances by whatever the loop asked to sleep. Save it as one file and run it with python file.py; it uses only the standard library.

def poll(get, sleep, clock, deadline_s=1200.0):
    end, delay = clock() + deadline_s, 5.0
    while clock() < end:
        s = get()
        if s["terminal"]:
            return s["sume_status"]
        wait = s.get("next_poll_after_seconds") or delay
        delay = min(delay * 1.5, 30.0)
        sleep(wait)
    return "timeout"

class Fake:
    def __init__(self, seq): self.seq, self.t, self.slept = list(seq), 0.0, []
    def get(self): return self.seq.pop(0) if len(self.seq) > 1 else self.seq[0]
    def sleep(self, n): self.slept.append(n); self.t += n
    def clock(self): return self.t

run = lambda f, **k: poll(f.get, f.sleep, f.clock, **k)
q = {"terminal": False, "sume_status": "queued"}
f = Fake([q, q, {"terminal": True, "sume_status": "completed"}])
assert run(f) == "completed" and len(f.slept) == 2
f = Fake([{**q, "next_poll_after_seconds": 7}, {"terminal": True, "sume_status": "failed"}])
assert run(f) == "failed" and f.slept == [7]
assert run(Fake([q]), deadline_s=100) == "timeout"
print("ok")

Cases worth a test

What this does not test

A fake clock proves your loop's logic, not Sume's behavior. Run one real job at the smallest settings, for example a short 480p Wan 3.0 clip, to prove the status and result calls against the live API. At the Sume rate of $0.0625 per second, a 4-second 480p Wan 3.0 clip is a quarter of a dollar.

Keep the injected-time pattern for every other retry loop too. The same sleep and clock arguments work for 429 and queue_full backoff.

Wiring it to the real API

In production, get is a small function that calls GET https://api.sume.com/v1/jobs/{id}/status with your key, and sleep and clock are time.sleep and time.monotonic. Nothing else changes, so the code you tested is the code that runs.

Because the loop only reads, it is safe to restart. A crashed worker can start poll again from the stored job id, and it never has a reason to submit a paid request.