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