From d506c7a65235a20033f7b4c54b5a183378209ccc Mon Sep 17 00:00:00 2001 From: Tool Backend Date: Tue, 3 Nov 2026 02:34:16 +0000 Subject: [PATCH] change: Implement 6: Verify every CLI status branch end to end and run the live credential-free source check MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Added executable, process-level verification for every CLI status branch plus a live credential-free source check. - `tests/test_status_branches.py` (new): each case starts a fresh interpreter that runs the real `tools.example.main` entry point as a subprocess with a scrubbed environment (no API/KEY/TOKEN/SECRET/PASSWORD/CREDENTIAL/AUTH variables) and an injected offline source registry. It asserts the whole contract — exactly one JSON document with the fixed five keys, the expected `status`, the matching exit code, and the echoed source/request — for `ok`, `no_format`, `no_size`, `access_error` (both candidate-fetch and search failures), `invalid_request`/unknown source and `partial_download`. It also proves the branches have mutually distinct exit codes, that `no_format`/`no_size` are distinct explanations and never access errors, that partial download names only files that really exist, and that the shipped `examples/input.json` request runs and parses. - `tests/test_live_source.py` (new): searches a known Wikimedia Commons topic for png at ≥64x64 with no credential variables present, asserting the process contract on every live answer; on success it records the observed JSON (record_property + artifact file) and independently re-fetches each reported URL to prove it serves real PNG bytes at the reported dimensions. When the approved hosts are unusable the case is reported as an explicit unresolved skip carrying the observed status/reason — synthetic results are never substituted. In this environment the run observed `access_error` because `upload.wikimedia.org` returned HTTP 429 for every candidate, and the test skipped with that exact reason; an always-passing offline credential-free dry-run case keeps the live file green. - No product code, facade paths or dependency declarations were changed; the target branch `task/1-task` is already contained in the work branch. Tests: Ran through the facade entrypoints with the prepared `.venv`: `tools.validate` (`python -m compileall tools tests`) → OK; `tools.test` (`pytest`) → 158 passed, 1 skipped (the live case skipped as an explicit unresolved egress failure: upload.wikimedia.org HTTP 429, reason recorded); `tools.smoke` (`python -m tools.example examples/input.json`) → exit 0 with a parsed dry_run envelope. Also ran the two new files directly: 16 passed, 1 skipped. Progress: 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. Published by developer-agent through Tool Backend. Closes #8 --- .agent-work/ledger.json | 5 + tests/test_live_source.py | 251 +++++++++++++++++++ tests/test_status_branches.py | 450 ++++++++++++++++++++++++++++++++++ 3 files changed, 706 insertions(+) create mode 100644 tests/test_live_source.py create mode 100644 tests/test_status_branches.py diff --git a/.agent-work/ledger.json b/.agent-work/ledger.json index 4ee58a1..123c0f3 100644 --- a/.agent-work/ledger.json +++ b/.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." + ] } } } diff --git a/tests/test_live_source.py b/tests/test_live_source.py new file mode 100644 index 0000000..790a6f0 --- /dev/null +++ b/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"] diff --git a/tests/test_status_branches.py b/tests/test_status_branches.py new file mode 100644 index 0000000..99b7970 --- /dev/null +++ b/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 -- 2.54.0