diff --git a/.agent-work/ledger.json b/.agent-work/ledger.json index 8240ce2..774c236 100644 --- a/.agent-work/ledger.json +++ b/.agent-work/ledger.json @@ -9,6 +9,12 @@ "events": [ "Implemented tools/errors.py, tools/sources.py and tests/test_sources.py for the credential-free source registry and HTTPS fetch layer. Verified with the facade commands tools.validate, tools.test (54 passed) and tools.smoke, all green on the integrated branch (target task/1-task was already at HEAD). No protected facade paths touched." ] + }, + "projects/product-54354138-1#5": { + "events": [ + "Implemented envelope status/exit-code table and response builder, request errors, the full flag/JSON-document CLI in tools/example.py, canonical offline examples/input.json, and CLI tests; facade validate/test/smoke and --help all pass (76 tests).", + "Addressed the changes-requested review on PR #13: requested width/height are now a minimum target size (selection no longer depends on --resize), and --resize now actually downscales downloaded images (stdlib PNG codec + ICO/PNG container support, shrink-to-fit, never upscaling) with per-item resize metadata and typed ImageResizeError for undecodable payloads. Updated tests/test_cli.py with minimum-size and offline resize/download contract tests (83 passed). Facade validate/test/smoke/help all green; examples/input.json stays the canonical dry_run request." + ] } } } diff --git a/examples/input.json b/examples/input.json index 6ed9590..b797f90 100644 --- a/examples/input.json +++ b/examples/input.json @@ -1,9 +1,13 @@ { "version": 1, - "mode": "validate", - "dry_run": true, - "command": "example", - "arguments": { - "request": "examples/input.json" - } + "topic": "red fox in the snow", + "width": 640, + "height": 480, + "format": "png", + "source": "wikimedia_commons", + "count": 5, + "download": false, + "output_dir": "downloads", + "resize": false, + "dry_run": true } diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..bd7f362 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,550 @@ +"""Contract tests for the native-process CLI in :mod:`tools.example`.""" + +from __future__ import annotations + +import json +import struct +import subprocess +import sys +import zlib +from pathlib import Path +from typing import Any, Sequence + +import pytest + +from tools import envelope, example +from tools.errors import SourceAccessError + +REPO_ROOT = Path(__file__).resolve().parents[1] +EXAMPLE_REQUEST = REPO_ROOT / "examples" / "input.json" + + +def run_cli(*args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, "-m", "tools.example", *args], + cwd=REPO_ROOT, + capture_output=True, + text=True, + check=False, + ) + + +def single_document(result: subprocess.CompletedProcess[str]) -> dict[str, Any]: + """Assert stdout holds exactly one machine-readable JSON document.""" + + lines = [line for line in result.stdout.splitlines() if line.strip()] + assert len(lines) == 1, result.stdout + assert result.stderr == "" + document = json.loads(lines[0]) + assert isinstance(document, dict) + assert set(document) == {"status", "source", "request", "items", "error"} + return document + + +def test_status_vocabulary_and_exit_codes_are_fixed() -> None: + assert envelope.STATUS_EXIT_CODES == { + "ok": 0, + "dry_run": 0, + "invalid_request": 2, + "no_format": 3, + "no_size": 4, + "access_error": 5, + "partial_download": 6, + } + assert envelope.STATUS_VOCABULARY == ( + "ok", + "dry_run", + "invalid_request", + "no_format", + "no_size", + "access_error", + "partial_download", + ) + + +def test_help_documents_every_parameter() -> None: + result = run_cli("--help") + + assert result.returncode == 0 + for fragment in ( + "REQUEST.json", + "--topic", + "--width", + "--height", + "--format", + "--source", + "--count", + "--download", + "--output-dir", + "--resize", + "--dry-run", + "png|ico", + "JSON", + ): + assert fragment in result.stdout, fragment + + +def test_example_request_document_is_canonical_dry_run() -> None: + document = json.loads(EXAMPLE_REQUEST.read_text(encoding="utf-8")) + assert document["dry_run"] is True + assert document["format"] in ("png", "ico") + + result = run_cli("examples/input.json") + + assert result.returncode == envelope.EXIT_OK + payload = single_document(result) + assert payload["status"] == envelope.STATUS_DRY_RUN + assert payload["source"] == document["source"] + assert payload["items"] == [] + assert payload["error"] is None + assert payload["request"]["topic"] == document["topic"] + assert payload["request"]["format"] == document["format"] + + +def test_flags_dry_run_emits_one_document() -> None: + result = run_cli("--topic", "fox", "--format", "png", "--width", "640", "--dry-run") + + assert result.returncode == envelope.EXIT_OK + payload = single_document(result) + assert payload["status"] == envelope.STATUS_DRY_RUN + assert payload["request"]["width"] == 640 + assert payload["request"]["dry_run"] is True + + +def test_unsupported_format_is_invalid_request() -> None: + result = run_cli("--topic", "fox", "--format", "gif") + + assert result.returncode == envelope.EXIT_INVALID_REQUEST + payload = single_document(result) + assert payload["status"] == envelope.STATUS_INVALID_REQUEST + assert payload["error"]["code"] == "request_format_unsupported" + assert payload["error"]["details"]["accepted_formats"] == ["png", "ico"] + + +def test_unknown_source_lists_available_ids() -> None: + result = run_cli("--topic", "fox", "--format", "png", "--source", "nope") + + assert result.returncode == envelope.EXIT_INVALID_REQUEST + payload = single_document(result) + assert payload["status"] == envelope.STATUS_INVALID_REQUEST + assert payload["error"]["code"] == "source_unknown" + available = payload["error"]["details"]["available_sources"] + assert "nope" not in available + assert "wikimedia_commons" in available + assert len(available) >= 2 + + +def test_no_request_is_invalid_request() -> None: + result = run_cli() + + assert result.returncode == envelope.EXIT_INVALID_REQUEST + payload = single_document(result) + assert payload["status"] == envelope.STATUS_INVALID_REQUEST + + +def test_document_combined_with_flags_is_invalid_request() -> None: + result = run_cli("examples/input.json", "--topic", "fox") + + assert result.returncode == envelope.EXIT_INVALID_REQUEST + payload = single_document(result) + assert payload["error"]["details"]["conflicting_flags"] == ["topic"] + + +@pytest.mark.parametrize( + "arguments", + [ + ("--format", "png"), + ("--topic", "fox"), + ("--topic", "fox", "--format", "png", "--count", "0"), + ("--topic", " ", "--format", "png"), + ], +) +def test_incomplete_requests_are_invalid(arguments: Sequence[str]) -> None: + result = run_cli(*arguments) + + assert result.returncode == envelope.EXIT_INVALID_REQUEST + assert single_document(result)["status"] == envelope.STATUS_INVALID_REQUEST + + +def test_missing_document_file_is_invalid_request(tmp_path: Path) -> None: + result = run_cli(str(tmp_path / "absent.json")) + + assert result.returncode == envelope.EXIT_INVALID_REQUEST + assert single_document(result)["status"] == envelope.STATUS_INVALID_REQUEST + + +def test_malformed_document_is_invalid_request(tmp_path: Path) -> None: + broken = tmp_path / "broken.json" + broken.write_text("{not json", encoding="utf-8") + + result = run_cli(str(broken)) + + assert result.returncode == envelope.EXIT_INVALID_REQUEST + assert single_document(result)["status"] == envelope.STATUS_INVALID_REQUEST + + +class FakeCandidate: + """Minimal stand-in for :class:`tools.sources.SourceCandidate`.""" + + def __init__( + self, + title: str, + file_url: str, + mime_type: str, + width: int | None, + height: int | None, + *, + payload: bytes = b"payload", + fail: bool = False, + ) -> None: + self.source_id = "fake" + self.title = title + self.file_url = file_url + self.page_url = file_url + ".html" + self.mime_type = mime_type + self.width = width + self.height = height + self.license = "CC0" + self.creator = "tester" + self._payload = payload + self._fail = fail + + def to_dict(self) -> dict[str, Any]: + return { + "source_id": self.source_id, + "title": self.title, + "file_url": self.file_url, + "page_url": self.page_url, + "mime_type": self.mime_type, + "width": self.width, + "height": self.height, + } + + def fetch_bytes(self) -> bytes: + if self._fail: + raise SourceAccessError("download refused", url=self.file_url) + return self._payload + + +class FakeRegistry: + def __init__(self, candidates: Sequence[FakeCandidate] = (), error: Exception | None = None) -> None: + self._candidates = list(candidates) + self._error = error + self.calls: list[tuple[str, str, int]] = [] + + def available_sources(self) -> Sequence[str]: + return ["fake"] + + def search(self, source_id: str, topic: str, limit: int = 10) -> list[FakeCandidate]: + self.calls.append((source_id, topic, limit)) + if self._error is not None: + raise self._error + return self._candidates[:limit] + + + + +def run_main(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], registry: FakeRegistry, *args: str) -> tuple[int, dict[str, Any]]: + monkeypatch.setattr(example, "default_registry", lambda: registry) + code = example.main(list(args)) + captured = capsys.readouterr() + assert captured.err == "" + lines = [line for line in captured.out.splitlines() if line.strip()] + assert len(lines) == 1, captured.out + return code, json.loads(lines[0]) + + +BASE_ARGS = ("--topic", "fox", "--format", "png", "--width", "100", "--height", "100", "--source", "fake") + + +def test_ok_lists_items_without_download(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + registry = FakeRegistry([FakeCandidate("fox one", "https://example.test/a.png", "image/png", 100, 100)]) + + code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS) + + assert code == envelope.EXIT_OK + assert payload["status"] == envelope.STATUS_OK + assert payload["source"] == "fake" + assert [item["file_url"] for item in payload["items"]] == ["https://example.test/a.png"] + assert payload["items"][0]["width"] == 100 + assert registry.calls == [("fake", "fox", 10)] + + +def test_no_format_when_source_has_no_matching_mime(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.jpg", "image/jpeg", 100, 100)]) + + code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS) + + assert code == envelope.EXIT_NO_FORMAT + assert payload["status"] == envelope.STATUS_NO_FORMAT + + +def test_no_size_when_only_smaller_images_exist(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 50, 50)]) + + code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS) + + assert code == envelope.EXIT_NO_SIZE + assert payload["status"] == envelope.STATUS_NO_SIZE + + +def test_requested_size_is_a_minimum_not_an_exact_match( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 400, 300)]) + + code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS) + + assert code == envelope.EXIT_OK + assert payload["status"] == envelope.STATUS_OK + assert payload["request"]["resize"] is False + assert [item["width"] for item in payload["items"]] == [400] + assert [item["height"] for item in payload["items"]] == [300] + + +def test_exact_size_is_accepted(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 100, 100)]) + + code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS) + + assert code == envelope.EXIT_OK + assert len(payload["items"]) == 1 + + +def test_resize_without_download_keeps_reported_source_size( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 400, 300)]) + + code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS, "--resize") + + assert code == envelope.EXIT_OK + assert payload["request"]["resize"] is True + assert payload["items"][0]["width"] == 400 + assert "download" not in payload["items"][0] + + +def test_download_success_writes_files(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: + registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 100, 100)]) + + code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS, "--download", "--output-dir", str(tmp_path)) + + assert code == envelope.EXIT_OK + download = payload["items"][0]["download"] + assert download["ok"] is True + target = Path(download["path"]) + assert target.parent == tmp_path + assert target.read_bytes() == b"payload" + + +def test_partial_download_reports_status(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: + registry = FakeRegistry( + [ + FakeCandidate("fox", "https://example.test/a.png", "image/png", 100, 100), + FakeCandidate("fox2", "https://example.test/b.png", "image/png", 100, 100, fail=True), + ] + ) + + code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS, "--count", "2", "--download", "--output-dir", str(tmp_path)) + + assert code == envelope.EXIT_PARTIAL_DOWNLOAD + assert payload["status"] == envelope.STATUS_PARTIAL_DOWNLOAD + + +def test_all_downloads_failing_is_access_error(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: + registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 100, 100, fail=True)]) + + code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS, "--download", "--output-dir", str(tmp_path)) + + assert code == envelope.EXIT_ACCESS_ERROR + assert payload["status"] == envelope.STATUS_ACCESS_ERROR + + +def test_search_access_error_maps_to_access_error(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + registry = FakeRegistry(error=SourceAccessError("host unreachable", url="https://example.test")) + + code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS) + + assert code == envelope.EXIT_ACCESS_ERROR + assert payload["status"] == envelope.STATUS_ACCESS_ERROR + assert payload["error"]["code"] == "source_access" + + +# --------------------------------------------------------------------------- +# Size filtering and real downscaling of downloaded images +# --------------------------------------------------------------------------- + +PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n" + + +def png_bytes(width: int, height: int, rgba: tuple[int, int, int, int] = (255, 0, 0, 255)) -> bytes: + """Build a minimal, valid RGBA PNG without any third-party dependency.""" + + raw = bytearray() + for _ in range(height): + raw.append(0) + raw += bytes(rgba) * width + + def chunk(kind: bytes, data: bytes) -> bytes: + crc = zlib.crc32(kind + data) & 0xFFFFFFFF + return struct.pack(">I", len(data)) + kind + data + struct.pack(">I", crc) + + header = struct.pack(">IIBBBBB", width, height, 8, 6, 0, 0, 0) + return ( + PNG_SIGNATURE + + chunk(b"IHDR", header) + + chunk(b"IDAT", zlib.compress(bytes(raw))) + + chunk(b"IEND", b"") + ) + + +def png_size(payload: bytes) -> tuple[int, int]: + assert payload.startswith(PNG_SIGNATURE) + return struct.unpack(">II", payload[16:24]) + + +def ico_bytes(inner_png: bytes, width: int, height: int) -> bytes: + header = struct.pack(" tuple[int, int, bytes]: + reserved, kind, count = struct.unpack(" None: + payload = png_bytes(200, 100) + registry = FakeRegistry( + [FakeCandidate("fox", "https://example.test/a.png", "image/png", 200, 100, payload=payload)] + ) + + code, document = run_main( + monkeypatch, capsys, registry, *BASE_ARGS, "--resize", "--download", "--output-dir", str(tmp_path) + ) + + assert code == envelope.EXIT_OK + download = document["items"][0]["download"] + assert download["ok"] is True + assert download["resize"]["applied"] is True + assert download["resize"]["from"] == {"width": 200, "height": 100} + assert download["resize"]["to"] == {"width": 100, "height": 50} + stored = Path(download["path"]).read_bytes() + assert png_size(stored) == (100, 50) + assert download["width"] == 100 + assert download["height"] == 50 + + +def test_download_without_resize_keeps_original_payload( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + payload = png_bytes(200, 100) + registry = FakeRegistry( + [FakeCandidate("fox", "https://example.test/a.png", "image/png", 200, 100, payload=payload)] + ) + + code, document = run_main( + monkeypatch, capsys, registry, *BASE_ARGS, "--download", "--output-dir", str(tmp_path) + ) + + assert code == envelope.EXIT_OK + download = document["items"][0]["download"] + assert "resize" not in download + assert Path(download["path"]).read_bytes() == payload + + +def test_resize_never_scales_up( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + payload = png_bytes(50, 50) + registry = FakeRegistry( + [FakeCandidate("fox", "https://example.test/a.png", "image/png", 50, 50, payload=payload)] + ) + + code, document = run_main( + monkeypatch, capsys, registry, *DOWNLOAD_ARGS, "--resize", "--download", "--output-dir", str(tmp_path) + ) + + assert code == envelope.EXIT_OK + download = document["items"][0]["download"] + assert download["resize"]["applied"] is False + assert download["resize"]["reason"] == "already_fits" + assert Path(download["path"]).read_bytes() == payload + + +def test_resize_downscales_png_stored_inside_ico( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + payload = ico_bytes(png_bytes(100, 100), 100, 100) + registry = FakeRegistry( + [FakeCandidate("fox", "https://example.test/a.ico", "image/x-icon", 100, 100, payload=payload)] + ) + + code, document = run_main( + monkeypatch, + capsys, + registry, + "--topic", + "fox", + "--format", + "ico", + "--source", + "fake", + "--width", + "50", + "--height", + "50", + "--resize", + "--download", + "--output-dir", + str(tmp_path), + ) + + assert code == envelope.EXIT_OK + download = document["items"][0]["download"] + assert download["resize"]["to"] == {"width": 50, "height": 50} + width, height, inner = ico_payload(Path(download["path"]).read_bytes()) + assert (width, height) == (50, 50) + assert png_size(inner) == (50, 50) + + +def test_undecodable_payload_with_resize_fails_the_item( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + registry = FakeRegistry( + [ + FakeCandidate( + "fox", "https://example.test/a.png", "image/png", 100, 100, payload=b"not-an-image" + ) + ] + ) + + code, document = run_main( + monkeypatch, capsys, registry, *BASE_ARGS, "--resize", "--download", "--output-dir", str(tmp_path) + ) + + assert code == envelope.EXIT_ACCESS_ERROR + assert document["status"] == envelope.STATUS_ACCESS_ERROR + download = document["items"][0]["download"] + assert download["ok"] is False + assert download["error"]["code"] == "resize_failed" diff --git a/tests/test_example.py b/tests/test_example.py index 1574b48..46cd291 100644 --- a/tests/test_example.py +++ b/tests/test_example.py @@ -1,4 +1,4 @@ -"""Tests for the placeholder toolkit CLI and its request document.""" +"""End-to-end tests for the example image-search CLI and its request document.""" from __future__ import annotations @@ -9,7 +9,7 @@ from pathlib import Path import pytest -from tools import example +from tools import envelope, example REPO_ROOT = Path(__file__).resolve().parents[1] EXAMPLE_REQUEST = REPO_ROOT / "examples" / "input.json" @@ -26,43 +26,58 @@ def run_cli(*args: str) -> subprocess.CompletedProcess[str]: ) -def test_example_request_document_is_valid_json_object() -> None: +def single_document(stdout: str) -> dict: + lines = [line for line in stdout.splitlines() if line.strip()] + assert len(lines) == 1, stdout + document = json.loads(lines[0]) + assert isinstance(document, dict) + return document + + +def test_example_request_document_is_canonical_dry_run_request() -> None: document = json.loads(EXAMPLE_REQUEST.read_text(encoding="utf-8")) assert isinstance(document, dict) - assert set(example.REQUIRED_FIELDS) <= set(document) + assert set(example.REQUEST_FIELDS) <= set(document) assert document["dry_run"] is True - assert document["mode"] == "validate" + assert document["format"] in example.ACCEPTED_FORMATS + assert isinstance(document["topic"], str) and document["topic"].strip() + assert document["source"] in example.default_registry().available_sources() -def test_example_request_document_passes_validation() -> None: - request = example.validate_request(example.load_request(EXAMPLE_REQUEST)) +def test_example_request_document_passes_normalization() -> None: + document = json.loads(EXAMPLE_REQUEST.read_text(encoding="utf-8")) + + request = example.normalize_request(document) - assert request["mode"] == "validate" assert request["dry_run"] is True - assert request["arguments"]["request"] == "examples/input.json" + assert request["format"] == "png" + assert request["source"] == "wikimedia_commons" + assert request["count"] == document["count"] def test_cli_prints_single_json_envelope_and_exits_zero() -> None: result = run_cli("examples/input.json") - assert result.returncode == 0, result.stderr + assert result.returncode == envelope.EXIT_OK, result.stderr assert result.stderr == "" - lines = [line for line in result.stdout.splitlines() if line.strip()] - assert len(lines) == 1 - - envelope = json.loads(lines[0]) - assert envelope == { - "status": "ok", - "command": "example", - "mode": "validate", - "dry_run": True, - "version": 1, - "request": "examples/input.json", - } + response = single_document(result.stdout) + assert response["status"] == envelope.STATUS_DRY_RUN + assert response["source"] == "wikimedia_commons" + assert response["items"] == [] + assert response["request"]["dry_run"] is True + assert response["error"] is None -def test_cli_help_exits_zero() -> None: +def test_cli_envelope_is_deterministic() -> None: + first = run_cli("examples/input.json") + second = run_cli("examples/input.json") + + assert first.returncode == second.returncode == envelope.EXIT_OK + assert first.stdout == second.stdout + + +def test_cli_help_exits_zero_and_documents_parameters() -> None: result = run_cli("--help") assert result.returncode == 0 @@ -74,10 +89,10 @@ def test_cli_reports_missing_request_document(tmp_path: Path) -> None: result = run_cli(str(missing)) - assert result.returncode == example.EXIT_INVALID_REQUEST - envelope = json.loads(result.stdout) - assert envelope["status"] == "error" - assert envelope["error"]["code"] == "request_not_found" + assert result.returncode == envelope.EXIT_INVALID_REQUEST + response = single_document(result.stdout) + assert response["status"] == envelope.STATUS_INVALID_REQUEST + assert response["error"]["code"] == "request_document" def test_cli_reports_malformed_json(tmp_path: Path) -> None: @@ -86,33 +101,32 @@ def test_cli_reports_malformed_json(tmp_path: Path) -> None: result = run_cli(str(malformed)) - assert result.returncode == example.EXIT_INVALID_REQUEST - assert json.loads(result.stdout)["error"]["code"] == "request_not_json" + assert result.returncode == envelope.EXIT_INVALID_REQUEST + response = single_document(result.stdout) + assert response["status"] == envelope.STATUS_INVALID_REQUEST + assert response["error"]["code"] == "request_document" + + +def test_cli_reports_non_object_document(tmp_path: Path) -> None: + payload = tmp_path / "list.json" + payload.write_text("[]", encoding="utf-8") + + result = run_cli(str(payload)) + + assert result.returncode == envelope.EXIT_INVALID_REQUEST + assert single_document(result.stdout)["status"] == envelope.STATUS_INVALID_REQUEST @pytest.mark.parametrize( - ("document", "code"), + "document", [ - ([], "request_not_object"), - ({"mode": "validate", "dry_run": True, "command": "example", "arguments": {}}, "request_field_missing"), - ({"version": 1, "mode": "validate", "dry_run": False, "command": "example", "arguments": {"request": "x"}}, "request_not_dry_run"), - ({"version": 1, "mode": "execute", "dry_run": True, "command": "example", "arguments": {"request": "x"}}, "request_mode_unsupported"), - ({"version": 1, "mode": "validate", "dry_run": True, "command": "example", "arguments": {}}, "request_field_missing"), - ({"version": 1, "mode": "validate", "dry_run": True, "command": "", "arguments": {"request": "x"}}, "request_field_invalid"), + {}, + {"topic": "red fox"}, + {"topic": "", "format": "png"}, + {"topic": "red fox", "format": "gif"}, + {"topic": "red fox", "format": "png", "count": 0}, ], ) -def test_validate_request_rejects_invalid_documents(document: object, code: str) -> None: - with pytest.raises(example.RequestError) as excinfo: - example.validate_request(document) - - assert excinfo.value.code == code - - -def test_envelope_is_deterministic() -> None: - request = example.validate_request(example.load_request(EXAMPLE_REQUEST)) - - first = example.build_envelope(request, "examples/input.json") - second = example.build_envelope(request, "examples/input.json") - - assert first == second - assert json.dumps(first, sort_keys=True) == json.dumps(second, sort_keys=True) +def test_normalize_request_rejects_invalid_documents(document: object) -> None: + with pytest.raises(example.RequestError): + example.normalize_request(document) # type: ignore[arg-type] diff --git a/tools/envelope.py b/tools/envelope.py new file mode 100644 index 0000000..8c351f8 --- /dev/null +++ b/tools/envelope.py @@ -0,0 +1,263 @@ +"""Status vocabulary, exit codes and the JSON response envelope. + +Every invocation of ``python -m tools.example`` writes exactly one JSON +document to stdout. The document always carries a ``status`` value from the +fixed vocabulary below and the process exit code is derived from that value, so +a calling AI agent can tell a successful search apart from "nothing matched", +a broken request and an unusable source without parsing prose. +""" + +from __future__ import annotations + +import json +import sys +from dataclasses import dataclass +from typing import Any, Iterable, Mapping, TextIO + +from .errors import ( + RequestError, + SearchRequestError, + SourceAccessError, + ToolkitError, + UnknownSourceError, +) + +__all__ = [ + "EXIT_ACCESS_ERROR", + "EXIT_INVALID_REQUEST", + "EXIT_NO_FORMAT", + "EXIT_NO_SIZE", + "EXIT_OK", + "EXIT_PARTIAL_DOWNLOAD", + "Response", + "ResponseBuilder", + "STATUS_ACCESS_ERROR", + "STATUS_DRY_RUN", + "STATUS_EXIT_CODES", + "STATUS_INVALID_REQUEST", + "STATUS_NO_FORMAT", + "STATUS_NO_SIZE", + "STATUS_OK", + "STATUS_PARTIAL_DOWNLOAD", + "STATUS_VOCABULARY", + "error_fragment", + "exit_code_for", + "is_success", + "status_for_error", +] + +#: Successful search/download. +STATUS_OK = "ok" +#: The request was validated and echoed without contacting a source. +STATUS_DRY_RUN = "dry_run" +#: The request itself is unusable (bad flags, bad document, unknown source or format). +STATUS_INVALID_REQUEST = "invalid_request" +#: The source answered but offers nothing in the requested format. +STATUS_NO_FORMAT = "no_format" +#: The source answered but offers nothing in the requested size. +STATUS_NO_SIZE = "no_size" +#: The source could not be reached, refused to answer or returned unusable data. +STATUS_ACCESS_ERROR = "access_error" +#: Some downloads succeeded and some failed. +STATUS_PARTIAL_DOWNLOAD = "partial_download" + +#: The single source of truth for the status -> exit code mapping. +STATUS_EXIT_CODES: dict[str, int] = { + STATUS_OK: 0, + STATUS_DRY_RUN: 0, + STATUS_INVALID_REQUEST: 2, + STATUS_NO_FORMAT: 3, + STATUS_NO_SIZE: 4, + STATUS_ACCESS_ERROR: 5, + STATUS_PARTIAL_DOWNLOAD: 6, +} + +#: Stable ordering used in help text and documentation. +STATUS_VOCABULARY: tuple[str, ...] = tuple(STATUS_EXIT_CODES) + +EXIT_OK = STATUS_EXIT_CODES[STATUS_OK] +EXIT_INVALID_REQUEST = STATUS_EXIT_CODES[STATUS_INVALID_REQUEST] +EXIT_NO_FORMAT = STATUS_EXIT_CODES[STATUS_NO_FORMAT] +EXIT_NO_SIZE = STATUS_EXIT_CODES[STATUS_NO_SIZE] +EXIT_ACCESS_ERROR = STATUS_EXIT_CODES[STATUS_ACCESS_ERROR] +EXIT_PARTIAL_DOWNLOAD = STATUS_EXIT_CODES[STATUS_PARTIAL_DOWNLOAD] + + +def exit_code_for(status: str) -> int: + """Return the process exit code for ``status``. + + Raises :class:`ValueError` for anything outside the fixed vocabulary, so a + typo can never silently produce an unknown exit code. + """ + + try: + return STATUS_EXIT_CODES[status] + except KeyError: + raise ValueError( + f"unknown status {status!r}; expected one of: {', '.join(STATUS_VOCABULARY)}" + ) from None + + +def is_success(status: str) -> bool: + """True when ``status`` maps onto a zero exit code.""" + + return exit_code_for(status) == 0 + + +def error_fragment( + code: str, + message: str, + *, + details: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """Build the machine-readable ``error`` fragment of a response.""" + + payload: dict[str, Any] = {"code": str(code), "message": str(message)} + if details: + payload["details"] = dict(details) + return payload + + +def status_for_error(error: BaseException) -> str: + """Map a toolkit error onto the fixed status vocabulary.""" + + if isinstance(error, (RequestError, UnknownSourceError, SearchRequestError)): + return STATUS_INVALID_REQUEST + if isinstance(error, SourceAccessError): + return STATUS_ACCESS_ERROR + if isinstance(error, ToolkitError): + return STATUS_ACCESS_ERROR + return STATUS_ACCESS_ERROR + + +@dataclass(frozen=True) +class Response: + """One complete machine-readable answer of the toolkit.""" + + status: str + source: str | None = None + request: Mapping[str, Any] | None = None + items: tuple[Mapping[str, Any], ...] = () + error: Mapping[str, Any] | None = None + + def __post_init__(self) -> None: + if self.status not in STATUS_EXIT_CODES: + raise ValueError( + f"unknown status {self.status!r}; expected one of: " + f"{', '.join(STATUS_VOCABULARY)}" + ) + if self.error is not None and "code" not in self.error: + raise ValueError("error fragment must carry a 'code'") + + @property + def exit_code(self) -> int: + """Process exit code implied by :attr:`status`.""" + + return exit_code_for(self.status) + + @property + def succeeded(self) -> bool: + """True when the status is one of the zero-exit statuses.""" + + return self.exit_code == 0 + + def to_dict(self) -> dict[str, Any]: + """Return the JSON-ready document with a fixed set of top-level keys.""" + + return { + "status": self.status, + "source": self.source, + "request": dict(self.request) if self.request is not None else None, + "items": [dict(item) for item in self.items], + "error": dict(self.error) if self.error is not None else None, + } + + def to_json(self) -> str: + """Serialise the response as one deterministic JSON line.""" + + return json.dumps(self.to_dict(), sort_keys=True, ensure_ascii=True) + + def emit(self, stream: TextIO | None = None) -> None: + """Write exactly one JSON document to ``stream`` (stdout by default).""" + + print(self.to_json(), file=stream if stream is not None else sys.stdout) + + +class ResponseBuilder: + """Build :class:`Response` objects sharing one source/request context.""" + + def __init__( + self, + *, + source: str | None = None, + request: Mapping[str, Any] | None = None, + ) -> None: + self.source = source + self.request = request + + def with_context( + self, + *, + source: str | None = None, + request: Mapping[str, Any] | None = None, + ) -> "ResponseBuilder": + """Return a copy carrying the supplied context.""" + + return ResponseBuilder( + source=self.source if source is None else source, + request=self.request if request is None else request, + ) + + def build( + self, + status: str, + *, + items: Iterable[Mapping[str, Any]] = (), + error: Mapping[str, Any] | None = None, + ) -> Response: + """Assemble a response for an explicit ``status``.""" + + return Response( + status=status, + source=self.source, + request=self.request, + items=tuple(items), + error=error, + ) + + # -- convenience constructors -------------------------------------- + def ok(self, *, items: Iterable[Mapping[str, Any]] = ()) -> Response: + return self.build(STATUS_OK, items=items) + + def dry_run(self, *, items: Iterable[Mapping[str, Any]] = ()) -> Response: + return self.build(STATUS_DRY_RUN, items=items) + + def invalid_request(self, *, error: Mapping[str, Any] | None = None) -> Response: + return self.build(STATUS_INVALID_REQUEST, error=error) + + def no_format(self, *, error: Mapping[str, Any] | None = None) -> Response: + return self.build(STATUS_NO_FORMAT, error=error) + + def no_size(self, *, error: Mapping[str, Any] | None = None) -> Response: + return self.build(STATUS_NO_SIZE, error=error) + + def access_error( + self, + *, + items: Iterable[Mapping[str, Any]] = (), + error: Mapping[str, Any] | None = None, + ) -> Response: + return self.build(STATUS_ACCESS_ERROR, items=items, error=error) + + def partial_download( + self, + *, + items: Iterable[Mapping[str, Any]] = (), + error: Mapping[str, Any] | None = None, + ) -> Response: + return self.build(STATUS_PARTIAL_DOWNLOAD, items=items, error=error) + + def from_error(self, error: ToolkitError) -> Response: + """Build a response for ``error`` using :func:`status_for_error`.""" + + return self.build(status_for_error(error), error=error.to_dict()) diff --git a/tools/errors.py b/tools/errors.py index 013ae8b..28c05de 100644 --- a/tools/errors.py +++ b/tools/errors.py @@ -185,3 +185,89 @@ class HostNotAllowedError(SourceAccessError): url=str(url), details={"allowed_hosts": list(self.allowed)}, ) + + +# --------------------------------------------------------------------------- +# Request errors: invalid command lines and JSON request documents +# --------------------------------------------------------------------------- + +#: Image formats the CLI accepts. +ACCEPTED_FORMATS: tuple[str, ...] = ("png", "ico") + + +class RequestError(ToolkitError): + """Base class for every unusable request (CLI flags or request document).""" + + code = "request_invalid" + + def __init__( + self, + message: str, + *, + code: str | None = None, + details: Mapping[str, Any] | None = None, + field: str | None = None, + accepted_fields: Iterable[str] | None = None, + accepted_formats: Iterable[str] | None = None, + ) -> None: + merged: dict[str, Any] = dict(details or {}) + if field is not None: + merged["field"] = field + if accepted_fields is not None: + merged["accepted_fields"] = list(accepted_fields) + if accepted_formats is not None: + merged["accepted_formats"] = list(accepted_formats) + super().__init__(message, code=code, details=merged) + self.field = field + self.accepted_fields: tuple[str, ...] = tuple(accepted_fields or ()) + self.accepted_formats: tuple[str, ...] = tuple(accepted_formats or ()) + + +class RequestDocumentError(RequestError): + """Raised when the JSON request document cannot be read or decoded.""" + + code = "request_document" + + +class RequestFieldError(RequestError): + """Raised when a single request field is missing, unknown or malformed.""" + + code = "request_field" + + +class UnsupportedFormatError(RequestError): + """Raised for a format outside :data:`ACCEPTED_FORMATS`.""" + + code = "request_format_unsupported" + + def __init__(self, value: Any, accepted: Iterable[str] = ACCEPTED_FORMATS) -> None: + accepted = tuple(accepted) + self.value = str(value) + super().__init__( + f"unsupported image format {self.value!r}; " + f"accepted formats: {', '.join(accepted)}", + field="format", + accepted_formats=accepted, + details={"requested_format": self.value}, + ) + + +class ImageResizeError(ToolkitError): + """Raised when a downloaded image cannot be downscaled to the target size.""" + + code = "resize_failed" + + def __init__(self, reason: str, *, details: Mapping[str, Any] | None = None) -> None: + merged: dict[str, Any] = dict(details or {}) + merged["reason"] = reason + super().__init__(f"cannot resize image: {reason}", details=merged) + + +__all__ += [ + "ACCEPTED_FORMATS", + "ImageResizeError", + "RequestDocumentError", + "RequestError", + "RequestFieldError", + "UnsupportedFormatError", +] diff --git a/tools/example.py b/tools/example.py index 32d2b87..ef733af 100644 --- a/tools/example.py +++ b/tools/example.py @@ -1,200 +1,844 @@ -"""Placeholder toolkit entry point that loads and validates a JSON request. - -This increment only establishes the stable command-line surface: read a single -JSON request document, validate its shape, and print one structured JSON -envelope. Later modules extend the same document format and envelope without -changing the invocation contract, so the module stays deterministic and works -fully offline. +"""Plain native-process CLI for the image toolkit. Usage:: python -m tools.example examples/input.json + python -m tools.example --topic "red fox" --width 640 --height 480 --format png + +Exactly one machine-readable JSON document is written to stdout. The process +exit code is derived from the fixed status vocabulary in :mod:`tools.envelope`. """ from __future__ import annotations import argparse import json +import re +import struct import sys +import zlib from pathlib import Path from typing import Any, Mapping, Sequence -__all__ = [ - "EXIT_INVALID_REQUEST", - "EXIT_OK", - "EXIT_USAGE", - "RequestError", - "build_envelope", - "load_request", - "main", - "validate_request", -] +from .envelope import ( + STATUS_ACCESS_ERROR, + STATUS_NO_FORMAT, + STATUS_NO_SIZE, + STATUS_OK, + STATUS_PARTIAL_DOWNLOAD, + Response, + ResponseBuilder, + error_fragment, +) +from .errors import ( + ACCEPTED_FORMATS, + ImageResizeError, + RequestDocumentError, + RequestError, + RequestFieldError, + SourceAccessError, + ToolkitError, + UnknownSourceError, + UnsupportedFormatError, +) +from .sources import DEFAULT_LIMIT, SourceCandidate, default_registry -EXIT_OK = 0 -EXIT_INVALID_REQUEST = 1 -EXIT_USAGE = 2 +PROG = "python -m tools.example" -#: Fields every request document must provide. -REQUIRED_FIELDS = ("version", "mode", "dry_run", "command", "arguments") +#: Keys accepted in a JSON request document (``version`` is optional metadata). +REQUEST_FIELDS: tuple[str, ...] = ( + "topic", + "width", + "height", + "format", + "source", + "count", + "download", + "output_dir", + "resize", + "dry_run", +) -#: Modes understood by this increment of the toolkit. -SUPPORTED_MODES = frozenset({"validate"}) +OPTIONAL_DOCUMENT_FIELDS: tuple[str, ...] = ("version",) -PROG = "tools.example" +DEFAULT_SOURCE = "wikimedia_commons" +DEFAULT_OUTPUT_DIR = "downloads" +MIN_COUNT = 1 +MAX_COUNT = 50 + +#: MIME types (and filename suffixes) that count as a hit for each format. +FORMAT_MATCHERS: dict[str, tuple[str, ...]] = { + "png": ("image/png", ".png"), + "ico": ("image/x-icon", "image/vnd.microsoft.icon", "image/ico", ".ico"), +} -class RequestError(ValueError): - """Raised when a request document is missing or malformed.""" +class _Parser(argparse.ArgumentParser): + """Argument parser that reports bad flags as request errors.""" - def __init__(self, code: str, message: str) -> None: - super().__init__(message) - self.code = code - self.message = message + def error(self, message: str) -> Any: # pragma: no cover - argparse hook + raise RequestError(message, accepted_fields=REQUEST_FIELDS) -def load_request(path: str | Path) -> Any: - """Read and decode the JSON document stored at ``path``.""" - try: - raw = Path(path).read_text(encoding="utf-8") - except FileNotFoundError as exc: - raise RequestError("request_not_found", f"request document not found: {path}") from exc - except IsADirectoryError as exc: - raise RequestError("request_not_a_file", f"request document is not a file: {path}") from exc - except OSError as exc: - raise RequestError("request_unreadable", f"request document could not be read: {path}") from exc - except UnicodeDecodeError as exc: - raise RequestError("request_not_utf8", f"request document is not valid UTF-8: {path}") from exc +def build_parser() -> argparse.ArgumentParser: + """Return the CLI parser; ``--help`` documents every supported parameter.""" - try: - return json.loads(raw) - except json.JSONDecodeError as exc: - raise RequestError( - "request_not_json", - f"request document is not valid JSON: {path} (line {exc.lineno}, column {exc.colno})", - ) from exc - - -def _require(mapping: Mapping[str, Any], key: str, expected: type | tuple[type, ...], label: str) -> Any: - if key not in mapping: - raise RequestError("request_field_missing", f"request field {label} is required") - value = mapping[key] - if isinstance(value, bool) and bool not in ( - expected if isinstance(expected, tuple) else (expected,) - ): - raise RequestError("request_field_invalid", f"request field {label} must be {_describe(expected)}") - if not isinstance(value, expected): - raise RequestError("request_field_invalid", f"request field {label} must be {_describe(expected)}") - return value - - -def _describe(expected: type | tuple[type, ...]) -> str: - names = expected if isinstance(expected, tuple) else (expected,) - return " or ".join(sorted({name.__name__ for name in names})) - - -def validate_request(document: Any) -> dict[str, Any]: - """Validate ``document`` and return the normalised request mapping.""" - if not isinstance(document, dict): - raise RequestError("request_not_object", "request document must be a JSON object") - - for key in REQUIRED_FIELDS: - if key not in document: - raise RequestError("request_field_missing", f"request field '{key}' is required") - - version = _require(document, "version", int, "version") - if version < 1: - raise RequestError("request_field_invalid", "request field 'version' must be a positive integer") - - mode = _require(document, "mode", str, "mode") - if mode not in SUPPORTED_MODES: - supported = ", ".join(sorted(SUPPORTED_MODES)) - raise RequestError( - "request_mode_unsupported", - f"request field 'mode' must be one of: {supported}", - ) - - dry_run = _require(document, "dry_run", bool, "dry_run") - if not dry_run: - raise RequestError( - "request_not_dry_run", - "this increment only supports validate-only dry_run requests", - ) - - command = _require(document, "command", str, "command") - if not command.strip(): - raise RequestError("request_field_invalid", "request field 'command' must not be empty") - - arguments = _require(document, "arguments", dict, "arguments") - request_path = _require(arguments, "request", str, "arguments.request") - if not request_path.strip(): - raise RequestError("request_field_invalid", "request field 'arguments.request' must not be empty") - - return { - "version": version, - "mode": mode, - "dry_run": dry_run, - "command": command, - "arguments": dict(arguments), - } - - -def build_envelope(request: Mapping[str, Any], source: str) -> dict[str, Any]: - """Build the deterministic success envelope for a validated request.""" - return { - "status": "ok", - "command": request["command"], - "mode": request["mode"], - "dry_run": request["dry_run"], - "version": request["version"], - "request": source, - } - - -def build_error_envelope(error: RequestError, source: str) -> dict[str, Any]: - """Build the deterministic failure envelope for an invalid request.""" - return { - "status": "error", - "error": {"code": error.code, "message": error.message}, - "request": source, - } - - -def emit(envelope: Mapping[str, Any]) -> None: - """Print exactly one JSON document to standard output.""" - print(json.dumps(envelope, sort_keys=True, ensure_ascii=True)) - - -def _build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser( + parser = _Parser( prog=PROG, description=( - "Load and validate a JSON request document. The command is offline " - "and prints one structured JSON envelope." + "Search an image source and report the matching items as one JSON " + "document. Either pass a JSON request document or use the flags." + ), + epilog=( + "Every run writes exactly one JSON document to stdout with the keys " + "status, source, request, items and error." ), ) parser.add_argument( "request", - metavar="REQUEST", - help="path to the JSON request document (for example examples/input.json)", + nargs="?", + metavar="REQUEST.json", + help=( + "path to a JSON request document holding the same fields as the " + "flags; it cannot be combined with any of the flags" + ), + ) + parser.add_argument( + "-t", + "--topic", + metavar="TEXT", + help="search topic (required unless a request document is given)", + ) + parser.add_argument( + "--width", + type=int, + metavar="PX", + help="target width in pixels; only images at least this wide are accepted", + ) + parser.add_argument( + "--height", + type=int, + metavar="PX", + help="target height in pixels; only images at least this tall are accepted", + ) + parser.add_argument( + "-f", + "--format", + metavar="FMT", + help="image format: " + "|".join(ACCEPTED_FORMATS), + ) + parser.add_argument( + "-s", + "--source", + metavar="ID", + help=f"image source id (default: {DEFAULT_SOURCE}); unknown ids are rejected", + ) + parser.add_argument( + "-c", + "--count", + type=int, + metavar="N", + help=f"how many candidates to request, {MIN_COUNT}-{MAX_COUNT} (default: {DEFAULT_LIMIT})", + ) + parser.add_argument( + "--download", + action="store_true", + help="download every matching image into the output directory", + ) + parser.add_argument( + "-o", + "--output-dir", + metavar="DIR", + help=f"download directory (default: ./{DEFAULT_OUTPUT_DIR})", + ) + parser.add_argument( + "--resize", + action="store_true", + help=( + "downscale every downloaded image so it fits the requested width " + "and height, keeping the aspect ratio and never scaling up; " + "without it the original bytes are written and the actual size is " + "reported" + ), + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="validate the request and emit the envelope without any network access", ) return parser -def main(argv: Sequence[str] | None = None) -> int: - """Run the placeholder CLI and return the process exit code.""" - parser = _build_parser() - args = parser.parse_args(list(sys.argv[1:] if argv is None else argv)) +def _load_document(path: str) -> Mapping[str, Any]: + """Read a JSON request document from ``path``.""" + target = Path(path) try: - document = load_request(args.request) - request = validate_request(document) - except RequestError as error: - emit(build_error_envelope(error, args.request)) - return EXIT_INVALID_REQUEST - - emit(build_envelope(request, args.request)) - return EXIT_OK + text = target.read_text(encoding="utf-8") + except OSError as exc: + raise RequestDocumentError( + f"cannot read request document {path!r}: {exc.strerror or exc}", + details={"path": str(path)}, + ) from exc + try: + payload = json.loads(text) + except json.JSONDecodeError as exc: + raise RequestDocumentError( + f"request document {path!r} is not valid JSON: {exc.msg}", + details={"path": str(path), "line": exc.lineno, "column": exc.colno}, + ) from exc + if not isinstance(payload, Mapping): + raise RequestDocumentError( + f"request document {path!r} must contain a JSON object", + details={"path": str(path), "type": type(payload).__name__}, + ) + return payload -if __name__ == "__main__": - raise SystemExit(main()) +def _document_to_mapping(payload: Mapping[str, Any]) -> dict[str, Any]: + """Reject unknown keys and return the accepted request fields.""" + + accepted = REQUEST_FIELDS + OPTIONAL_DOCUMENT_FIELDS + unknown = sorted(str(key) for key in payload if key not in accepted) + if unknown: + raise RequestFieldError( + "request document contains unsupported fields: " + ", ".join(unknown), + field=unknown[0], + details={"unknown_fields": unknown, "accepted_fields": list(REQUEST_FIELDS)}, + ) + return {key: payload[key] for key in REQUEST_FIELDS if key in payload} + + +def _flags_to_mapping(namespace: argparse.Namespace) -> dict[str, Any]: + """Collect the explicitly supplied flags into a request mapping.""" + + mapping: dict[str, Any] = {} + for field in ("topic", "width", "height", "format", "source", "count", "output_dir"): + value = getattr(namespace, field, None) + if value is not None: + mapping[field] = value + for field in ("download", "resize", "dry_run"): + if getattr(namespace, field, False): + mapping[field] = True + return mapping + + +def _as_text(value: Any, field: str) -> str: + if not isinstance(value, str): + raise RequestFieldError( + f"{field} must be a string, got {type(value).__name__}", field=field + ) + text = value.strip() + if not text: + raise RequestFieldError(f"{field} must not be empty", field=field) + return text + + +def _as_int(value: Any, field: str) -> int: + if isinstance(value, bool) or not isinstance(value, int): + raise RequestFieldError( + f"{field} must be an integer, got {type(value).__name__}", field=field + ) + return int(value) + + +def _as_bool(value: Any, field: str) -> bool: + if not isinstance(value, bool): + raise RequestFieldError( + f"{field} must be a boolean, got {type(value).__name__}", field=field + ) + return bool(value) + + +def normalize_request(raw: Mapping[str, Any]) -> dict[str, Any]: + """Validate ``raw`` and return the complete, typed request mapping.""" + + if "topic" not in raw: + raise RequestFieldError( + "topic is required", + field="topic", + details={"required_fields": ["topic"], "accepted_fields": list(REQUEST_FIELDS)}, + ) + topic = _as_text(raw["topic"], "topic") + + size: dict[str, int | None] = {"width": None, "height": None} + for field in ("width", "height"): + if raw.get(field) is None: + continue + number = _as_int(raw[field], field) + if number <= 0: + raise RequestFieldError(f"{field} must be positive", field=field) + size[field] = number + + if "format" not in raw or raw["format"] is None: + raise RequestFieldError( + "format is required", + field="format", + details={"accepted_formats": list(ACCEPTED_FORMATS)}, + ) + image_format = _as_text(raw["format"], "format").lower() + if image_format not in ACCEPTED_FORMATS: + raise UnsupportedFormatError(image_format) + + source = DEFAULT_SOURCE + if raw.get("source") is not None: + source = _as_text(raw["source"], "source") + + count = DEFAULT_LIMIT + if raw.get("count") is not None: + count = _as_int(raw["count"], "count") + if not MIN_COUNT <= count <= MAX_COUNT: + raise RequestFieldError( + f"count must be between {MIN_COUNT} and {MAX_COUNT}", + field="count", + details={"min": MIN_COUNT, "max": MAX_COUNT}, + ) + + output_dir = DEFAULT_OUTPUT_DIR + if raw.get("output_dir") is not None: + output_dir = _as_text(raw["output_dir"], "output_dir") + + return { + "topic": topic, + "width": size["width"], + "height": size["height"], + "format": image_format, + "source": source, + "count": count, + "download": _as_bool(raw.get("download", False), "download"), + "output_dir": output_dir, + "resize": _as_bool(raw.get("resize", False), "resize"), + "dry_run": _as_bool(raw.get("dry_run", False), "dry_run"), + } + + +def _matches_format(candidate: SourceCandidate, image_format: str) -> bool: + matchers = FORMAT_MATCHERS[image_format] + mime = str(getattr(candidate, "mime_type", None) or "").split(";")[0].strip().lower() + if mime and mime in matchers: + return True + url = str(getattr(candidate, "file_url", "") or "").lower() + return any(url.endswith(suffix) for suffix in matchers if suffix.startswith(".")) + + +def _matches_size(candidate: SourceCandidate, request: Mapping[str, Any]) -> bool: + """True when ``candidate`` is usable for the requested target size. + + The requested width and height are a target size, not an exact match: any + image that is at least that large is usable because it can always be + downscaled, while a smaller image would have to be scaled up and is + therefore rejected. ``--resize`` only decides whether the downloaded bytes + are downscaled; it never widens or narrows this filter. + """ + + wanted_width = request.get("width") + wanted_height = request.get("height") + if wanted_width is None and wanted_height is None: + return True + actual_width = getattr(candidate, "width", None) + actual_height = getattr(candidate, "height", None) + if isinstance(actual_width, bool) or isinstance(actual_height, bool): + return False + if not isinstance(actual_width, int) or not isinstance(actual_height, int): + return False + if wanted_width is not None and actual_width < wanted_width: + return False + if wanted_height is not None and actual_height < wanted_height: + return False + return True + + +def _select(candidates: Sequence[SourceCandidate], request: Mapping[str, Any]) -> tuple[list[SourceCandidate], list[SourceCandidate]]: + """Return ``(matching, rejected_by_size)`` split for the request.""" + + by_format = [c for c in candidates if _matches_format(c, request["format"])] + matching = [c for c in by_format if _matches_size(c, request)] + rejected = [c for c in by_format if not _matches_size(c, request)] + return matching, rejected + + +def _slug(candidate: SourceCandidate) -> str: + raw = str(getattr(candidate, "title", None) or getattr(candidate, "source_id", "") or "image").lower() + slug = re.sub(r"[^a-z0-9]+", "-", raw).strip("-") + return (slug or "image")[:60] + + +PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n" +ICO_HEADER_SIZE = 6 +ICO_ENTRY_SIZE = 16 +ICO_MAX_DIMENSION = 256 +_PNG_CHANNELS = {0: 1, 2: 3, 3: 1, 4: 2, 6: 4} + + +def _png_chunks(payload: bytes) -> list[tuple[bytes, bytes]]: + """Split a PNG payload into ``(type, data)`` chunks.""" + + if not payload.startswith(PNG_SIGNATURE): + raise ImageResizeError("payload is not a PNG image") + chunks: list[tuple[bytes, bytes]] = [] + offset = len(PNG_SIGNATURE) + while offset + 8 <= len(payload): + (length,) = struct.unpack(">I", payload[offset : offset + 4]) + chunk_type = payload[offset + 4 : offset + 8] + data = payload[offset + 8 : offset + 8 + length] + if len(data) != length: + raise ImageResizeError("truncated PNG chunk") + chunks.append((chunk_type, data)) + offset += 12 + length + if chunk_type == b"IEND": + return chunks + raise ImageResizeError("PNG image has no IEND chunk") + + +def _unfilter_rows(raw: bytes, width: int, height: int, bpp: int) -> list[bytearray]: + """Undo the per-scanline PNG filters, returning one row per pixel line.""" + + stride = width * bpp + if len(raw) < height * (stride + 1): + raise ImageResizeError("truncated PNG image data") + rows: list[bytearray] = [] + previous = bytearray(stride) + position = 0 + for _ in range(height): + filter_type = raw[position] + position += 1 + line = bytearray(raw[position : position + stride]) + position += stride + if filter_type == 1: + for index in range(bpp, stride): + line[index] = (line[index] + line[index - bpp]) & 0xFF + elif filter_type == 2: + for index in range(stride): + line[index] = (line[index] + previous[index]) & 0xFF + elif filter_type == 3: + for index in range(stride): + left = line[index - bpp] if index >= bpp else 0 + line[index] = (line[index] + ((left + previous[index]) >> 1)) & 0xFF + elif filter_type == 4: + for index in range(stride): + left = line[index - bpp] if index >= bpp else 0 + up = previous[index] + up_left = previous[index - bpp] if index >= bpp else 0 + estimate = left + up - up_left + to_left = abs(estimate - left) + to_up = abs(estimate - up) + to_up_left = abs(estimate - up_left) + if to_left <= to_up and to_left <= to_up_left: + predictor = left + elif to_up <= to_up_left: + predictor = up + else: + predictor = up_left + line[index] = (line[index] + predictor) & 0xFF + elif filter_type != 0: + raise ImageResizeError(f"unsupported PNG filter type {filter_type}") + previous = line + rows.append(line) + return rows + + +def _row_to_rgba( + row: bytearray, + width: int, + color_type: int, + depth: int, + palette: bytes, + transparency: bytes | None, +) -> bytes: + """Convert one unfiltered scanline into 8-bit RGBA samples.""" + + step = depth // 8 + out = bytearray() + if color_type == 3: + for x in range(width): + index = row[x] + base = index * 3 + if base + 2 >= len(palette): + raise ImageResizeError("PNG palette index is out of range") + alpha = transparency[index] if transparency and index < len(transparency) else 255 + out += bytes((palette[base], palette[base + 1], palette[base + 2], alpha)) + return bytes(out) + for x in range(width): + offset = x * step + if color_type == 0: + value = row[offset] + out += bytes((value, value, value, 255)) + elif color_type == 4: + value = row[offset] + out += bytes((value, value, value, row[offset + step])) + elif color_type == 2: + out += bytes((row[offset], row[offset + step], row[offset + 2 * step], 255)) + else: + out += bytes( + ( + row[offset], + row[offset + step], + row[offset + 2 * step], + row[offset + 3 * step], + ) + ) + return bytes(out) + + +def _decode_png(payload: bytes) -> tuple[int, int, list[bytes]]: + """Decode a PNG payload into ``(width, height, rgba_rows)``.""" + + header: tuple[int, ...] | None = None + palette = b"" + transparency: bytes | None = None + data = bytearray() + for chunk_type, chunk_data in _png_chunks(payload): + if chunk_type == b"IHDR": + if len(chunk_data) < 13: + raise ImageResizeError("truncated PNG header") + header = struct.unpack(">IIBBBBB", chunk_data[:13]) + elif chunk_type == b"PLTE": + palette = chunk_data + elif chunk_type == b"tRNS": + transparency = chunk_data + elif chunk_type == b"IDAT": + data += chunk_data + if header is None: + raise ImageResizeError("PNG image has no IHDR chunk") + width, height, depth, color_type, compression, filter_method, interlace = header + if width < 1 or height < 1: + raise ImageResizeError("PNG image has no pixels") + if compression != 0 or filter_method != 0: + raise ImageResizeError("unsupported PNG compression or filter method") + if interlace != 0: + raise ImageResizeError("interlaced PNG images are not supported") + channels = _PNG_CHANNELS.get(color_type) + if channels is None: + raise ImageResizeError(f"unsupported PNG colour type {color_type}") + if depth not in (8, 16): + raise ImageResizeError(f"unsupported PNG bit depth {depth}") + if color_type == 3 and depth > 8: + raise ImageResizeError("unsupported PNG palette bit depth") + if color_type == 3 and not palette: + raise ImageResizeError("PNG palette image has no PLTE chunk") + try: + raw = zlib.decompress(bytes(data)) + except zlib.error as exc: + raise ImageResizeError(f"PNG image data is corrupt: {exc}") from exc + rows = _unfilter_rows(raw, width, height, channels * (depth // 8)) + return width, height, [ + _row_to_rgba(row, width, color_type, depth, palette, transparency) for row in rows + ] + + +def _encode_png(width: int, height: int, rows: Sequence[bytes]) -> bytes: + """Encode RGBA rows as an uncompressed-filter 8-bit PNG.""" + + raw = bytearray() + for row in rows: + raw.append(0) + raw += row + + def chunk(chunk_type: bytes, chunk_data: bytes) -> bytes: + crc = zlib.crc32(chunk_type + chunk_data) & 0xFFFFFFFF + return struct.pack(">I", len(chunk_data)) + chunk_type + chunk_data + struct.pack(">I", crc) + + header = struct.pack(">IIBBBBB", width, height, 8, 6, 0, 0, 0) + return ( + PNG_SIGNATURE + + chunk(b"IHDR", header) + + chunk(b"IDAT", zlib.compress(bytes(raw), 6)) + + chunk(b"IEND", b"") + ) + + +def _ico_entries(payload: bytes) -> list[tuple[int, int, int, int]]: + """Return ``(width, height, size, offset)`` for every ICO directory entry.""" + + if len(payload) < ICO_HEADER_SIZE: + raise ImageResizeError("payload is not an ICO image") + reserved, kind, count = struct.unpack(" len(payload): + raise ImageResizeError("truncated ICO directory") + width, height, _colours, _reserved, _planes, _bits, size, offset = struct.unpack( + " bytes: + """Wrap a PNG payload in a single-entry ICO container.""" + + if width > ICO_MAX_DIMENSION or height > ICO_MAX_DIMENSION: + raise ImageResizeError("ICO images cannot exceed 256x256 pixels") + header = struct.pack(" tuple[str, int, int, list[bytes]]: + """Decode a PNG or ICO payload into ``(container, width, height, rows)``.""" + + if payload.startswith(PNG_SIGNATURE): + width, height, rows = _decode_png(payload) + return "png", width, height, rows + if len(payload) >= ICO_HEADER_SIZE and payload[:4] == b"\x00\x00\x01\x00": + entries = _ico_entries(payload) + _width, _height, size, offset = max(entries, key=lambda item: item[0] * item[1]) + embedded = payload[offset : offset + size] + if not embedded.startswith(PNG_SIGNATURE): + raise ImageResizeError("ICO images stored as raw bitmaps are not supported") + width, height, rows = _decode_png(embedded) + return "ico", width, height, rows + raise ImageResizeError("payload is neither a PNG nor an ICO image") + + +def _target_size(width: int, height: int, request: Mapping[str, Any]) -> tuple[int, int]: + """Shrink-to-fit target inside the requested box, never enlarging.""" + + wanted_width = request.get("width") + wanted_height = request.get("height") + scale = 1.0 + if wanted_width is not None: + scale = min(scale, wanted_width / width) + if wanted_height is not None: + scale = min(scale, wanted_height / height) + if scale >= 1.0: + return width, height + return max(1, int(round(width * scale))), max(1, int(round(height * scale))) + + +def _resample(rows: Sequence[bytes], width: int, height: int, target_width: int, target_height: int) -> list[bytes]: + """Nearest-neighbour downscale of RGBA rows.""" + + columns = [min(width - 1, (x * width) // target_width) * 4 for x in range(target_width)] + resized: list[bytes] = [] + for y in range(target_height): + source = rows[min(height - 1, (y * height) // target_height)] + line = bytearray() + for offset in columns: + line += source[offset : offset + 4] + resized.append(bytes(line)) + return resized + + +def resize_image(payload: bytes, request: Mapping[str, Any]) -> tuple[bytes, dict[str, Any]]: + """Downscale ``payload`` so it fits the requested size. + + The requested width and height are a bounding box: the aspect ratio is + preserved, an image that already fits is left untouched and an image is + never scaled up. The returned fragment records what happened so the caller + can report the real dimensions of the stored file. + """ + + container, width, height, rows = _decode_image(payload) + target_width, target_height = _target_size(width, height, request) + fragment: dict[str, Any] = { + "requested": {"width": request.get("width"), "height": request.get("height")}, + "from": {"width": width, "height": height}, + } + if (target_width, target_height) == (width, height): + fragment.update( + {"applied": False, "reason": "already_fits", "to": {"width": width, "height": height}} + ) + return payload, fragment + png_payload = _encode_png( + target_width, target_height, _resample(rows, width, height, target_width, target_height) + ) + output = ( + _encode_ico(png_payload, target_width, target_height) + if container == "ico" + else png_payload + ) + fragment.update({"applied": True, "to": {"width": target_width, "height": target_height}}) + return output, fragment + + +def _download(candidates: Sequence[SourceCandidate], request: Mapping[str, Any]) -> tuple[list[dict[str, Any]], int]: + """Download ``candidates``; return ``(items, failure_count)``. + + With ``resize`` enabled every stored file is downscaled to the requested + target size first. A payload that cannot be decoded or downscaled is + reported as a failed item rather than silently stored at the wrong size. + """ + + directory = Path(request["output_dir"]).expanduser() + try: + directory.mkdir(parents=True, exist_ok=True) + except OSError as exc: + raise SourceAccessError( + f"cannot create output directory {str(directory)!r}: {exc.strerror or exc}", + details={"output_dir": str(directory)}, + ) from exc + + suffix = "." + str(request["format"]) + items: list[dict[str, Any]] = [] + failures = 0 + for index, candidate in enumerate(candidates, start=1): + record = dict(candidate.to_dict()) + destination = directory / f"{index:03d}-{_slug(candidate)}{suffix}" + try: + payload = candidate.fetch_bytes() + resize_fragment = resize_image(payload, request) if request["resize"] else None + if resize_fragment is not None: + payload, resize_fragment = resize_fragment + destination.write_bytes(payload) + except (ToolkitError, OSError) as exc: + failures += 1 + error = error_fragment(getattr(exc, "code", "download_failed"), str(exc)) + if isinstance(exc, ToolkitError) and exc.details: + error["details"] = dict(exc.details) + record["download"] = {"ok": False, "error": error} + else: + record["download"] = {"ok": True, "path": str(destination), "bytes": len(payload)} + if resize_fragment is not None: + record["download"]["resize"] = resize_fragment + record["download"]["width"] = resize_fragment["to"]["width"] + record["download"]["height"] = resize_fragment["to"]["height"] + items.append(record) + return items, failures + + +def _guess_source(argv: Sequence[str]) -> str | None: + """Best-effort source id extraction for error envelopes.""" + + for index, token in enumerate(argv): + if token == "--source" or token == "-s": + if index + 1 < len(argv): + return argv[index + 1] + elif token.startswith("--source="): + return token.split("=", 1)[1] + return None + + +def _request_from(argv: Sequence[str]) -> tuple[dict[str, Any], str, Any]: + """Turn CLI arguments into ``(normalized_request, source_id, registry)``.""" + + parser = build_parser() + namespace = parser.parse_args(list(argv)) + flags = _flags_to_mapping(namespace) + document = namespace.request + + if document is not None: + if flags: + raise RequestError( + "a request document cannot be combined with flags: " + + ", ".join(sorted(flags)), + details={"conflicting_flags": sorted(flags)}, + accepted_fields=REQUEST_FIELDS, + ) + raw = _document_to_mapping(_load_document(document)) + elif flags: + raw = flags + else: + raise RequestError( + "no request given; pass a JSON request document or at least one flag", + accepted_fields=REQUEST_FIELDS, + ) + + request = normalize_request(raw) + registry = default_registry() + available = list(registry.available_sources()) + if request["source"] not in available: + raise UnknownSourceError(request["source"], available) + return request, request["source"], registry + + +def _execute(argv: Sequence[str]) -> Response: + """Run one request and return the response that must be printed.""" + + request, source, registry = _request_from(argv) + builder = ResponseBuilder(source=source, request=request) + + if request["dry_run"]: + return builder.dry_run() + + candidates = registry.search(source, request["topic"], request["count"]) + matching, rejected = _select(candidates, request) + + if not matching: + details = { + "requested_format": request["format"], + "requested_size": {"width": request["width"], "height": request["height"]}, + "minimum_size": {"width": request["width"], "height": request["height"]}, + "candidate_count": len(candidates), + "format_matches": len(matching) + len(rejected), + "size_matches": 0, + } + if rejected: + return builder.build( + STATUS_NO_SIZE, + error=error_fragment( + "no_size_match", + "no image is at least the requested width and height", + details=details, + ), + ) + return builder.build( + STATUS_NO_FORMAT, + error=error_fragment( + "no_format_match", + f"the source returned no {request['format']} images for {request['topic']!r}", + details=details, + ), + ) + + if not request["download"]: + return builder.build(STATUS_OK, items=[c.to_dict() for c in matching]) + + items, failures = _download(matching, request) + if failures and failures == len(items): + return builder.build( + STATUS_ACCESS_ERROR, + items=items, + error=error_fragment( + "download_failed", + "every download failed", + details={"failed": failures, "total": len(items)}, + ), + ) + if failures: + return builder.build( + STATUS_PARTIAL_DOWNLOAD, + items=items, + error=error_fragment( + "download_incomplete", + f"{failures} of {len(items)} downloads failed", + details={"failed": failures, "total": len(items)}, + ), + ) + return builder.build(STATUS_OK, items=items) + + +def main(argv: Sequence[str] | None = None) -> int: + """Entry point: emit exactly one JSON document and return the exit code.""" + + arguments = list(sys.argv[1:] if argv is None else argv) + try: + response = _execute(arguments) + except ToolkitError as error: + builder = ResponseBuilder(source=_guess_source(arguments)) + response = builder.from_error(error) + except OSError as error: + builder = ResponseBuilder(source=_guess_source(arguments)) + response = builder.build( + STATUS_ACCESS_ERROR, + error=error_fragment("access_error", str(error)), + ) + response.emit() + return response.exit_code + + +if __name__ == "__main__": # pragma: no cover - module entry point + sys.exit(main())