Test a video poll loop with unittest and a local server, no spend

A poll loop is the part of a video integration that you test least and that fails in the most expensive way. A clip that costs $0.125 at its smallest is cheap, and a loop that never exits, or one that sleeps for 30 seconds between every check in your test suite, is not. You can test the loop completely without a key, an account or a network by pointing it at a small local server.

This version uses only the Python standard library: unittest, http.server and urllib. If you already use pytest and respx, /blog/pytest-respx-fake-sume-job-queued-then-completed-test-poll-loop-no-network shows that route.

The file

wait is the code under test. It requests the polling URL, returns on a terminal status and otherwise calls a sleep function you pass in. The fake server hands out one status per request from a list. Because sleep is a parameter, the test records the pauses instead of taking them. Run it with python3 test_wait.py; it ran two tests in about one second here.

import json, threading, unittest, urllib.request
from http.server import BaseHTTPRequestHandler, HTTPServer

def wait(url, key, sleep, every=30):
    while True:
        req = urllib.request.Request(url, headers={"Authorization": f"Bearer {key}"})
        job = json.load(urllib.request.urlopen(req))
        if job["status"] in ("completed", "failed", "cancelled"):
            return job
        sleep(every)

def serve(statuses):
    class H(BaseHTTPRequestHandler):
        def log_message(self, *a): pass
        def do_GET(self):
            body = json.dumps({"status": statuses.pop(0)}).encode()
            self.send_response(200); self.send_header("Content-Length", str(len(body)))
            self.end_headers(); self.wfile.write(body)
    srv = HTTPServer(("127.0.0.1", 0), H)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv, f"http://127.0.0.1:{srv.server_port}/v1/videos/job_1"

class WaitTest(unittest.TestCase):
    def check(self, statuses, final, sleeps):
        srv, url = serve(list(statuses)); naps = []
        self.assertEqual(wait(url, "k", naps.append)["status"], final)
        self.assertEqual(naps, sleeps); srv.shutdown()
    def test_completes(self): self.check(["pending", "in_progress", "completed"], "completed", [30, 30])
    def test_failed_stops(self): self.check(["failed"], "failed", [])
unittest.main()

What the tests assert

The first test feeds pending, in_progress, completed, and expects two sleeps of 30 seconds, which is the cadence Sume's guide recommends. The second feeds failed and expects no sleep at all: a terminal state must not wait one more interval before the caller sees it. Both are properties that a hand-run against the live API would only show you by accident.

Cases to add

Extend the status list rather than the server. Add a cancelled case, a job that sits in pending for fifty polls to prove your own timeout trips, and a server that answers 500 once to see what your retry does. Keep a real poll body from your own account as a fixture so the fake matches the live shape, since the live response also carries id, polling_url, model and, when finished, unsigned_urls and usage.cost.

None of this replaces one live run. Do that once with the smallest job you can find, for example the 2 second Wan 3.0 clip at 480p, then let the fake server carry the regression tests.