diff --git a/.agent-work/ledger.json b/.agent-work/ledger.json index 774c236..152f680 100644 --- a/.agent-work/ledger.json +++ b/.agent-work/ledger.json @@ -15,6 +15,12 @@ "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." ] + }, + "projects/product-54354138-1#6": { + "events": [ + "Implemented byte-based format and minimum-size filtering for the image search path. New tools/images.py: detect_format (png/ico plus jpeg/gif/bmp/webp so \"other format\" is distinguishable from \"unreadable\"), read_png_dimensions (IHDR), read_ico_dimensions (ICONDIR entries, 0 => 256, largest entry), inspect_image/ImageInfo, ImageFormatError (code image_unreadable), and png_bytes/ico_bytes builders. tools/example.py: candidate_cap(count) bounded window (count*4, max 50) requested from the source, _select() fetches and parses every candidate, KeptImage.to_item() reports source/url/format/actual width+height/size_bytes, _download() writes the already verified bytes, and _execute() distinguishes ok/no_format/no_size/access_error/partial_download with the existing exit codes; metadata-based filtering removed. tests/test_images.py added (31 tests) and tests/test_cli.py fixtures aligned to real image bytes. All facade commands pass: compileall, pytest (114 passed), smoke, help, package import.", + "Addressed changes-requested review on PR #14: the CLI no longer turns common live-source results into partial_download/access_error. Any payload whose real format is not the requested png/ico (jpeg, gif, bmp, webp, svg, tiff, pdf, or an unrecognised body such as an HTML page) is now a format mismatch feeding no_format, with observed_formats/other_format_candidates detail; only unfetchable candidates are failures, and access_error is emitted only when nothing at all could be fetched. images.detect_format gained svg/tiff/pdf recognition and ImageFormatError is scoped to genuinely unusable png/ico payloads. Statuses/exits unchanged (no_format 3, no_size 4, access_error 5). Tests extended in tests/test_images.py (svg/tiff/pdf detection, mixed-format window stays ok, ico-only-foreign yields no_format, no_size preserved alongside other formats, non-image body is no_format, unfetchable candidate does not demote a search, corrupt png does not hide a match, empty window) and tests/test_cli.py aligned. Facade validate/test/smoke/help all green (129 passed) on the integrated branch; no protected facade paths touched." + ] } } } diff --git a/tests/test_cli.py b/tests/test_cli.py index bd7f362..9adbc3f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -194,7 +194,7 @@ class FakeCandidate: width: int | None, height: int | None, *, - payload: bytes = b"payload", + payload: bytes | None = None, fail: bool = False, ) -> None: self.source_id = "fake" @@ -223,7 +223,11 @@ class FakeCandidate: def fetch_bytes(self) -> bytes: if self._fail: raise SourceAccessError("download refused", url=self.file_url) - return self._payload + if self._payload is not None: + return self._payload + # Default to bytes that really are a PNG of the advertised size, because + # the CLI filters on downloaded bytes and never on source metadata. + return png_bytes(self.width or 1, self.height or 1) class FakeRegistry: @@ -265,18 +269,29 @@ def test_ok_lists_items_without_download(monkeypatch: pytest.MonkeyPatch, capsys 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)] + item = payload["items"][0] + assert item["url"] == "https://example.test/a.png" + assert item["file_url"] == "https://example.test/a.png" + assert item["source"] == "fake" + assert item["format"] == "png" + assert item["width"] == 100 + assert item["height"] == 100 + assert registry.calls == [("fake", "fox", example.candidate_cap(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)]) +def test_no_format_when_every_payload_really_is_another_format( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + registry = FakeRegistry( + [FakeCandidate("fox", "https://example.test/a.jpg", "image/jpeg", 100, 100, payload=gif_bytes())] + ) code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS) assert code == envelope.EXIT_NO_FORMAT assert payload["status"] == envelope.STATUS_NO_FORMAT + assert payload["error"]["details"]["unreadable_candidates"] == 0 + assert payload["error"]["details"]["requested_format"] == "png" def test_no_size_when_only_smaller_images_exist(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: @@ -311,7 +326,7 @@ def test_exact_size_is_accepted(monkeypatch: pytest.MonkeyPatch, capsys: pytest. assert len(payload["items"]) == 1 -def test_resize_without_download_keeps_reported_source_size( +def test_resize_without_download_keeps_actual_size( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: registry = FakeRegistry([FakeCandidate("fox", "https://example.test/a.png", "image/png", 400, 300)]) @@ -325,7 +340,10 @@ def test_resize_without_download_keeps_reported_source_size( 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)]) + expected = png_bytes(100, 100) + registry = FakeRegistry( + [FakeCandidate("fox", "https://example.test/a.png", "image/png", 100, 100, payload=expected)] + ) code, payload = run_main(monkeypatch, capsys, registry, *BASE_ARGS, "--download", "--output-dir", str(tmp_path)) @@ -334,7 +352,8 @@ def test_download_success_writes_files(monkeypatch: pytest.MonkeyPatch, capsys: assert download["ok"] is True target = Path(download["path"]) assert target.parent == tmp_path - assert target.read_bytes() == b"payload" + # The verified bytes are stored verbatim, without a second download. + assert target.read_bytes() == expected def test_partial_download_reports_status(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: @@ -349,6 +368,9 @@ def test_partial_download_reports_status(monkeypatch: pytest.MonkeyPatch, capsys assert code == envelope.EXIT_PARTIAL_DOWNLOAD assert payload["status"] == envelope.STATUS_PARTIAL_DOWNLOAD + # The readable candidate is still reported and downloaded. + assert payload["items"][0]["download"]["ok"] is True + assert payload["error"]["details"]["failed"] == 1 def test_all_downloads_failing_is_access_error(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None: @@ -358,6 +380,7 @@ def test_all_downloads_failing_is_access_error(monkeypatch: pytest.MonkeyPatch, assert code == envelope.EXIT_ACCESS_ERROR assert payload["status"] == envelope.STATUS_ACCESS_ERROR + assert payload["error"]["code"] == "candidate_unreadable" def test_search_access_error_maps_to_access_error(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: @@ -398,6 +421,12 @@ def png_bytes(width: int, height: int, rgba: tuple[int, int, int, int] = (255, 0 ) +def gif_bytes(width: int = 100, height: int = 100) -> bytes: + """Build a payload that really is a GIF, i.e. a format we must never accept.""" + + return b"GIF89a" + struct.pack(" tuple[int, int]: assert payload.startswith(PNG_SIGNATURE) return struct.unpack(">II", payload[16:24]) @@ -528,9 +557,16 @@ def test_resize_downscales_png_stored_inside_ico( assert png_size(inner) == (50, 50) -def test_undecodable_payload_with_resize_fails_the_item( +def test_non_image_payload_is_no_format_not_access_error( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path ) -> None: + """A body that is not a png really is "not the requested format". + + Live sources answer with HTML error pages, licence documents and vector + files; those must produce the topic verdict ``no_format`` (exit 3) instead + of an access error or a partial download. + """ + registry = FakeRegistry( [ FakeCandidate( @@ -543,8 +579,9 @@ def test_undecodable_payload_with_resize_fails_the_item( 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" + assert code == envelope.EXIT_NO_FORMAT + assert document["status"] == envelope.STATUS_NO_FORMAT + assert document["error"]["code"] == "no_format_match" + assert document["error"]["details"]["other_format_candidates"] == 1 + assert document["items"] == [] + assert list(tmp_path.iterdir()) == [] diff --git a/tests/test_images.py b/tests/test_images.py new file mode 100644 index 0000000..24effb9 --- /dev/null +++ b/tests/test_images.py @@ -0,0 +1,654 @@ +"""Tests for byte-based format and size filtering. + +Two layers are covered: + +* :mod:`tools.images` -- the stdlib-only detector and the stdlib-only builders + for the two accepted formats, so fixtures are real images rather than magic + strings; +* the CLI search path -- candidates are filtered on the *downloaded bytes*, so + a source that mislabels a file, claims the wrong size or serves an unreadable + body produces the correct ``ok`` / ``no_format`` / ``no_size`` / + ``access_error`` answer. Every CLI test injects a fake byte-fetch callable, + so nothing here touches the network. +""" + +from __future__ import annotations + +import json +import struct +from typing import Any, Mapping, Sequence + +import pytest + +from tools import envelope, example, images +from tools.errors import SourceAccessError +from tools.images import ImageFormatError, ImageInfo + +# --------------------------------------------------------------------------- +# Real-format builders used as fixtures +# --------------------------------------------------------------------------- + + +def png_payload(width: int, height: int) -> bytes: + """A real PNG built with :func:`tools.images.png_bytes`.""" + + return images.png_bytes(width, height) + + +def ico_payload(width: int, height: int) -> bytes: + """A real ICO (PNG-compressed entry) built with :func:`tools.images.ico_bytes`.""" + + return images.ico_bytes(width, height) + + +def gif_payload(width: int = 100, height: int = 100) -> bytes: + """A payload whose real format is GIF, i.e. one we must never accept.""" + + return b"GIF89a" + struct.pack(" bytes: + """A payload whose real format is JPEG, i.e. another format we must not accept.""" + + return b"\xff\xd8\xff\xe0" + b"\x00" * 12 + + +def svg_payload(width: int = 100, height: int = 100) -> bytes: + """A real SVG document, the most common non-bitmap hit on a live source.""" + + return ( + '\n' + f'\n' + ).encode("utf-8") + + +def tiff_payload() -> bytes: + """A payload whose real format is TIFF.""" + + return b"II*\x00" + b"\x00" * 16 + + +def pdf_payload() -> bytes: + """A payload whose real format is PDF, served by document-friendly sources.""" + + return b"%PDF-1.7\n%\xe2\xe3\xcf\xd3\n" + + +# --------------------------------------------------------------------------- +# tools.images: detection +# --------------------------------------------------------------------------- + + +def test_png_bytes_builds_a_png_with_the_requested_dimensions() -> None: + payload = png_payload(13, 7) + + assert payload.startswith(images.PNG_SIGNATURE) + assert images.detect_format(payload) == images.PNG + assert images.inspect_image(payload) == ImageInfo(format="png", width=13, height=7) + # The IHDR chunk really carries the dimensions, so the bytes are the source + # of truth rather than a helper function agreeing with itself. + assert payload[12:16] == b"IHDR" + assert struct.unpack(">II", payload[16:24]) == (13, 7) + assert images.read_png_dimensions(payload) == (13, 7) + + +def test_ico_bytes_builds_a_directory_that_reports_the_requested_sizes() -> None: + payload = ico_payload(48, 37) + + assert images.detect_format(payload) == images.ICO + assert images.inspect_image(payload) == ImageInfo(format="ico", width=48, height=37) + # One directory entry, and its data range lies inside the payload. + reserved, kind, count = struct.unpack(" None: + payload = ico_payload(256, 256) + + assert payload[6] == 0 and payload[7] == 0 + assert images.read_ico_dimensions(payload) == (256, 256) + + +def test_ico_dimensions_use_the_largest_directory_entry() -> None: + payload = images.ico_bytes(16, 16, frames=[(16, 16), (32, 32), (256, 256)]) + + assert len(images.ico_entries(payload)) == 3 + assert images.read_ico_dimensions(payload) == (256, 256) + assert images.inspect_image(payload).format == "ico" + + +@pytest.mark.parametrize( + ("payload", "expected"), + [ + (png_payload(2, 2), images.PNG), + (ico_payload(2, 2), images.ICO), + (gif_payload(), images.GIF), + (jpeg_payload(), images.JPEG), + (b"BM" + b"\x00" * 20, images.BMP), + (b"RIFF" + b"\x00\x00\x00\x00" + b"WEBP" + b"\x00" * 8, images.WEBP), + (svg_payload(), images.SVG), + (svg_payload().replace(b" None: + assert images.detect_format(payload) == expected + + +@pytest.mark.parametrize("payload", [b"", b"payload", b"not-an-image", b""]) +def test_detect_format_returns_none_for_non_images(payload: bytes) -> None: + assert images.detect_format(payload) is None + + +@pytest.mark.parametrize("payload", [gif_payload(), jpeg_payload(), svg_payload(), tiff_payload(), pdf_payload()]) +def test_inspect_image_rejects_formats_whose_size_is_not_readable(payload: bytes) -> None: + with pytest.raises(ImageFormatError) as excinfo: + images.inspect_image(payload) + + assert excinfo.value.code == "image_unreadable" + # The format is named, so the CLI can report it as another format rather + # than as an access problem. + assert excinfo.value.details["format"] == images.detect_format(payload) + + +@pytest.mark.parametrize( + "payload", + [ + b"not-an-image", + images.PNG_SIGNATURE, + images.PNG_SIGNATURE + struct.pack(">I", 13) + b"IDAT" + b"\x00" * 13, + b"\x00\x00\x01\x00" + struct.pack(" None: + with pytest.raises(ImageFormatError): + images.inspect_image(payload) + + +def test_png_dimensions_reject_a_zero_sized_image() -> None: + payload = images.PNG_SIGNATURE + struct.pack(">I", 13) + b"IHDR" + struct.pack(">IIBBBBB", 0, 0, 8, 6, 0, 0, 0) + + with pytest.raises(ImageFormatError): + images.read_png_dimensions(payload) + + +# --------------------------------------------------------------------------- +# CLI search path: filtering on downloaded bytes +# --------------------------------------------------------------------------- + + +class StubCandidate: + """Candidate whose bytes are supplied by a fake fetch callable.""" + + def __init__( + self, + title: str, + url: str, + *, + payload: bytes | None = None, + error: Exception | None = None, + mime_type: str | None = None, + width: int | None = None, + height: int | None = None, + ) -> None: + self.source_id = "fake" + self.title = title + self.file_url = url + self.page_url = url + ".html" + # Deliberately misleading metadata: filtering must ignore all of this. + self.mime_type = mime_type + self.width = width + self.height = height + self._payload = payload + self._error = error + self.fetch_calls = 0 + + def fetch_bytes(self) -> bytes: + self.fetch_calls += 1 + if self._error is not None: + raise self._error + assert self._payload is not None + return self._payload + + +class StubRegistry: + """Registry stand-in that returns a fixed candidate list.""" + + def __init__(self, candidates: Sequence[StubCandidate]) -> None: + self.candidates = list(candidates) + 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[StubCandidate]: + self.calls.append((source_id, topic, limit)) + return self.candidates[:limit] + + +def run_search( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + candidates: Sequence[StubCandidate], + *args: str, +) -> tuple[int, dict[str, Any], StubRegistry]: + registry = StubRegistry(candidates) + 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 + document = json.loads(lines[0]) + assert isinstance(document, Mapping) + return code, document, registry + + +BASE_ARGS = ("--topic", "red fox", "--format", "png", "--source", "fake") + + +def test_png_candidate_is_kept_with_its_actual_dimensions( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + payload = png_payload(800, 600) + candidate = StubCandidate( + "fox", + "https://example.test/fox.png", + payload=payload, + # The source claims something else entirely; the bytes win. + mime_type="image/jpeg", + width=12, + height=12, + ) + + code, document, registry = run_search( + monkeypatch, capsys, [candidate], *BASE_ARGS, "--width", "640", "--height", "480" + ) + + assert code == envelope.EXIT_OK + assert document["status"] == envelope.STATUS_OK + assert document["error"] is None + item = document["items"][0] + assert item["source"] == "fake" + assert item["url"] == "https://example.test/fox.png" + assert item["format"] == "png" + assert (item["width"], item["height"]) == (800, 600) + assert item["size_bytes"] == len(payload) + assert candidate.fetch_calls == 1 + # The bounded candidate window is requested from the source, not --count. + assert registry.calls == [("fake", "red fox", example.candidate_cap(10))] + + +def test_ico_request_with_only_png_payloads_yields_no_format( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + candidates = [ + StubCandidate("fox", "https://example.test/a.png", payload=png_payload(400, 400), mime_type="image/x-icon"), + StubCandidate("fox2", "https://example.test/b.png", payload=png_payload(500, 500), mime_type="image/ico"), + ] + + code, document, _ = run_search( + monkeypatch, + capsys, + candidates, + "--topic", + "red fox", + "--format", + "ico", + "--source", + "fake", + "--width", + "64", + "--height", + "64", + ) + + assert code == envelope.EXIT_NO_FORMAT + assert document["status"] == envelope.STATUS_NO_FORMAT + assert document["items"] == [] + assert document["error"]["code"] == "no_format_match" + assert document["error"]["details"]["requested_format"] == "ico" + assert document["error"]["details"]["unreadable_candidates"] == 0 + assert document["error"]["details"]["format_matches"] == 0 + + +def test_ico_payload_is_kept_for_an_ico_request( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + candidate = StubCandidate("fox", "https://example.test/fox.ico", payload=ico_payload(64, 48)) + + code, document, _ = run_search( + monkeypatch, + capsys, + [candidate], + "--topic", + "red fox", + "--format", + "ico", + "--source", + "fake", + "--width", + "32", + "--height", + "32", + ) + + assert code == envelope.EXIT_OK + item = document["items"][0] + assert item["format"] == "ico" + assert (item["width"], item["height"]) == (64, 48) + + +def test_all_too_small_payloads_yield_no_size( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + candidates = [ + StubCandidate("small", "https://example.test/a.png", payload=png_payload(100, 100), width=4000, height=4000), + StubCandidate("tall", "https://example.test/b.png", payload=png_payload(900, 100)), + ] + + code, document, _ = run_search( + monkeypatch, capsys, candidates, *BASE_ARGS, "--width", "200", "--height", "200" + ) + + assert code == envelope.EXIT_NO_SIZE + assert document["status"] == envelope.STATUS_NO_SIZE + assert document["items"] == [] + assert document["error"]["code"] == "no_size_match" + assert document["error"]["details"]["format_matches"] == 2 + assert document["error"]["details"]["unreadable_candidates"] == 0 + + +def test_exactly_minimum_size_is_accepted( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + candidate = StubCandidate("edge", "https://example.test/edge.png", payload=png_payload(200, 200)) + + code, document, _ = run_search( + monkeypatch, capsys, [candidate], *BASE_ARGS, "--width", "200", "--height", "200" + ) + + assert code == envelope.EXIT_OK + assert (document["items"][0]["width"], document["items"][0]["height"]) == (200, 200) + + +def test_no_format_and_no_size_and_access_error_stay_distinct( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + format_code, format_document, _ = run_search( + monkeypatch, + capsys, + [StubCandidate("gif", "https://example.test/a.png", payload=gif_payload())], + *BASE_ARGS, + ) + size_code, size_document, _ = run_search( + monkeypatch, + capsys, + [StubCandidate("small", "https://example.test/a.png", payload=png_payload(10, 10))], + *BASE_ARGS, + "--width", + "100", + "--height", + "100", + ) + access_code, access_document, _ = run_search( + monkeypatch, + capsys, + [StubCandidate("broken", "https://example.test/a.png", error=SourceAccessError("refused"))], + *BASE_ARGS, + ) + + assert format_document["status"] == envelope.STATUS_NO_FORMAT + assert size_document["status"] == envelope.STATUS_NO_SIZE + assert access_document["status"] == envelope.STATUS_ACCESS_ERROR + assert len({format_code, size_code, access_code}) == 3 + assert (format_code, size_code, access_code) == ( + envelope.EXIT_NO_FORMAT, + envelope.EXIT_NO_SIZE, + envelope.EXIT_ACCESS_ERROR, + ) + + +def test_fetch_failure_yields_access_error( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + candidates = [ + StubCandidate( + "broken", + "https://example.test/a.png", + error=SourceAccessError("image source returned HTTP status 503", url="https://example.test/a.png"), + ) + ] + + code, document, _ = run_search(monkeypatch, capsys, candidates, *BASE_ARGS) + + assert code == envelope.EXIT_ACCESS_ERROR + assert document["status"] == envelope.STATUS_ACCESS_ERROR + assert document["items"] == [] + assert document["error"]["code"] == "candidate_unreadable" + assert document["error"]["details"]["failures"][0]["url"] == "https://example.test/a.png" + + +def test_non_image_payload_is_no_format_not_access_error( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Bytes that are no image at all are still "not the requested format". + + A live source answers with HTML error pages, licence documents and vector + files; none of them is a png, so the topic verdict is ``no_format`` and the + caller must not see an access error (or a partial download) instead. + """ + + candidate = StubCandidate( + "junk", "https://example.test/a.png", payload=b"not-an-image", mime_type="image/png" + ) + + code, document, _ = run_search(monkeypatch, capsys, [candidate], *BASE_ARGS) + + assert code == envelope.EXIT_NO_FORMAT + assert document["status"] == envelope.STATUS_NO_FORMAT + assert document["items"] == [] + assert document["error"]["code"] == "no_format_match" + details = document["error"]["details"] + assert details["other_format_candidates"] == 1 + assert details["observed_formats"] == {example.UNKNOWN_FORMAT: 1} + assert details["unreadable_candidates"] == 0 + + +def test_search_keeps_png_when_the_window_also_serves_other_formats( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """The real source mixes formats; a working search stays ``ok``. + + Every non-png payload in the bounded window used to be classified as an + unreadable candidate, which demoted a successful search to + ``partial_download`` (exit 6). They are format mismatches now. + """ + + candidates = [ + StubCandidate("gif", "https://example.test/a.gif", payload=gif_payload(), mime_type="image/png"), + StubCandidate("jpeg", "https://example.test/a.jpg", payload=jpeg_payload(), mime_type="image/png"), + StubCandidate("svg", "https://example.test/a.svg", payload=svg_payload(), mime_type="image/png"), + StubCandidate("tiff", "https://example.test/a.tiff", payload=tiff_payload(), mime_type="image/png"), + StubCandidate("pdf", "https://example.test/a.pdf", payload=pdf_payload(), mime_type="image/png"), + StubCandidate("page", "https://example.test/a.html", payload=b"rate limited"), + StubCandidate("fox", "https://example.test/fox.png", payload=png_payload(800, 600)), + ] + + code, document, _ = run_search( + monkeypatch, capsys, candidates, *BASE_ARGS, "--width", "640", "--height", "480" + ) + + assert code == envelope.EXIT_OK + assert document["status"] == envelope.STATUS_OK + assert document["error"] is None + assert [item["url"] for item in document["items"]] == ["https://example.test/fox.png"] + assert (document["items"][0]["width"], document["items"][0]["height"]) == (800, 600) + + +def test_unfetchable_candidate_does_not_demote_a_usable_search( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """A working search stays ``ok`` even when one candidate cannot be fetched.""" + + candidates = [ + StubCandidate( + "missing", + "https://example.test/missing.png", + error=SourceAccessError("image source returned HTTP status 503"), + ), + StubCandidate("fox", "https://example.test/fox.png", payload=png_payload(300, 200)), + ] + + code, document, _ = run_search( + monkeypatch, capsys, candidates, *BASE_ARGS, "--width", "100", "--height", "100" + ) + + assert code == envelope.EXIT_OK + assert document["status"] == envelope.STATUS_OK + assert [item["url"] for item in document["items"]] == ["https://example.test/fox.png"] + + +def test_ico_request_with_only_foreign_formats_yields_no_format_not_access_error( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """REQ-007: ico may be unavailable, and that is ``no_format`` (exit 3).""" + + candidates = [ + StubCandidate("gif", "https://example.test/a.gif", payload=gif_payload()), + StubCandidate("jpeg", "https://example.test/a.jpg", payload=jpeg_payload()), + StubCandidate("svg", "https://example.test/a.svg", payload=svg_payload()), + StubCandidate("tiff", "https://example.test/a.tiff", payload=tiff_payload()), + StubCandidate("page", "https://example.test/a.html", payload=b""), + ] + + code, document, _ = run_search( + monkeypatch, + capsys, + candidates, + "--topic", + "red fox", + "--format", + "ico", + "--source", + "fake", + "--width", + "64", + "--height", + "64", + ) + + assert code == envelope.EXIT_NO_FORMAT + assert document["status"] == envelope.STATUS_NO_FORMAT + details = document["error"]["details"] + assert details["requested_format"] == "ico" + assert details["other_format_candidates"] == 5 + assert details["observed_formats"] == { + "gif": 1, + "jpeg": 1, + "svg": 1, + "tiff": 1, + example.UNKNOWN_FORMAT: 1, + } + assert details["unreadable_candidates"] == 0 + + +def test_no_size_survives_other_formats_in_the_same_window( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """REQ-008: format-matching but too-small images keep their own status.""" + + candidates = [ + StubCandidate("gif", "https://example.test/a.gif", payload=gif_payload()), + StubCandidate("small", "https://example.test/a.png", payload=png_payload(100, 100)), + StubCandidate("svg", "https://example.test/a.svg", payload=svg_payload()), + ] + + code, document, _ = run_search( + monkeypatch, capsys, candidates, *BASE_ARGS, "--width", "200", "--height", "200" + ) + + assert code == envelope.EXIT_NO_SIZE + assert document["status"] == envelope.STATUS_NO_SIZE + details = document["error"]["details"] + assert details["format_matches"] == 1 + assert details["other_format_candidates"] == 2 + assert details["unreadable_candidates"] == 0 + + +def test_corrupt_payload_of_the_requested_format_does_not_hide_a_valid_match( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + corrupt = images.PNG_SIGNATURE + struct.pack(">I", 13) + b"IDAT" + b"\x00" * 13 + candidates = [ + StubCandidate("broken", "https://example.test/broken.png", payload=corrupt), + StubCandidate("fox", "https://example.test/fox.png", payload=png_payload(200, 200)), + ] + + code, document, _ = run_search( + monkeypatch, capsys, candidates, *BASE_ARGS, "--width", "100", "--height", "100" + ) + + assert code == envelope.EXIT_OK + assert document["status"] == envelope.STATUS_OK + assert [item["url"] for item in document["items"]] == ["https://example.test/fox.png"] + + +def test_empty_search_window_is_no_format_not_access_error( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + code, document, _ = run_search(monkeypatch, capsys, [], *BASE_ARGS) + + assert code == envelope.EXIT_NO_FORMAT + assert document["status"] == envelope.STATUS_NO_FORMAT + assert document["error"]["details"]["candidate_count"] == 0 + assert document["error"]["details"]["unreadable_candidates"] == 0 + + +def test_candidate_window_is_bounded_by_the_requested_count( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + candidates = [ + StubCandidate(f"fox{n}", f"https://example.test/{n}.png", payload=png_payload(100, 100)) + for n in range(40) + ] + + code, document, registry = run_search( + monkeypatch, capsys, candidates, *BASE_ARGS, "--count", "3" + ) + + assert code == envelope.EXIT_OK + assert len(document["items"]) == 3 + assert registry.calls == [("fake", "red fox", example.candidate_cap(3))] + assert example.candidate_cap(3) == 12 + # Fetching stops as soon as enough usable images were found. + assert sum(candidate.fetch_calls for candidate in candidates) == 3 + + +def test_filtering_walks_past_rejected_candidates_until_the_count_is_met( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + candidates = [ + StubCandidate("gif", "https://example.test/0.png", payload=gif_payload()), + StubCandidate("tiny", "https://example.test/1.png", payload=png_payload(10, 10)), + StubCandidate("ok1", "https://example.test/2.png", payload=png_payload(120, 120)), + StubCandidate("ok2", "https://example.test/3.png", payload=png_payload(130, 130)), + StubCandidate("never", "https://example.test/4.png", payload=png_payload(140, 140)), + ] + + code, document, _ = run_search( + monkeypatch, capsys, candidates, *BASE_ARGS, "--count", "2", "--width", "100", "--height", "100" + ) + + assert code == envelope.EXIT_OK + assert [item["url"] for item in document["items"]] == [ + "https://example.test/2.png", + "https://example.test/3.png", + ] + assert [(item["width"], item["height"]) for item in document["items"]] == [(120, 120), (130, 130)] + assert candidates[-1].fetch_calls == 0 diff --git a/tools/example.py b/tools/example.py index ef733af..ce667c4 100644 --- a/tools/example.py +++ b/tools/example.py @@ -17,9 +17,11 @@ import re import struct import sys import zlib +from dataclasses import dataclass, field from pathlib import Path from typing import Any, Mapping, Sequence +from . import images from .envelope import ( STATUS_ACCESS_ERROR, STATUS_NO_FORMAT, @@ -41,6 +43,7 @@ from .errors import ( UnknownSourceError, UnsupportedFormatError, ) +from .images import ImageFormatError, ImageInfo from .sources import DEFAULT_LIMIT, SourceCandidate, default_registry PROG = "python -m tools.example" @@ -66,11 +69,21 @@ 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"), -} +#: How many candidates are fetched per wanted image. Nothing a source reports +#: about a hit is trusted, so a candidate is only known to be usable once its +#: bytes have been downloaded and parsed; the source is therefore asked for a +#: window of ``count * SAMPLE_FACTOR`` hits and the window is cut short as soon +#: as ``count`` usable images have been found. +SAMPLE_FACTOR = 4 +#: Hard upper bound on the number of candidates inspected in one invocation. +MAX_CANDIDATE_CAP = MAX_COUNT + + +def candidate_cap(count: int) -> int: + """Return the bounded number of candidates to fetch for ``count`` images.""" + + wanted = max(MIN_COUNT, int(count)) + return min(MAX_CANDIDATE_CAP, wanted * SAMPLE_FACTOR) class _Parser(argparse.ArgumentParser): @@ -312,49 +325,162 @@ def normalize_request(raw: Mapping[str, Any]) -> dict[str, Any]: } -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 _meets_minimum_size(info: ImageInfo, request: Mapping[str, Any]) -> bool: + """True when the *real* pixel size of an image is usable for the request. - -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. + The requested width and height are a minimum, 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): + if wanted_width is not None and info.width < wanted_width: 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: + if wanted_height is not None and info.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.""" +def _failure(candidate: SourceCandidate, fragment: Mapping[str, Any]) -> dict[str, Any]: + """Describe one candidate whose bytes could not be turned into an image.""" - 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 + return { + "source": str(getattr(candidate, "source_id", "") or ""), + "url": str(getattr(candidate, "file_url", "") or ""), + "error": dict(fragment), + } + + +@dataclass(frozen=True) +class KeptImage: + """One candidate whose downloaded bytes really match the request.""" + + candidate: SourceCandidate + info: ImageInfo + payload: bytes + + def to_item(self) -> dict[str, Any]: + """Report ``source``, ``url``, ``format`` and the *actual* dimensions.""" + + source = str(getattr(self.candidate, "source_id", "") or "") + url = str(getattr(self.candidate, "file_url", "") or "") + item: dict[str, Any] = { + "source": source, + "url": url, + # Kept as an alias so existing consumers keep working unchanged. + "file_url": url, + "format": self.info.format, + "width": self.info.width, + "height": self.info.height, + "size_bytes": len(self.payload), + } + for key in ("title", "page_url", "mime_type"): + value = getattr(self.candidate, key, None) + if value is not None: + item[key] = value + return item + + +@dataclass(frozen=True) +class Selection: + """Outcome of inspecting a bounded window of downloaded candidates. + + ``format_rejects`` counts candidates whose real payload is *not* the + requested format -- including payloads that are not a recognised image at + all. ``size_rejects`` counts candidates of the requested format that are + too small. ``failures`` only holds candidates whose bytes could not be + fetched at all, because that is the one situation worth an access error. + """ + + kept: tuple[KeptImage, ...] = () + format_rejects: int = 0 + size_rejects: int = 0 + other_formats: Mapping[str, int] = field(default_factory=dict) + failures: tuple[dict[str, Any], ...] = () + + @property + def inspected(self) -> int: + """Number of candidates whose bytes were read successfully.""" + + return len(self.kept) + self.format_rejects + self.size_rejects + + +#: Bucket name for a downloaded payload that is not a recognised image at all. +UNKNOWN_FORMAT = "unknown" + + +def _count_other_format(other_formats: dict[str, int], detected: str | None) -> None: + """Record which other format a rejected candidate really was.""" + + key = detected or UNKNOWN_FORMAT + other_formats[key] = other_formats.get(key, 0) + 1 + + +def _select(candidates: Sequence[SourceCandidate], request: Mapping[str, Any]) -> Selection: + """Fetch and inspect candidates, keeping only what the real bytes justify. + + Every kept item has been downloaded and parsed: its format comes from the + container signature and its width/height come from the PNG ``IHDR`` or the + ICO directory, never from what the source claimed. A candidate that serves + other-format bytes -- a JPEG, an SVG, a TIFF, an HTML error page, anything + that is not a real ``png``/``ico`` payload -- counts as a *format rejection* + so the caller gets ``no_format`` instead of an access error, and a candidate + of the wanted format that is too small counts as a size rejection. Only a + candidate whose bytes cannot be fetched at all is recorded as a failure. + """ + + wanted_format = str(request["format"]) + limit = int(request["count"]) + kept: list[KeptImage] = [] + format_rejects = 0 + size_rejects = 0 + other_formats: dict[str, int] = {} + failures: list[dict[str, Any]] = [] + + for candidate in candidates: + if len(kept) >= limit: + break + try: + payload = candidate.fetch_bytes() + except ToolkitError as error: + failures.append(_failure(candidate, error.to_dict())) + continue + except OSError as error: + failures.append( + _failure(candidate, error_fragment("fetch_failed", str(error))) + ) + continue + detected = images.detect_format(payload) + if detected != wanted_format: + # Whatever these bytes are (another image format, or no recognisable + # image at all), they are not the requested format: that is a format + # verdict about the topic, never an access problem. + format_rejects += 1 + _count_other_format(other_formats, detected) + continue + try: + info = images.inspect_image(payload) + except ImageFormatError as error: + # The signature claims the requested format but the header is + # corrupt, so the payload itself is unusable rather than merely the + # wrong format. + failures.append(_failure(candidate, error.to_dict())) + continue + if not _meets_minimum_size(info, request): + size_rejects += 1 + continue + kept.append(KeptImage(candidate=candidate, info=info, payload=payload)) + + return Selection( + kept=tuple(kept), + format_rejects=format_rejects, + size_rejects=size_rejects, + other_formats=dict(other_formats), + failures=tuple(failures), + ) def _slug(candidate: SourceCandidate) -> str: @@ -665,12 +791,16 @@ def resize_image(payload: bytes, request: Mapping[str, Any]) -> tuple[bytes, dic return output, fragment -def _download(candidates: Sequence[SourceCandidate], request: Mapping[str, Any]) -> tuple[list[dict[str, Any]], int]: - """Download ``candidates``; return ``(items, failure_count)``. +def _download( + selected: Sequence[KeptImage], request: Mapping[str, Any] +) -> tuple[list[dict[str, Any]], int]: + """Write the already-fetched payloads; 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. + The bytes verified by :func:`_select` are stored as they are (or downscaled + first when ``resize`` is enabled), so an item can never be downloaded twice + or stored in a format that was not verified. A payload that cannot be + downscaled to the requested size is reported as a failed item instead of + being silently stored at the wrong size. """ directory = Path(request["output_dir"]).expanduser() @@ -685,11 +815,11 @@ def _download(candidates: Sequence[SourceCandidate], request: Mapping[str, Any]) 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}" + for index, entry in enumerate(selected, start=1): + record = entry.to_item() + destination = directory / f"{index:03d}-{_slug(entry.candidate)}{suffix}" + payload = entry.payload 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 @@ -756,7 +886,27 @@ def _request_from(argv: Sequence[str]) -> tuple[dict[str, Any], str, Any]: def _execute(argv: Sequence[str]) -> Response: - """Run one request and return the response that must be printed.""" + """Run one request and return the response that must be printed. + + The search path downloads a bounded window of candidates and filters that + window on the parsed bytes, so the answer distinguishes three different + situations through both the status and the exit code: + + * ``ok`` -- at least one image really is the requested format and at least + the requested width and height; + * ``no_format`` -- the source had hits, but none of them really are the + requested format; + * ``no_size`` -- images of the requested format existed, but all of them + are smaller than requested; + * ``access_error`` -- the source could not be searched, or not a single + candidate could be fetched, which is never reported as a "no match". + + A candidate whose downloaded bytes are a different format (a JPEG, an SVG, + an HTML error page, ...) is a format mismatch about the topic, not an access + problem: it feeds ``no_format`` instead of turning a working search into an + error, and a stray unfetchable candidate never demotes a usable answer to + ``partial_download``. + """ request, source, registry = _request_from(argv) builder = ResponseBuilder(source=source, request=request) @@ -764,58 +914,82 @@ def _execute(argv: Sequence[str]) -> Response: if request["dry_run"]: return builder.dry_run() - candidates = registry.search(source, request["topic"], request["count"]) - matching, rejected = _select(candidates, request) + cap = candidate_cap(request["count"]) + candidates = registry.search(source, request["topic"], cap) + selection = _select(candidates, request) + unreadable = list(selection.failures) - if not matching: + if not selection.kept: 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), + "candidates_inspected": selection.inspected, + "format_matches": selection.size_rejects, + "other_format_candidates": selection.format_rejects, + "observed_formats": dict(selection.other_formats), "size_matches": 0, + "unreadable_candidates": len(unreadable), } - if rejected: + if selection.size_rejects: return builder.build( STATUS_NO_SIZE, error=error_fragment( "no_size_match", - "no image is at least the requested width and height", + "the source returned no images at least the requested width and height", details=details, ), ) - return builder.build( - STATUS_NO_FORMAT, + if selection.format_rejects or not unreadable: + # Either the source really had no images in the requested format, or + # it had no hits at all; both answer the topic, not an outage. + 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, + ), + ) + return builder.access_error( error=error_fragment( - "no_format_match", - f"the source returned no {request['format']} images for {request['topic']!r}", - details=details, - ), + "candidate_unreadable", + "none of the candidates could be downloaded", + details={**details, "failures": unreadable}, + ) ) if not request["download"]: - return builder.build(STATUS_OK, items=[c.to_dict() for c in matching]) + # Every reported item was fetched and verified against its real bytes, so + # an unfetchable candidate is simply skipped: it must not demote an + # otherwise usable answer to a partial-download exit code. + return builder.build(STATUS_OK, items=[entry.to_item() for entry in selection.kept]) - items, failures = _download(matching, request) - if failures and failures == len(items): + items, download_failures = _download(selection.kept, request) + failed = download_failures + len(unreadable) + if download_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)}, + details={"failed": download_failures, "total": len(items)}, ), ) - if failures: + if failed: 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)}, + f"{failed} of {len(items) + len(unreadable)} images could not be handled", + details={ + "failed": failed, + "total": len(items) + len(unreadable), + "failures": unreadable, + }, ), ) return builder.build(STATUS_OK, items=items) diff --git a/tools/images.py b/tools/images.py new file mode 100644 index 0000000..5c1f31a --- /dev/null +++ b/tools/images.py @@ -0,0 +1,403 @@ +"""Stdlib-only inspection and construction of the image formats we accept. + +The CLI decides what to keep from the **downloaded bytes** and never from the +metadata a source reports about a file: a search hit that claims to be a +``image/png`` of 800x600 pixels is kept only when the bytes really are a PNG of +at least the requested size. This module is the single place that knows how to +read the real format and the real pixel dimensions out of those bytes. + +It deliberately depends on nothing but the standard library: + +* :func:`detect_format` recognises the container signatures of the two accepted + formats (``png``, ``ico``) plus the formats a real image source commonly + serves instead (``jpeg``, ``gif``, ``bmp``, ``webp``, ``svg``, ``tiff``, + ``pdf``). Every recognised-but-different format, and every payload that is + not a recognised image at all, counts as *another format* rather than an + access problem: "the source has images, but none of them are the format you + asked for" is a different answer from "the source could not be used". +* :func:`read_png_dimensions` reads the ``IHDR`` header of a PNG. +* :func:`read_ico_dimensions` reads the ICONDIR/ICONDIRENTRY directory of an + ICO file (the directory stores the real sizes, including the ``0``-means-256 + encoding). +* :func:`inspect_image` combines both into an :class:`ImageInfo`. +* :func:`png_bytes` and :func:`ico_bytes` *build* the same formats so tests (and + callers) can produce real payloads without Pillow or any other dependency. + +:class:`ImageFormatError` -- a :class:`~tools.errors.ToolkitError` subclass with +a stable ``code`` -- is reserved for bytes that are genuinely unusable: an +unknown container, or a ``png``/``ico`` payload whose header is truncated or +corrupt. A readable payload of another format is never an error, so the CLI +keeps a working search out of the ``access_error`` status. +""" + +from __future__ import annotations + +import struct +import zlib +from dataclasses import dataclass +from typing import Any, Iterable, Sequence + +from .errors import ToolkitError + +__all__ = [ + "BMP", + "GIF", + "ICO", + "ICO_ENTRY_SIZE", + "ICO_HEADER_SIZE", + "ICO_MAX_DIMENSION", + "ImageFormatError", + "ImageInfo", + "JPEG", + "KNOWN_FORMATS", + "PDF", + "PNG", + "PNG_SIGNATURE", + "SVG", + "TIFF", + "WEBP", + "detect_format", + "ico_bytes", + "ico_entries", + "inspect_image", + "png_bytes", + "read_ico_dimensions", + "read_png_dimensions", +] + +#: The two formats the CLI accepts as search requests. +PNG = "png" +ICO = "ico" +#: Formats that are recognised (so they can be reported as "another format") +#: but never accepted by the CLI. +JPEG = "jpeg" +GIF = "gif" +BMP = "bmp" +WEBP = "webp" +#: Vector and document formats a live image source serves instead of a bitmap. +#: Naming them lets a search say *which* other formats the topic offered instead +#: of degrading into an access problem. +SVG = "svg" +TIFF = "tiff" +PDF = "pdf" + +#: Every format :func:`detect_format` can name. +KNOWN_FORMATS: tuple[str, ...] = (PNG, ICO, JPEG, GIF, BMP, WEBP, SVG, TIFF, PDF) + +#: The 8-byte signature every PNG file starts with. +PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n" + +#: ICONDIR: reserved (0), type (1 = icon) and the entry count. +ICO_HEADER_SIZE = 6 +#: One ICONDIRENTRY: 16 bytes of width/height/colours/planes/bpp/size/offset. +ICO_ENTRY_SIZE = 16 +#: A single ICO dimension byte of 0 encodes 256 pixels. +ICO_MAX_DIMENSION = 256 + +#: Magic prefixes of the recognised but unsupported formats. +_JPEG_SIGNATURE = b"\xff\xd8\xff" +_GIF_SIGNATURES = (b"GIF87a", b"GIF89a") +_BMP_SIGNATURE = b"BM" +_RIFF_SIGNATURE = b"RIFF" +_WEBP_SIGNATURE = b"WEBP" +_TIFF_SIGNATURES = (b"II*\x00", b"MM\x00*") +_PDF_SIGNATURE = b"%PDF-" +#: How many leading bytes are searched for an `` None: + super().__init__(message, details=details) + + +@dataclass(frozen=True) +class ImageInfo: + """The real format and real pixel size of one payload.""" + + format: str + width: int + height: int + + def to_dict(self) -> dict[str, Any]: + """JSON-ready projection used in CLI items.""" + + return {"format": self.format, "width": self.width, "height": self.height} + + @property + def area(self) -> int: + return self.width * self.height + + +def _probe(payload: bytes) -> bytes: + return bytes(payload[:_SIGNATURE_PROBE]) + + +def _looks_like_svg(data: bytes) -> bool: + """True when *data* starts with an SVG root element (or its XML prologue).""" + + head = data[:_SVG_PROBE] + if head.startswith(b"\xef\xbb\xbf"): + head = head[3:] + head = head.lstrip() + if head.lower().startswith(b" str | None: + """Return the real container format of *payload*, or ``None``. + + A recognised format other than ``png``/``ico`` is a perfectly good answer: + the source does have images, just not in the requested format. ``None`` + means the bytes are not a recognised image at all -- still "not the requested + format" from the caller's point of view, and never an access problem on its + own. + """ + + if not isinstance(payload, (bytes, bytearray, memoryview)): + return None + data = bytes(payload) + if data.startswith(PNG_SIGNATURE): + return PNG + if data[:4] == b"\x00\x00\x01\x00": + return ICO + if data.startswith(_JPEG_SIGNATURE): + return JPEG + if data.startswith(_GIF_SIGNATURES): + return GIF + if data.startswith(_BMP_SIGNATURE) and len(data) >= 14: + return BMP + if data.startswith(_RIFF_SIGNATURE) and data[8:12] == _WEBP_SIGNATURE: + return WEBP + if data.startswith(_TIFF_SIGNATURES): + return TIFF + if data.startswith(_PDF_SIGNATURE): + return PDF + if _looks_like_svg(data): + return SVG + return None + + +def read_png_dimensions(payload: bytes) -> tuple[int, int]: + """Return ``(width, height)`` from the ``IHDR`` chunk of a PNG payload.""" + + data = bytes(payload) + if not data.startswith(PNG_SIGNATURE): + raise ImageFormatError( + "payload does not start with the PNG signature", + details={"signature": _probe(data).hex()}, + ) + offset = len(PNG_SIGNATURE) + if len(data) < offset + 8: + raise ImageFormatError("PNG header is truncated") + (length,) = struct.unpack(">I", data[offset : offset + 4]) + chunk_type = data[offset + 4 : offset + 8] + if chunk_type != b"IHDR": + raise ImageFormatError( + "PNG payload does not start with an IHDR chunk", + details={"first_chunk": chunk_type.decode("latin-1")}, + ) + if length != 13: + raise ImageFormatError( + "PNG IHDR chunk has an unexpected length", + details={"length": length}, + ) + header = data[offset + 8 : offset + 8 + length] + if len(header) != length: + raise ImageFormatError("PNG IHDR chunk is truncated") + width, height = struct.unpack(">II", header[:8]) + if width < 1 or height < 1: + raise ImageFormatError( + "PNG image declares no pixels", + details={"width": width, "height": height}, + ) + return int(width), int(height) + + +def ico_entries(payload: bytes) -> tuple[tuple[int, int, int, int], ...]: + """Return ``(width, height, size, offset)`` for every ICO directory entry.""" + + data = bytes(payload) + if len(data) < ICO_HEADER_SIZE: + raise ImageFormatError("ICO header is truncated") + reserved, kind, count = struct.unpack(" len(data): + raise ImageFormatError( + "ICO directory entry points outside the payload", + details={"index": index, "size": size, "offset": offset, "bytes": len(data)}, + ) + entries.append((width, height, size, offset)) + return tuple(entries) + + +def read_ico_dimensions(payload: bytes) -> tuple[int, int]: + """Return the dimensions of the largest image stored in an ICO payload. + + The ICONDIRENTRY directory is authoritative: it carries one width/height pair + per stored image, so no decoding of the embedded bitmap or PNG is needed. + """ + + entries = ico_entries(payload) + width, height, _size, _offset = max(entries, key=lambda entry: entry[0] * entry[1]) + return int(width), int(height) + + +def inspect_image(payload: bytes) -> ImageInfo: + """Return the real format and pixel size of *payload*. + + Raises :class:`ImageFormatError` when the bytes are not an image at all, or + when they are not one of the two formats whose dimensions can be read. + """ + + image_format = detect_format(payload) + if image_format is None: + raise ImageFormatError( + "downloaded payload is not a recognised image format", + details={"signature": _probe(bytes(payload)).hex()}, + ) + if image_format == PNG: + width, height = read_png_dimensions(payload) + return ImageInfo(format=PNG, width=width, height=height) + if image_format == ICO: + width, height = read_ico_dimensions(payload) + return ImageInfo(format=ICO, width=width, height=height) + raise ImageFormatError( + f"pixel dimensions are only readable for png and ico, not {image_format}", + details={"format": image_format}, + ) + + +# --------------------------------------------------------------------------- +# Builders: the same formats, produced with the standard library only +# --------------------------------------------------------------------------- + + +def _png_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) + + +def png_bytes( + width: int, + height: int, + *, + rgba: tuple[int, int, int, int] = (255, 255, 255, 255), + rows: Sequence[bytes] | None = None, + level: int = 6, +) -> bytes: + """Build a valid 8-bit RGBA PNG of exactly ``width`` x ``height`` pixels. + + ``rows`` may supply raw (filter-0) RGBA scanlines; otherwise every pixel is + filled with ``rgba``. The result is a real PNG file: a correct signature, + a 13-byte ``IHDR``, a zlib-compressed ``IDAT`` and an empty ``IEND``. + """ + + if isinstance(width, bool) or isinstance(height, bool): + raise ValueError("width and height must be integers") + if width < 1 or height < 1: + raise ValueError("width and height must be >= 1") + if rows is None: + pixel = bytes(rgba) + line = pixel * width + raw = bytearray() + for _ in range(height): + raw.append(0) + raw += line + raw_bytes = bytes(raw) + else: + if len(rows) != height: + raise ValueError("rows must contain exactly height scanlines") + raw = bytearray() + for row in rows: + if len(row) != width * 4: + raise ValueError("each scanline must hold width * 4 bytes") + raw.append(0) + raw += row + raw_bytes = bytes(raw) + header = struct.pack(">IIBBBBB", width, height, 8, 6, 0, 0, 0) + return ( + PNG_SIGNATURE + + _png_chunk(b"IHDR", header) + + _png_chunk(b"IDAT", zlib.compress(raw_bytes, level)) + + _png_chunk(b"IEND", b"") + ) + + +def ico_bytes( + width: int, + height: int, + *, + rgba: tuple[int, int, int, int] = (255, 255, 255, 255), + frames: Iterable[tuple[int, int]] | None = None, +) -> bytes: + """Build an ICO payload whose directory lists the requested sizes. + + By default the ICO stores one embedded PNG of ``width`` x ``height`` pixels + and the directory entry repeats those dimensions (``0`` encodes 256). Pass + ``frames`` to describe several directory entries instead, for example + ``ico_bytes(48, 48, frames=[(16, 16), (48, 48)])``: the dimensions of the + *largest* entry then become the reported size, which is exactly what + :func:`read_ico_dimensions` returns. + """ + + sizes = list(frames) if frames is not None else [(width, height)] + if not sizes: + raise ValueError("an ICO payload needs at least one frame") + payloads: list[bytes] = [] + for frame_width, frame_height in sizes: + if not 1 <= frame_width <= ICO_MAX_DIMENSION: + raise ValueError("ICO widths must be between 1 and 256 pixels") + if not 1 <= frame_height <= ICO_MAX_DIMENSION: + raise ValueError("ICO heights must be between 1 and 256 pixels") + payloads.append(png_bytes(frame_width, frame_height, rgba=rgba)) + header = struct.pack("