.agent-work/ledger.json

@@ -26,6 +26,11 @@
       "events": [
         "Completed issue 7: stdlib PNG/ICO decode/encode/wrap/unwrap plus resample helpers live in tools/images.py; tools/example.py saves selected images into the caller directory, requires --output-dir with --download (invalid_request otherwise), verifies every stored file by decoding it before reporting its path, downscales to exactly the requested width/height in the requested format only when --resize is set (never upscaling, otherwise byte-identical), and reports a per-item save failure with the partial_download status/exit code. New tests in tests/test_download_resize.py cover all of that; tests/test_cli.py resize expectation aligned with the exact-size requirement. Facade validate/smoke/help/import all pass and `pytest` reports 142 passed on the revision that already contains target task/1-task."
       ]
+    },
+    "projects/product-54354138-1#8": {
+      "events": [
+        "Added tests/test_status_branches.py (subprocess-driven checks for ok, no_format, no_size, access_error x2, invalid_request/unknown-source, partial_download: status + exit code + source/request echo + single JSON document, distinctness, saved-path truthfulness, shipped example run) and tests/test_live_source.py (credential-free live Commons png search with recorded observation and real-dimension/URL re-verification, explicit unresolved skip with observed reason when egress is unusable). Facade validate/test/smoke all succeed; target branch already integrated; no product or protected facade files touched."
+      ]
     }
   }
 }

tests/test_live_source.py

@@ -0,0 +1,251 @@
+"""Live, credential-free end-to-end check against the real Wikimedia Commons.
+
+The case drives the shipped CLI as a subprocess with a scrubbed environment: no
+API key, token, password or other credential variable is passed in, because the
+source is queried anonymously (REQ-003).
+
+When the approved hosts are usable the observed JSON is recorded and every
+reported item is re-fetched independently to prove that its URL works and that
+the reported dimensions are the real pixel size of the PNG bytes.  When egress
+to the approved hosts is unavailable or the host refuses to serve files, the
+case is reported as an explicit *unresolved skip* carrying the observed status
+and error reason.  Synthetic or canned results are never substituted for a live
+observation.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import urllib.parse
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from tools import envelope, images, sources
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+SOURCE_ID = "wikimedia_commons"
+
+#: Topic with known PNG files on Wikimedia Commons (the Nuvola icon set).
+#: Further known topics are tried when the source answers but has no PNG of the
+#: requested size, so a reachable source is not reported as unreachable.
+KNOWN_TOPICS = (
+    "Nuvola icons",
+    "Wikimedia Commons logo",
+    "red fox in the snow",
+)
+MIN_WIDTH = 64
+MIN_HEIGHT = 64
+COUNT = 2
+LIVE_TIMEOUT = 180
+
+#: Environment variable names that could carry a credential of some kind.
+_CREDENTIAL_MARKERS = ("API", "AUTH", "CREDENTIAL", "KEY", "PASSWORD", "SECRET", "TOKEN")
+
+
+def _is_credential_name(name: str) -> bool:
+    """True when *name* looks like it would carry a credential."""
+
+    upper = name.upper()
+    return any(marker in upper for marker in _CREDENTIAL_MARKERS)
+
+
+def scrubbed_environment() -> dict[str, str]:
+    """Return an environment with every credential-looking variable removed."""
+
+    env = {
+        name: value
+        for name, value in os.environ.items()
+        if not _is_credential_name(name)
+    }
+    env["PYTHONPATH"] = str(REPO_ROOT)
+    return env
+
+
+def run_cli(*args: str) -> subprocess.CompletedProcess[str]:
+    """Run ``python -m tools.example`` in a credential-free subprocess."""
+
+    return subprocess.run(
+        [sys.executable, "-m", "tools.example", *args],
+        cwd=REPO_ROOT,
+        env=scrubbed_environment(),
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=LIVE_TIMEOUT,
+    )
+
+
+def single_document(result: subprocess.CompletedProcess[str]) -> dict[str, Any]:
+    """Assert the process wrote exactly one complete JSON envelope."""
+
+    assert result.stderr == "", result.stderr
+    lines = [line for line in result.stdout.splitlines() if line.strip()]
+    assert len(lines) == 1, result.stdout
+    document = json.loads(lines[0])
+    assert isinstance(document, dict)
+    assert set(document) == {"status", "source", "request", "items", "error"}
+    return document
+
+
+def _access_reason(document: dict[str, Any]) -> str:
+    """Describe the observed access failure carried by an access-error answer."""
+
+    error = document.get("error") or {}
+    message = str(error.get("message") or "no error message reported")
+    details = error.get("details") or {}
+    reasons: list[str] = []
+    for failure in details.get("failures") or ():
+        fragment = failure.get("error") or {}
+        text = str(fragment.get("message") or "")
+        if text:
+            reasons.append(text)
+    if reasons:
+        return f"{message}; first candidate failure: {reasons[0]}"
+    return message
+
+
+def _record(record_property: Any, tmp_path: Path, observations: list[dict[str, Any]]) -> str:
+    """Persist the raw observation so it is available as evidence."""
+
+    encoded = json.dumps(observations, sort_keys=True, ensure_ascii=True)
+    record_property("live_source_observation", encoded)
+    artifact = tmp_path / "live-source-observation.json"
+    artifact.write_text(encoded, encoding="utf-8")
+    print("LIVE SOURCE OBSERVATION " + encoded, flush=True)
+    return encoded
+
+
+def test_cli_needs_no_credentials_and_dry_run_answers() -> None:
+    env = scrubbed_environment()
+    assert sorted(name for name in env if _is_credential_name(name)) == []
+
+    result = run_cli(
+        "--topic",
+        KNOWN_TOPICS[0],
+        "--format",
+        "png",
+        "--width",
+        str(MIN_WIDTH),
+        "--height",
+        str(MIN_HEIGHT),
+        "--source",
+        SOURCE_ID,
+        "--dry-run",
+    )
+    document = single_document(result)
+
+    assert result.returncode == envelope.EXIT_OK
+    assert document["status"] == envelope.STATUS_DRY_RUN
+    assert document["source"] == SOURCE_ID
+    assert document["items"] == []
+    assert document["error"] is None
+    assert document["request"]["format"] == "png"
+
+
+def test_live_wikimedia_commons_png_search_without_credentials(
+    tmp_path: Path, record_property: Any
+) -> None:
+    observations: list[dict[str, Any]] = []
+    final: dict[str, Any] | None = None
+
+    for topic in KNOWN_TOPICS:
+        result = run_cli(
+            "--topic",
+            topic,
+            "--format",
+            "png",
+            "--width",
+            str(MIN_WIDTH),
+            "--height",
+            str(MIN_HEIGHT),
+            "--count",
+            str(COUNT),
+            "--source",
+            SOURCE_ID,
+        )
+        document = single_document(result)
+        observations.append(
+            {"topic": topic, "exit_code": result.returncode, "response": document}
+        )
+
+        # The process contract holds for every live answer, matched or not.
+        assert document["source"] == SOURCE_ID
+        request = document["request"]
+        if request is not None:
+            assert request["source"] == SOURCE_ID
+            assert request["format"] == "png"
+        assert result.returncode == envelope.STATUS_EXIT_CODES[document["status"]]
+
+        status = document["status"]
+        final = document
+        if status == envelope.STATUS_ACCESS_ERROR:
+            # The approved hosts are unusable from here: no further topic can
+            # succeed, and the failure must not be disguised as a no-match.
+            break
+        if status == envelope.STATUS_OK:
+            break
+        if status in (envelope.STATUS_NO_FORMAT, envelope.STATUS_NO_SIZE):
+            continue
+        raise AssertionError(f"unexpected live status {status!r}: {json.dumps(document)}")
+
+    assert final is not None
+    _record(record_property, tmp_path, observations)
+    status = final["status"]
+
+    if status == envelope.STATUS_ACCESS_ERROR:
+        pytest.skip(
+            "unresolved live check: the approved Wikimedia host is not usable "
+            f"from this environment ({_access_reason(final)})"
+        )
+
+    if status in (envelope.STATUS_NO_FORMAT, envelope.STATUS_NO_SIZE):
+        pytest.skip(
+            "unresolved live check: the source answered but served no png image "
+            f"of at least {MIN_WIDTH}x{MIN_HEIGHT} px for the known topics "
+            f"({', '.join(KNOWN_TOPICS)}); observed status {status!r}"
+        )
+
+    assert status == envelope.STATUS_OK, json.dumps(final)
+    assert final["error"] is None
+    items = final["items"]
+    assert items, json.dumps(final)
+    assert len(items) <= COUNT
+
+    for item in items:
+        assert item["source"] == SOURCE_ID
+        assert item["format"] == "png"
+        assert isinstance(item["width"], int) and item["width"] >= MIN_WIDTH
+        assert isinstance(item["height"], int) and item["height"] >= MIN_HEIGHT
+        parts = urllib.parse.urlsplit(item["url"])
+        assert parts.scheme == "https"
+        assert parts.hostname in sources.APPROVED_HOSTS
+        assert sources.is_approved_url(item["url"])
+
+    # Prove the reported URLs really serve the reported PNG bytes and that the
+    # dimensions in the JSON are the real pixel size, not source metadata.
+    for item in items:
+        try:
+            response = sources.urllib_transport(
+                item["url"], {"User-Agent": sources.USER_AGENT}, 30.0
+            )
+        except OSError as error:
+            pytest.skip(
+                "unresolved live check: the reported URL became unreachable "
+                f"({item['url']}: {error})"
+            )
+        if response.status != 200:
+            pytest.skip(
+                "unresolved live check: the reported URL stopped working "
+                f"({item['url']} returned HTTP {response.status})"
+            )
+        assert images.detect_format(response.body) == "png", item["url"]
+        assert images.read_png_dimensions(response.body) == (
+            item["width"],
+            item["height"],
+        ), item["url"]

tests/test_status_branches.py

@@ -0,0 +1,450 @@
+"""End-to-end process-level checks for every CLI status branch.
+
+Each case launches a fresh interpreter that runs the real CLI entry point
+(:func:`tools.example.main`) as a subprocess, with a scrubbed environment (no
+API key, token or credential variable) and an injected offline source registry,
+so the assertions cover the whole observable contract of the process: exactly
+one JSON document on stdout, the expected ``status``, the matching exit code and
+the echoed source and request.
+
+The branches exercised here are the ones a caller must be able to tell apart:
+
+* ``ok`` -- matching images really were found;
+* ``no_format`` -- the source had hits but none in the requested format;
+* ``no_size`` -- images of the requested format existed but were too small;
+* ``access_error`` -- the source could not be searched or fetched;
+* ``invalid_request`` -- the request named an unknown source;
+* ``partial_download`` -- some images were saved and some could not be fetched.
+
+Only the *source* is replaced by the injected registry; argument parsing,
+request validation, format/size selection, download handling and envelope
+rendering all run unmodified inside the subprocess.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from tools import envelope
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+EXAMPLE_REQUEST = REPO_ROOT / "examples" / "input.json"
+
+SOURCE_ID = "wikimedia_commons"
+TOPIC = "red fox"
+
+#: Environment variable names that could carry a credential of some kind.
+_CREDENTIAL_MARKERS = ("API", "AUTH", "CREDENTIAL", "KEY", "PASSWORD", "SECRET", "TOKEN")
+
+#: CLI arguments shared by every injected-registry scenario.
+BASE_ARGS = (
+    "--topic",
+    TOPIC,
+    "--format",
+    "png",
+    "--width",
+    "100",
+    "--height",
+    "100",
+    "--source",
+    SOURCE_ID,
+)
+
+#: Process-level driver: replaces the source registry, then runs the real CLI.
+#:
+#: It never re-implements CLI logic.  It only provides an offline stand-in for
+#: the remote source so every status branch can be driven deterministically
+#: through a real operating-system process and its real exit code.
+_DRIVER_SOURCE = '''
+import sys
+
+from tools import example, images
+from tools.errors import SourceUnreachableError
+
+PNG_MATCHING = images.png_bytes(160, 120)
+PNG_TOO_SMALL = images.png_bytes(20, 20)
+JPEG_PAYLOAD = bytes([0xFF, 0xD8, 0xFF, 0xE0]) + bytes(32)
+
+
+class Candidate:
+    """Minimal stand-in for tools.sources.SourceCandidate."""
+
+    source_id = "wikimedia_commons"
+    mime_type = "image/png"
+    page_url = None
+
+    def __init__(self, name, payload=None, fail=False, width=None, height=None):
+        self.title = name
+        self.file_url = "https://upload.wikimedia.org/practice/" + name + ".png"
+        self.width = width
+        self.height = height
+        self._payload = payload
+        self._fail = fail
+
+    def to_dict(self):
+        return {
+            "source_id": self.source_id,
+            "title": self.title,
+            "file_url": self.file_url,
+        }
+
+    def fetch_bytes(self):
+        if self._fail:
+            raise SourceUnreachableError(
+                self.file_url,
+                source_id=self.source_id,
+                reason="simulated transport failure",
+            )
+        return bytes(self._payload)
+
+
+class Registry:
+    """Offline registry exposing the same surface the CLI uses."""
+
+    def __init__(self, candidates=(), error=None):
+        self._candidates = list(candidates)
+        self._error = error
+
+    def available_sources(self):
+        return ["wikimedia_commons"]
+
+    def search(self, source_id, topic, limit=10):
+        if self._error is not None:
+            raise self._error
+        return self._candidates[:limit]
+
+
+SCENARIOS = {
+    "ok": lambda: Registry(
+        [Candidate("good", PNG_MATCHING, width=160, height=120)]
+    ),
+    "no_format": lambda: Registry([Candidate("jpeg", JPEG_PAYLOAD)]),
+    "no_size": lambda: Registry(
+        [Candidate("small", PNG_TOO_SMALL, width=20, height=20)]
+    ),
+    "access_error": lambda: Registry([Candidate("broken", fail=True)]),
+    "search_access_error": lambda: Registry(
+        error=SourceUnreachableError(
+            "https://commons.wikimedia.org/w/api.php",
+            source_id="wikimedia_commons",
+            reason="simulated transport failure",
+        )
+    ),
+    "partial_download": lambda: Registry(
+        [
+            Candidate("good", PNG_MATCHING, width=160, height=120),
+            Candidate("broken", fail=True),
+        ]
+    ),
+}
+
+
+def main():
+    scenario = sys.argv[1]
+    registry = SCENARIOS[scenario]()
+    example.default_registry = lambda: registry
+    raise SystemExit(example.main(sys.argv[2:]))
+
+
+if __name__ == "__main__":
+    main()
+'''
+
+
+def _is_credential_name(name: str) -> bool:
+    """True when *name* looks like it would carry a credential."""
+
+    upper = name.upper()
+    return any(marker in upper for marker in _CREDENTIAL_MARKERS)
+
+
+def scrubbed_environment() -> dict[str, str]:
+    """Return an environment with every credential-looking variable removed."""
+
+    env = {
+        name: value
+        for name, value in os.environ.items()
+        if not _is_credential_name(name)
+    }
+    env["PYTHONPATH"] = str(REPO_ROOT)
+    return env
+
+
+def run_cli(*args: str) -> subprocess.CompletedProcess[str]:
+    """Run ``python -m tools.example`` in a scrubbed subprocess."""
+
+    return subprocess.run(
+        [sys.executable, "-m", "tools.example", *args],
+        cwd=REPO_ROOT,
+        env=scrubbed_environment(),
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=120,
+    )
+
+
+def run_scenario(
+    tmp_path: Path, scenario: str, *args: str
+) -> subprocess.CompletedProcess[str]:
+    """Run the driver subprocess for *scenario* with the real CLI arguments."""
+
+    driver = tmp_path / "status_branch_driver.py"
+    if not driver.exists():
+        driver.write_text(_DRIVER_SOURCE, encoding="utf-8")
+    return subprocess.run(
+        [sys.executable, str(driver), scenario, *args],
+        cwd=REPO_ROOT,
+        env=scrubbed_environment(),
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=120,
+    )
+
+
+def single_document(result: subprocess.CompletedProcess[str]) -> dict[str, Any]:
+    """Assert the process wrote exactly one complete JSON envelope."""
+
+    assert result.stderr == "", result.stderr
+    lines = [line for line in result.stdout.splitlines() if line.strip()]
+    assert len(lines) == 1, result.stdout
+    document = json.loads(lines[0])
+    assert isinstance(document, dict)
+    assert set(document) == {"status", "source", "request", "items", "error"}
+    return document
+
+
+def test_subprocess_environment_carries_no_credentials() -> None:
+    env = scrubbed_environment()
+
+    leaked = sorted(name for name in env if _is_credential_name(name))
+    assert leaked == []
+
+
+@pytest.mark.parametrize(
+    "scenario,status",
+    [
+        ("ok", envelope.STATUS_OK),
+        ("no_format", envelope.STATUS_NO_FORMAT),
+        ("no_size", envelope.STATUS_NO_SIZE),
+        ("access_error", envelope.STATUS_ACCESS_ERROR),
+        ("partial_download", envelope.STATUS_PARTIAL_DOWNLOAD),
+    ],
+)
+def test_status_branch_emits_its_own_status_and_exit_code(
+    tmp_path: Path, scenario: str, status: str
+) -> None:
+    args = BASE_ARGS
+    if scenario == "partial_download":
+        args = BASE_ARGS + (
+            "--count",
+            "2",
+            "--download",
+            "--output-dir",
+            str(tmp_path / "downloads"),
+        )
+
+    result = run_scenario(tmp_path, scenario, *args)
+    document = single_document(result)
+
+    assert document["status"] == status
+    assert result.returncode == envelope.STATUS_EXIT_CODES[status]
+    assert document["source"] == SOURCE_ID
+
+
+def test_status_branch_echoes_the_normalized_request(
+    tmp_path: Path,
+) -> None:
+    """``ok``/``no_format``/``no_size``/``partial_download`` echo the request."""
+
+    for scenario, args in (
+        ("ok", BASE_ARGS),
+        ("no_format", BASE_ARGS),
+        ("no_size", BASE_ARGS),
+        (
+            "partial_download",
+            BASE_ARGS
+            + (
+                "--count",
+                "2",
+                "--download",
+                "--output-dir",
+                str(tmp_path / "downloads-echo"),
+            ),
+        ),
+    ):
+        result = run_scenario(tmp_path, scenario, *args)
+        document = single_document(result)
+
+        assert document["request"] is not None, scenario
+        assert document["request"]["source"] == SOURCE_ID
+        assert document["request"]["topic"] == TOPIC
+        assert document["request"]["format"] == "png"
+        assert document["request"]["width"] == 100
+        assert document["request"]["height"] == 100
+
+
+def test_branches_are_mutually_distinguishable() -> None:
+    branches = [
+        envelope.STATUS_OK,
+        envelope.STATUS_NO_FORMAT,
+        envelope.STATUS_NO_SIZE,
+        envelope.STATUS_ACCESS_ERROR,
+        envelope.STATUS_INVALID_REQUEST,
+        envelope.STATUS_PARTIAL_DOWNLOAD,
+    ]
+
+    codes = {status: envelope.STATUS_EXIT_CODES[status] for status in branches}
+
+    assert len(set(branches)) == len(branches)
+    assert len(set(codes.values())) == len(branches), codes
+
+
+def test_ok_branch_reports_the_verified_png_items(tmp_path: Path) -> None:
+    result = run_scenario(tmp_path, "ok", *BASE_ARGS)
+    document = single_document(result)
+
+    assert document["status"] == envelope.STATUS_OK
+    assert document["error"] is None
+    assert document["items"]
+    item = document["items"][0]
+    assert item["format"] == "png"
+    assert item["width"] == 160
+    assert item["height"] == 120
+    assert item["source"] == SOURCE_ID
+    assert item["url"].startswith("https://")
+
+
+def test_no_format_and_no_size_are_distinct_and_not_access_errors(
+    tmp_path: Path,
+) -> None:
+    format_result = run_scenario(tmp_path, "no_format", *BASE_ARGS)
+    size_result = run_scenario(tmp_path, "no_size", *BASE_ARGS)
+    format_document = single_document(format_result)
+    size_document = single_document(size_result)
+
+    assert format_document["status"] == envelope.STATUS_NO_FORMAT
+    assert size_document["status"] == envelope.STATUS_NO_SIZE
+    assert format_result.returncode == envelope.EXIT_NO_FORMAT
+    assert size_result.returncode == envelope.EXIT_NO_SIZE
+    assert format_document["error"]["code"] == "no_format_match"
+    assert size_document["error"]["code"] == "no_size_match"
+    assert format_document["error"]["message"] != size_document["error"]["message"]
+    for document in (format_document, size_document):
+        assert document["items"] == []
+        assert document["source"] == SOURCE_ID
+        assert document["error"]["code"] != "source_access"
+        assert document["error"]["code"] != "source_unreachable"
+
+
+def test_access_error_branch_is_not_reported_as_a_no_match(tmp_path: Path) -> None:
+    result = run_scenario(tmp_path, "access_error", *BASE_ARGS)
+    document = single_document(result)
+
+    assert document["status"] == envelope.STATUS_ACCESS_ERROR
+    assert result.returncode == envelope.EXIT_ACCESS_ERROR
+    assert document["items"] == []
+    assert document["source"] == SOURCE_ID
+    assert document["request"]["source"] == SOURCE_ID
+    assert document["error"]["code"] == "candidate_unreadable"
+    details = document["error"]["details"]
+    assert details["unreadable_candidates"] == 1
+    assert details["candidates_inspected"] == 0
+    assert details["failures"][0]["error"]["code"] == "source_unreachable"
+    assert details["failures"][0]["source"] == SOURCE_ID
+
+
+def test_search_failure_branch_is_access_error_with_the_source_echoed(
+    tmp_path: Path,
+) -> None:
+    # A source that cannot even be searched is an access error, never a
+    # "nothing matched" answer; the requested source id is still echoed.
+    result = run_scenario(tmp_path, "search_access_error", *BASE_ARGS)
+    document = single_document(result)
+
+    assert document["status"] == envelope.STATUS_ACCESS_ERROR
+    assert result.returncode == envelope.EXIT_ACCESS_ERROR
+    assert document["items"] == []
+    assert document["source"] == SOURCE_ID
+    assert document["error"]["code"] == "source_unreachable"
+    assert document["error"]["details"]["source_id"] == SOURCE_ID
+
+
+def test_partial_download_branch_names_only_saved_files(tmp_path: Path) -> None:
+    output_dir = tmp_path / "downloads"
+
+    result = run_scenario(
+        tmp_path,
+        "partial_download",
+        *BASE_ARGS,
+        "--count",
+        "2",
+        "--download",
+        "--output-dir",
+        str(output_dir),
+    )
+    document = single_document(result)
+
+    assert document["status"] == envelope.STATUS_PARTIAL_DOWNLOAD
+    assert result.returncode == envelope.EXIT_PARTIAL_DOWNLOAD
+    details = document["error"]["details"]
+    assert details["failed"] == 1
+    assert details["saved"] == 1
+    assert details["total"] == 2
+    assert len(details["saved_paths"]) == 1
+    saved_path = Path(details["saved_paths"][0])
+    assert saved_path.is_file()
+    assert document["items"][0]["download"]["ok"] is True
+    assert document["source"] == SOURCE_ID
+
+
+def test_unknown_source_branch_is_invalid_request_with_the_source_echoed() -> None:
+    result = run_cli(
+        "--topic",
+        TOPIC,
+        "--format",
+        "png",
+        "--width",
+        "100",
+        "--height",
+        "100",
+        "--source",
+        "no-such-source",
+        "--dry-run",
+    )
+    document = single_document(result)
+
+    assert document["status"] == envelope.STATUS_INVALID_REQUEST
+    assert result.returncode == envelope.EXIT_INVALID_REQUEST
+    assert document["source"] == "no-such-source"
+    assert document["error"]["code"] == "source_unknown"
+    assert SOURCE_ID in document["error"]["details"]["available_sources"]
+
+
+def test_shipped_example_request_runs_and_parses() -> None:
+    expected = json.loads(EXAMPLE_REQUEST.read_text(encoding="utf-8"))
+
+    result = run_cli("examples/input.json")
+    document = single_document(result)
+
+    # The shipped example is a dry run, so it validates and echoes the request
+    # without contacting any source.
+    assert result.returncode == envelope.EXIT_OK
+    assert document["status"] == envelope.STATUS_DRY_RUN
+    assert document["source"] == expected["source"]
+    assert document["items"] == []
+    assert document["error"] is None
+    assert document["request"]["topic"] == expected["topic"]
+    assert document["request"]["format"] == expected["format"]
+    assert document["request"]["width"] == expected["width"]
+    assert document["request"]["height"] == expected["height"]
+    assert document["request"]["count"] == expected["count"]
+    assert document["request"]["dry_run"] is True