"""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"