Test a media API integration with offline fixtures

Test a media API integration in layers: run deterministic fixtures against your application's decision logic, test the HTTP adapter separately, then use a small, explicitly controlled live check when you need evidence about credentials and delivery. A green mock test can prove that your client chose the expected branch. It cannot prove that a provider accepted a job or that a file exists.

This guide focuses on building a useful offline test suite before connecting a worker to live media processing. It uses Python's standard library and makes no network requests.

Separate the boundaries before writing fixtures

A media workflow usually has several boundaries: submitting work, observing a job, retrieving an artifact and handing that artifact to another service. Give each boundary a narrow application interface. For example, lookup(job_id) can return a parsed status document or raise an application-defined transport exception. Your business logic should not need to know how an HTTP library represents a response.

That separation lets a test inject observations without creating media jobs. Keep submission out of a status classifier entirely: a temporary inability to observe a job should not silently submit another one. The worker restart guide explains where durable job identity belongs in a running workflow.

Do not replace the function you want to test with a mock. Import your real classifier, adapter or consumer, and replace only its external dependency. Otherwise the test may merely confirm behavior you wrote into the mock.

Choose fixtures by decisions, not payload size

The Tornado job-status reference documents GET /jobs/{id}, authenticated with x-api-key, and the status values used below. A fixture for the classifier needs only the fields the classifier consumes. Label that fixture as a projection; it is not a complete API response or a schema validation test.

Grouping several outcomes in this table does not make their recovery policies identical. Your application may present a cancellation differently from a technical failure. Preserve the original status in your durable record.

Run a deterministic status sequence in Python

The following complete example tests a small classifier. TimeoutError is the adapter's normalized exception in this example. A real adapter must explicitly translate the timeout exception raised by its chosen HTTP library; built-in TimeoutError does not automatically catch every library's timeout type.

from unittest.mock import Mock, call

JOB = "550e8400-e29b-41d4-a716-446655440000"
ACTIVE = {"Pending", "Processing"}
STOPPED = {"Failed", "Warning", "Skipped", "Cancelled", "CancelledByAdmin"}


def inspect_once(job_id, lookup):
    # lookup is an application adapter, not an HTTP library response.
    try:
        payload = lookup(job_id)
    except TimeoutError:
        return "unresolved"
    if not isinstance(payload, dict) or payload.get("id") != job_id:
        return "review_response"
    status = payload.get("status")
    if not isinstance(status, str):
        return "review_response"
    if status in ACTIVE:
        return "wait"
    if status in STOPPED:
        return "record_outcome"
    if status == "Completed":
        return "inspect_delivery"
    return "review_response"


lookup = Mock(side_effect=[
    {"id": JOB, "status": "Pending"},
    TimeoutError("synthetic timeout"),
    {"id": JOB, "status": "Completed"},
])
assert [inspect_once(JOB, lookup) for _ in range(3)] == [
    "wait", "unresolved", "inspect_delivery"
]
assert lookup.call_args_list == [call(JOB), call(JOB), call(JOB)]

for status in STOPPED:
    assert inspect_once(JOB, Mock(return_value={
        "id": JOB, "status": status
    })) == "record_outcome"

for payload in [None, {}, {"id": "another-job", "status": "Completed"},
                {"id": JOB, "status": "FutureStatus"},
                {"id": JOB, "status": []}]:
    assert inspect_once(JOB, Mock(return_value=payload)) == "review_response"

These twelve assertions were executed locally against the example. No API credential, network, sleep or real media file is involved. Python's mock examples document iterable side_effect values: successive calls consume successive entries, and an exception entry is raised. A fourth lookup in this fixture would exhaust the sequence, so define exactly the observations each test needs.

The classifier deliberately returns inspect_delivery for Completed even without a URL. Delivery validation is a separate boundary. Test that boundary with its own requirements for the selected destination; do not turn an absent URL into proof that a job failed or that the object is inaccessible. See verify storage delivery before processing.

Add adapter and consumer tests separately

The classifier cannot catch a misspelled authentication header or an incorrect endpoint. Test the actual HTTP adapter with an injected transport and assert its request method, path, authentication header placement and timeout configuration. Use a fake credential value. Cover a non-success HTTP response, invalid JSON and the specific timeout class your library raises. Decide which conditions are retryable in the adapter contract, then test that mapping explicitly.

For the consumer boundary, use a spy or mock for the downstream service and assert it is not called after an unresolved lookup or unsuccessful outcome. Test the real orchestration function that connects your decisions to actions. A classifier-only test cannot prove those actions are suppressed.

For persistence, execute the real database transition against an isolated temporary database. Replay the same completion twice and verify that your accepted work count follows your deduplication rule. Restart the worker between acceptance and completion. These are different tests from a response fixture; the duplicate-event guide provides a concrete local outbox example.

Keep fixtures small, explicit and safe to commit

Construct synthetic fixtures rather than saving raw production responses. A response can contain source URLs, signed access URLs and customer metadata. Removing the API key is not enough. Use reserved example domains and fabricated identifiers, and keep any temporary file contents equally synthetic. The diagnostic logging guide explains why complete responses should not become routine diagnostic records.

Record the public contract URL and the date you checked it beside the fixture. When a contract changes, review what the application should do before updating expected results. A mock is under your control, so it can remain green while the service changes. Include unknown fields and unknown status values in robustness tests rather than assuming every future response will match today's happy path.

Know when an offline test has reached its limit

Offline tests do not establish provider availability, delivery permissions, current source accessibility, production latency or billing. A separate live check can examine those boundaries using content you are authorized to process, a controlled account and an explicit cost budget. Keep it out of the default unit-test command so a routine test run cannot unexpectedly submit jobs.

Record whether a result came from a fixture, local integration test or live request. Keep a live failure's job ID for diagnosis, and reconcile its state before resubmitting. Never replace a failing assertion with a fixture copied from a successful run just to make the suite pass.

Ready to connect the tested client? Follow the first media workflow and Python integration tutorial, then retain the offline suite as regression coverage around your own application behavior.