US01-01 follow-up: replace donor characterization iteration with YAML ledger + full suite (#46)

tests/characterization/conftest.py

@@ -0,0 +1,107 @@
+"""Shared fixtures for donor characterization tests (US01-01).
+
+Builds a small deterministic synthetic photo library in a temp dir — never the
+real library, never _IGNORE/ contents. Every image has a stable logical fixture
+ID (FX_*) so captured outputs stay keyed to identity, not to paths.
+"""
+import shutil
+import sqlite3
+import sys
+from pathlib import Path
+
+import numpy as np
+import pytest
+from PIL import Image
+
+REPO = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO))
+
+EXIFTOOL = shutil.which("exiftool")
+
+# Stable logical fixture IDs → deterministic generation recipe (seed, size).
+# Structured low-frequency content (blocks + gradient) so phash survives resize.
+FIXTURES = {
+    "fx-blocks-01": {"seed": 1, "size": (800, 600)},
+    "fx-blocks-02": {"seed": 2, "size": (800, 600)},
+    "fx-blocks-03": {"seed": 3, "size": (640, 480)},
+    "fx-blocks-04": {"seed": 4, "size": (800, 600)},
+}
+
+
+def make_image_array(fixture_id: str) -> np.ndarray:
+    spec = FIXTURES[fixture_id]
+    w, h = spec["size"]
+    rng = np.random.default_rng(spec["seed"])
+    base = np.zeros((h, w, 3), dtype=np.uint8)
+    for _ in range(6):
+        x0 = int(rng.integers(0, w - 150))
+        y0 = int(rng.integers(0, h - 150))
+        col = rng.integers(0, 256, 3)
+        base[y0:y0 + 150, x0:x0 + 150] = col
+    grad = np.linspace(0, 120, w, dtype=np.uint8)
+    base[:, :, 0] = np.clip(base[:, :, 0].astype(int) + grad[None, :], 0, 255)
+    return base
+
+
+def write_fixture(fixture_id: str, path: Path, quality: int = 95):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    Image.fromarray(make_image_array(fixture_id)).save(path, quality=quality)
+    return path
+
+
+@pytest.fixture
+def img(tmp_path):
+    """One fx-blocks-01 JPEG in a temp dir."""
+    return write_fixture("fx-blocks-01", tmp_path / "fx-blocks-01.jpg")
+
+
+@pytest.fixture
+def library(tmp_path):
+    """Temp library:
+        root/fx-blocks-01.jpg
+        root/album_a/fx-blocks-02.jpg
+        root/album_a/fx-blocks-02.PNG          (uppercase ext, distinct file)
+        root/album_b/fx-blocks-03.jpg
+        root/_IGNORE/secret.jpg                (must never be discovered)
+        root/album_b/.@__thumb/thumb.jpg       (must never be discovered)
+        root/notes.txt                         (unsupported)
+    """
+    root = tmp_path / "lib"
+    write_fixture("fx-blocks-01", root / "fx-blocks-01.jpg")
+    write_fixture("fx-blocks-02", root / "album_a" / "fx-blocks-02.jpg")
+    img = Image.fromarray(make_image_array("fx-blocks-02"))
+    (root / "album_a").mkdir(parents=True, exist_ok=True)
+    img.save(root / "album_a" / "fx-blocks-02.PNG")
+    write_fixture("fx-blocks-03", root / "album_b" / "fx-blocks-03.jpg")
+    write_fixture("fx-blocks-04", root / "_IGNORE" / "secret.jpg")
+    write_fixture("fx-blocks-04", root / "album_b" / ".@__thumb" / "thumb.jpg")
+    (root / "notes.txt").write_text("not an image")
+    return root
+
+
+@pytest.fixture
+def db(tmp_path):
+    import photo_analyzer as pa
+    conn = pa.get_db(str(tmp_path / "test.db"))
+    yield conn
+    conn.close()
+
+
+def seed_analyzed(conn: sqlite3.Connection, path: str, **overrides):
+    """Insert a row and mark it analyzed with a deterministic result."""
+    import photo_analyzer as pa
+    result = {
+        "description": "A red block pattern.",
+        "tags": ["blocks", "test"],
+        "people_count": 0,
+        "setting": "indoor",
+        "time_of_day": "unknown",
+        "season": "unknown",
+        "mood": "calm",
+        "location_hint": None,
+        "approx_year": None,
+    }
+    result.update(overrides)
+    pa.upsert_pending(conn, path)
+    pa.mark_analyzed(conn, path, result, raw="{}")
+    return result

tests/characterization/test_donor_ledger.py

@@ -0,0 +1,88 @@
+"""Ledger lint (US01-01): every donor-ledger row must carry a real source
+reference, a target location, and either existing characterization test IDs or
+a real pending backlog story."""
+import re
+from pathlib import Path
+
+import yaml
+
+REPO = Path(__file__).resolve().parents[2]
+LEDGER = REPO / "donor_ledger.yaml"
+STORIES = REPO / "delivery_backlog" / "stories"
+TESTS_DIR = Path(__file__).resolve().parent
+
+CLASSIFICATIONS = {"reuse", "extract", "refactor", "replace"}
+REQUIRED_AREAS = {"discovery", "hashing", "imaging", "nsfw", "vision", "exif",
+                  "database", "ui", "configuration", "logging", "cancellation",
+                  "error"}
+STATUSES = {"characterized", "pending"}
+
+
+def load_rows():
+    rows = yaml.safe_load(LEDGER.read_text(encoding="utf-8"))["rows"]
+    assert rows, "empty ledger"
+    return rows
+
+
+def collect_test_ids() -> set:
+    """module::function for every test in this suite."""
+    ids = set()
+    for f in TESTS_DIR.glob("test_*.py"):
+        for m in re.finditer(r"^def (test_\w+)", f.read_text(encoding="utf-8"),
+                             re.MULTILINE):
+            ids.add(f"{f.stem}::{m.group(1)}")
+    return ids
+
+
+def test_rows_have_required_fields_and_unique_ids():
+    rows = load_rows()
+    ids = [r["id"] for r in rows]
+    assert len(ids) == len(set(ids)), "duplicate row ids"
+    for r in rows:
+        for field in ("id", "area", "source", "classification", "rationale",
+                      "target", "status"):
+            assert r.get(field), f"{r.get('id', '?')}: missing {field}"
+        assert r["classification"] in CLASSIFICATIONS, r["id"]
+        assert r["status"] in STATUSES, r["id"]
+        assert len(str(r["rationale"]).strip()) >= 20, \
+            f"{r['id']}: rationale too thin to count as a documented reason"
+
+
+def test_source_references_resolve():
+    for r in load_rows():
+        src = r["source"]
+        f = REPO / src["file"]
+        assert f.is_file(), f"{r['id']}: source file {src['file']} missing"
+        text = f.read_text(encoding="utf-8")
+        for sym in src["symbols"]:
+            assert sym in text, f"{r['id']}: symbol {sym!r} not found in {src['file']}"
+
+
+def test_rows_have_tests_or_pending_story():
+    known_tests = collect_test_ids()
+    for r in load_rows():
+        tests = r.get("tests", [])
+        pending = r.get("pending_story")
+        assert tests or pending, f"{r['id']}: neither tests nor pending_story"
+        for t in tests:
+            assert t in known_tests, f"{r['id']}: unknown test id {t}"
+        if pending:
+            matches = list(STORIES.glob(f"{pending}-*.md"))
+            assert matches, f"{r['id']}: pending_story {pending} has no story file"
+        if r["status"] == "characterized":
+            assert tests, f"{r['id']}: characterized rows need test ids"
+
+
+def test_all_required_areas_covered():
+    covered = {r["area"] for r in load_rows()}
+    assert REQUIRED_AREAS <= covered, f"uncovered areas: {REQUIRED_AREAS - covered}"
+    assert covered <= REQUIRED_AREAS, f"unknown areas: {covered - REQUIRED_AREAS}"
+
+
+def test_no_legacy_file_moved():
+    # US01-01 explicitly forbids moving/archiving donors; the ledger's source
+    # files must all still exist at their original locations.
+    for donor in ("photo_analyzer.py", "nsfwtag/scoring.py", "nsfwtag/exif.py",
+                  "nsfwtag/server.py", "webapp/query.py", "webapp/runner.py",
+                  "webapp/server.py"):
+        assert (REPO / donor).is_file(), f"donor moved: {donor}"

tests/characterization/test_donors.py

@@ -1,295 +0,0 @@
-from __future__ import annotations
-
-import base64
-import importlib
-import json
-import logging
-import os
-import subprocess
-import sys
-import tempfile
-import unittest
-from pathlib import Path
-from types import SimpleNamespace
-from unittest import mock
-
-from PIL import Image
-
-
-REPO = Path(__file__).resolve().parents[2]
-FIXTURES = REPO / "tests" / "fixtures" / "characterization"
-
-
-class DonorCharacterizationTests(unittest.TestCase):
-    """Golden contracts for donor behavior; these are not target-architecture tests."""
-
-    @classmethod
-    def setUpClass(cls):
-        cls.import_dir = tempfile.TemporaryDirectory(prefix="donor-import-")
-        old_cwd = Path.cwd()
-        sys.path.insert(0, str(REPO))
-        try:
-            os.chdir(cls.import_dir.name)
-            cls.analyzer = importlib.import_module("photo_analyzer")
-            cls.nsfw_scoring = importlib.import_module("nsfwtag.scoring")
-            cls.nsfw_exif = importlib.import_module("nsfwtag.exif")
-            cls.nsfw_webapp = importlib.import_module("nsfwtag.webapp")
-            cls.web_query = importlib.import_module("webapp.query")
-        finally:
-            os.chdir(old_cwd)
-        cls.manifest = json.loads((FIXTURES / "manifest.json").read_text())["fixtures"]
-        cls.expected = json.loads((FIXTURES / "expected_outputs.json").read_text())["outputs"]
-
-    @classmethod
-    def tearDownClass(cls):
-        for handler in list(logging.getLogger().handlers) + list(
-            logging.getLogger("history").handlers
-        ):
-            handler.close()
-        cls.import_dir.cleanup()
-
-    def setUp(self):
-        self.tmp = tempfile.TemporaryDirectory(prefix="donor-fixtures-")
-        self.root = Path(self.tmp.name)
-
-    def tearDown(self):
-        self.analyzer._stop.clear()
-        self.tmp.cleanup()
-
-    def _make_discovery_library(self) -> None:
-        recipes = {item["id"]: item for item in self.manifest}
-        colors = {
-            "DISC-ROOT-JPEG": (10, 20, 30),
-            "DISC-UPPER-JPEG": (40, 50, 60),
-            "DISC-NESTED-WEBP": (70, 80, 90),
-            "DISC-TIFF": (100, 110, 120),
-        }
-        sizes = {
-            "DISC-ROOT-JPEG": (12, 8),
-            "DISC-UPPER-JPEG": (8, 12),
-            "DISC-NESTED-WEBP": (9, 9),
-            "DISC-TIFF": (16, 4),
-        }
-        for fixture_id, color in colors.items():
-            path = self.root / recipes[fixture_id]["relative_path"]
-            path.parent.mkdir(parents=True, exist_ok=True)
-            Image.new("RGB", sizes[fixture_id], color).save(path)
-        unsupported = self.root / recipes["DISC-UNSUPPORTED"]["relative_path"]
-        unsupported.write_text("not-an-image", encoding="utf-8")
-
-    def _relative(self, paths) -> list[str]:
-        return [str(Path(path).relative_to(self.root)) for path in paths]
-
-    def test_discovery_outputs(self):
-        self._make_discovery_library()
-
-        analyzer_paths = self.analyzer.discover_photos(self.root)
-        nsfw_recursive = self.nsfw_scoring.discover_images(self.root, recursive=True)
-        nsfw_shallow = self.nsfw_scoring.discover_images(self.root, recursive=False)
-
-        self.assertEqual(
-            self._relative(analyzer_paths), self.expected["analyzer_recursive_discovery"]
-        )
-        self.assertEqual(self._relative(nsfw_recursive), self.expected["nsfw_recursive_discovery"])
-        self.assertEqual(self._relative(nsfw_shallow), self.expected["nsfw_shallow_discovery"])
-
-    def test_image_preparation_and_hash_outputs(self):
-        raw_path = self.root / "hash.bin"
-        raw_path.write_bytes(b"photo-pipeline-donor")
-        self.assertEqual(self.analyzer._sha1_file(raw_path), self.expected["hash_bytes_sha1"])
-
-        gradient = Image.new("RGB", (40, 20))
-        gradient.putdata(
-            [(x * 5 % 256, y * 11 % 256, (x + y) * 7 % 256) for y in range(20) for x in range(40)]
-        )
-        image_path = self.root / "gradient.png"
-        gradient.save(image_path)
-        old_edge = self.analyzer.MAX_LONG_EDGE
-        try:
-            self.analyzer.MAX_LONG_EDGE = 16
-            encoded, mime = self.analyzer.prepare_image(image_path)
-        finally:
-            self.analyzer.MAX_LONG_EDGE = old_edge
-        prepared_path = self.root / "prepared.jpg"
-        prepared_path.write_bytes(base64.b64decode(encoded))
-        with Image.open(prepared_path) as prepared:
-            self.assertEqual(prepared.size, (16, 8))
-            self.assertEqual(prepared.mode, "RGB")
-        self.assertEqual(mime, "image/jpeg")
-        phash = self.analyzer._phash_image(image_path)
-        self.assertRegex(phash or "", r"^[0-9a-f]{16}$")
-        self.assertEqual(phash, self.analyzer._phash_image(image_path))
-
-    def test_database_status_and_fts_outputs(self):
-        db_path = self.root / "characterization.sqlite"
-        conn = self.analyzer.get_db(str(db_path))
-        self.addCleanup(conn.close)
-        photo_path = str(self.root / "lake.jpg")
-        self.analyzer.upsert_pending(conn, photo_path)
-        result = {
-            "description": "Three adults walk beside a lake.",
-            "tags": ["people", "lake", "summer"],
-            "people_count": 3,
-            "setting": "outdoor",
-            "time_of_day": "afternoon",
-            "season": "summer",
-            "mood": "relaxed",
-            "location_hint": "Como, Italy",
-            "approx_year": 2021,
-        }
-        self.analyzer.mark_analyzed(conn, photo_path, result, json.dumps(result))
-        row = conn.execute("SELECT * FROM photos WHERE path = ?", (photo_path,)).fetchone()
-        self.assertEqual(row["status"], "analyzed")
-        self.assertEqual(json.loads(row["tags"]), result["tags"])
-        found = self.web_query.search(conn, q="lake")
-        self.assertEqual(found["total"], 1)
-        self.assertEqual(found["rows"][0]["path"], photo_path)
-        self.analyzer.mark_exif_written(conn, photo_path)
-        self.assertEqual(conn.execute("SELECT status FROM photos").fetchone()[0], "exif_written")
-
-    def test_caption_and_variant_outputs(self):
-        result = {
-            "description": "Three adults walk beside a lake.",
-            "tags": ["people", "lake", "summer"],
-            "mood": "relaxed",
-            "location_hint": "Como, Italy",
-            "approx_year": 2021,
-        }
-        self.assertEqual(
-            self.analyzer.build_exif_caption_from_result(result), self.expected["caption"]
-        )
-        self.assertEqual(
-            self.analyzer._strip_variant_markers("IMG_0001-bearbeitet (1920x1080)"),
-            self.expected["variant_base"],
-        )
-
-    def test_vision_dry_run_output(self):
-        result, raw, tokens = self.analyzer.analyze_image(
-            None, self.root / "fixture.jpg", dry_run=True
-        )
-        self.assertEqual(result["tags"], self.expected["dry_run_tags"])
-        self.assertEqual(json.loads(raw), result)
-        self.assertEqual(tokens, {"prompt": 0, "completion": 0, "total": 0})
-
-    def test_nsfw_cache_output(self):
-        cache_path = self.root / "scores.csv"
-        values = self.expected["nsfw_cache"]
-        self.nsfw_scoring._save_cache(cache_path, values)
-        self.assertEqual(self.nsfw_scoring.load_cache(cache_path), values)
-        self.assertEqual(
-            cache_path.read_text(encoding="utf-8").splitlines(),
-            ["path,nsfw_score", "a.jpg,0.1250", "b.jpg,0.9876"],
-        )
-
-    def test_review_html_output(self):
-        logical = self.root / "album & one" / "<portrait>.jpg"
-        page = self.nsfw_webapp.render_page([(logical, 0.8123), (logical, 0.7)], 0.6)
-        self.assertIn("&lt;portrait&gt;.jpg", page)
-        self.assertIn("album &amp; one", page)
-        self.assertIn(f'data-score="{0.8123:.4f}"', page)
-        self.assertIn(self.expected["ui_total"], page)
-        self.assertIn(self.expected["ui_threshold"], page)
-        self.assertEqual(page.count('class="card"'), 1)
-
-    def test_cli_entry_point_help(self):
-        env = os.environ.copy()
-        env["PYTHONPATH"] = str(REPO)
-        commands = [
-            ([sys.executable, str(REPO / "photo_analyzer.py"), "--help"], "--group-variants"),
-            ([sys.executable, str(REPO / "nsfw_tag.py"), "--help"], "--review-min"),
-            ([sys.executable, "-m", "nsfwtag", "--help"], "--threshold"),
-        ]
-        for command, marker in commands:
-            with self.subTest(command=command):
-                result = subprocess.run(
-                    command,
-                    cwd=self.root,
-                    env=env,
-                    capture_output=True,
-                    text=True,
-                    timeout=30,
-                )
-                self.assertEqual(result.returncode, 0, result.stderr)
-                self.assertIn(marker, result.stdout)
-
-    def test_exif_command_contracts(self):
-        calls = []
-
-        def fake_run(command, **kwargs):
-            calls.append(command)
-            if "-json" in command:
-                return subprocess.CompletedProcess(
-                    command,
-                    0,
-                    stdout=json.dumps(
-                        [{"ImageDescription": "User caption", "Keywords": ["family"]}]
-                    ),
-                    stderr="",
-                )
-            return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
-
-        with mock.patch.object(self.analyzer.subprocess, "run", side_effect=fake_run):
-            self.assertTrue(
-                self.analyzer.write_exif("fixture.jpg", "AI caption", ["family", "lake"])
-            )
-        write_command = calls[-1]
-        self.assertIn("-overwrite_original", write_command)
-        self.assertIn("-ImageDescription=AI caption | User caption", write_command)
-        self.assertEqual(write_command.count("-Keywords=family"), 1)
-        self.assertEqual(write_command.count("-Keywords=lake"), 1)
-
-        with mock.patch.object(
-            self.nsfw_exif.subprocess,
-            "run",
-            return_value=subprocess.CompletedProcess([], 0, stdout="", stderr=""),
-        ) as run:
-            self.assertTrue(self.nsfw_exif.write_keyword("fixture.jpg"))
-            command = run.call_args.args[0]
-        self.assertIn("-Keywords-=nsfw", command)
-        self.assertIn("-Keywords+=nsfw", command)
-        self.assertIn("-Subject-=nsfw", command)
-        self.assertIn("-Subject+=nsfw", command)
-
-    def test_error_and_configuration_fallbacks(self):
-        with mock.patch.dict(os.environ, {"MAX_WORKERS": "not-an-int"}):
-            self.assertEqual(self.analyzer._env_int("MAX_WORKERS", 3), 3)
-        invalid = subprocess.CompletedProcess([], 0, stdout="not-json", stderr="")
-        with mock.patch.object(self.nsfw_exif.subprocess, "run", return_value=invalid):
-            self.assertEqual(self.nsfw_exif._read_keywords(["fixture.jpg"]), {})
-
-    def test_history_log_shape(self):
-        sink = mock.Mock()
-        result = {
-            "description": "Fixture description",
-            "tags": ["fixture"],
-            "mood": "calm",
-            "setting": "indoor",
-            "people_count": 0,
-            "location_hint": None,
-            "approx_year": None,
-        }
-        with mock.patch.object(self.analyzer.history_log, "info", sink):
-            self.analyzer.log_history(
-                "fixture.jpg", "analyzed", result=result, tokens={"total": 12}
-            )
-        entry = json.loads(sink.call_args.args[0])
-        self.assertEqual(entry["path"], "fixture.jpg")
-        self.assertEqual(entry["status"], "analyzed")
-        self.assertEqual(entry["tokens_total"], 12)
-        self.assertEqual(entry["tags"], ["fixture"])
-        self.assertRegex(entry["ts"], r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}$")
-
-    def test_cooperative_cancellation_contract(self):
-        self.analyzer._stop.clear()
-        fake_stderr = SimpleNamespace(write=mock.Mock(), flush=mock.Mock())
-        with (
-            mock.patch.object(self.analyzer.sys, "__stderr__", fake_stderr),
-            mock.patch.object(self.analyzer.log, "warning"),
-        ):
-            self.analyzer._handle_sigint(None, None)
-        self.assertTrue(self.analyzer._stop.is_set())
-        fake_stderr.write.assert_called_once_with("\a")
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/characterization/test_ledger.py

@@ -1,62 +0,0 @@
-from __future__ import annotations
-
-import ast
-import json
-import unittest
-from pathlib import Path
-
-REPO = Path(__file__).resolve().parents[2]
-LEDGER = REPO / "donor_ledger" / "ledger.json"
-TRACEABILITY = REPO / "tests" / "characterization" / "test_traceability.json"
-DONOR_TESTS = REPO / "tests" / "characterization" / "test_donors.py"
-
-
-def source_symbols(path: Path) -> set[str]:
-    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
-    names = set()
-    for node in ast.walk(tree):
-        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
-            names.add(node.name)
-        elif isinstance(node, (ast.Assign, ast.AnnAssign)):
-            targets = node.targets if isinstance(node, ast.Assign) else [node.target]
-            names.update(t.id for t in targets if isinstance(t, ast.Name))
-        elif isinstance(node, ast.ImportFrom):
-            names.update(alias.asname or alias.name for alias in node.names)
-    return names
-
-
-class DonorLedgerLintTests(unittest.TestCase):
-    def test_ledger_is_complete_and_resolvable(self):
-        ledger = json.loads(LEDGER.read_text(encoding="utf-8"))
-        traceability = json.loads(TRACEABILITY.read_text(encoding="utf-8"))
-        entries = ledger["entries"]
-        donor_test_symbols = source_symbols(DONOR_TESTS)
-
-        self.assertEqual(ledger["story"], "US01-01")
-        self.assertEqual(len({entry["id"] for entry in entries}), len(entries))
-        self.assertEqual(set(ledger["required_areas"]), {entry["area"] for entry in entries})
-        self.assertFalse((REPO / "legacy_cli_archive").exists())
-
-        allowed = set(ledger["classifications"])
-        for entry in entries:
-            with self.subTest(entry=entry["id"]):
-                self.assertIn(entry["classification"], allowed)
-                self.assertTrue(entry["rationale"].strip())
-                self.assertTrue(entry["target"].strip())
-                self.assertTrue(entry["test_ids"])
-                self.assertEqual(entry["migration_status"], "inventoried")
-                for test_id in entry["test_ids"]:
-                    self.assertIn(test_id, traceability)
-                    class_name, method_name = traceability[test_id].split(".", 1)
-                    self.assertEqual(class_name, "DonorCharacterizationTests")
-                    self.assertIn(method_name, donor_test_symbols)
-                for source in entry["sources"]:
-                    file_name, separator, symbol = source.partition("::")
-                    source_path = REPO / file_name
-                    self.assertTrue(source_path.is_file(), source)
-                    if separator:
-                        self.assertIn(symbol, source_symbols(source_path), source)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/characterization/test_nsfwtag.py

@@ -0,0 +1,95 @@
+"""Characterize nsfwtag donors: discovery, score cache, EXIF safety keywords.
+
+score_images' model inference is NOT run here (needs the ~350 MB local model);
+its cache-reuse path is characterized instead — the model is only invoked for
+paths missing from the cache.
+"""
+import subprocess
+
+import pytest
+
+import nsfwtag.exif as nexif
+import nsfwtag.scoring as scoring
+from nsfwtag import EXTS, KEYWORD
+from conftest import EXIFTOOL, write_fixture
+
+needs_exiftool = pytest.mark.skipif(not EXIFTOOL, reason="exiftool not on PATH")
+
+
+def test_exts_contract_differs_from_photo_analyzer():
+    import photo_analyzer as pa
+    assert EXTS == {".jpg", ".jpeg", ".png", ".heic", ".heif", ".webp"}
+    assert pa.SUPPORTED_EXTENSIONS - EXTS == {".tiff", ".tif"}, \
+        "known donor divergence: nsfwtag skips TIFF"
+
+
+def test_discover_images_flat_recursive_limit(tmp_path):
+    a = write_fixture("fx-blocks-01", tmp_path / "a.jpg")
+    b = write_fixture("fx-blocks-02", tmp_path / "sub" / "b.jpg")
+    (tmp_path / "c.txt").write_text("x")
+    assert scoring.discover_images(tmp_path) == [a], "flat by default"
+    assert scoring.discover_images(tmp_path, recursive=True) == [a, b]
+    assert scoring.discover_images(tmp_path, recursive=True, limit=1) == [a]
+
+
+def test_discover_images_dedupes_symlinked_paths(tmp_path):
+    a = write_fixture("fx-blocks-01", tmp_path / "a.jpg")
+    (tmp_path / "link.jpg").symlink_to(a)
+    found = scoring.discover_images(tmp_path)
+    assert len(found) == 1, "same resolved file listed once"
+
+
+def test_score_cache_roundtrip_and_tolerance(tmp_path):
+    csv = tmp_path / "scores.csv"
+    scoring._save_cache(csv, {"/a.jpg": 0.91234, "/b.jpg": 0.1})
+    cache = scoring.load_cache(csv)
+    assert cache == {"/a.jpg": 0.9123, "/b.jpg": 0.1}, "4-decimal persistence"
+    csv.write_text("path,nsfw_score\n/ok.jpg,0.5\n/bad.jpg,not-a-float\n")
+    assert scoring.load_cache(csv) == {"/ok.jpg": 0.5}, "bad rows dropped silently"
+    assert scoring.load_cache(tmp_path / "missing.csv") == {}
+
+
+def test_score_images_cache_hit_skips_model(tmp_path):
+    img = write_fixture("fx-blocks-01", tmp_path / "a.jpg")
+    csv = tmp_path / "scores.csv"
+    scoring._save_cache(csv, {str(img): 0.42})
+    scored, errors = scoring.score_images([img], csv)  # would download a model on miss
+    assert scored == [(img, 0.42)]
+    assert errors == 0
+
+
+@needs_exiftool
+def test_write_remove_keyword_idempotent(img):
+    assert nexif.write_keyword(str(img))
+    assert nexif.write_keyword(str(img)), "second write is a no-op, not a dup"
+    r = subprocess.run(["exiftool", "-j", "-Keywords", "-Subject", str(img)],
+                       capture_output=True, text=True)
+    assert r.stdout.count(KEYWORD) == 2, "once in Keywords, once in Subject"
+    assert nexif.read_tagged([str(img)]) == {str(img)}
+    assert nexif.remove_keyword(str(img))
+    assert nexif.read_tagged([str(img)]) == set()
+
+
+@needs_exiftool
+def test_read_marks_nsfw_wins_over_sfw(img, tmp_path):
+    safe = write_fixture("fx-blocks-02", tmp_path / "safe.jpg")
+    subprocess.run(["exiftool", "-m", "-overwrite_original",
+                    "-Keywords=nsfw", "-Keywords=sfw", str(img)],
+                   check=True, capture_output=True)
+    subprocess.run(["exiftool", "-m", "-overwrite_original",
+                    "-Keywords=sfw", str(safe)], check=True, capture_output=True)
+    marks = nexif.read_marks([str(img), str(safe)])
+    assert marks["nsfw"] == {str(img)}, "both keywords → nsfw wins (never reads safe)"
+    assert marks["sfw"] == {str(safe)}
+
+
+@needs_exiftool
+def test_read_exif_filters_noise_and_prepends_map(img):
+    subprocess.run(["exiftool", "-m", "-overwrite_original",
+                    "-GPSLatitude=48.8584", "-GPSLatitudeRef=N",
+                    "-GPSLongitude=2.2945", "-GPSLongitudeRef=E", str(img)],
+                   check=True, capture_output=True)
+    rec = nexif.read_exif(str(img))
+    assert list(rec)[0] == "Map"
+    assert rec["Map"] == "https://www.google.com/maps?q=48.858400,2.294500"
+    assert "SourceFile" not in rec and "ExifToolVersion" not in rec

tests/characterization/test_pa_album.py

@@ -0,0 +1,28 @@
+"""Characterize album labeling (donor: album_label, fit_label; webapp.query.album_of
+mirrors album_label — asserted identical here)."""
+from pathlib import Path
+
+import photo_analyzer as pa
+from webapp import query
+
+
+def test_album_label_leaf_folder_relative():
+    lib = Path("/lib")
+    assert pa.album_label(Path("/lib/Urlaub/Rom/a.jpg"), lib) == "Urlaub/Rom"
+    assert pa.album_label(Path("/lib/Urlaub/Venedig/a.jpg"), lib) == "Urlaub/Venedig"
+    assert pa.album_label(Path("/lib/a.jpg"), lib) == "(root)"
+    assert pa.album_label(Path("/elsewhere/x/a.jpg"), lib) == "x", \
+        "outside library → bare parent name"
+    assert pa.album_label(Path("/lib/x/a.jpg"), None) == "x"
+
+
+def test_webapp_album_of_mirrors_album_label():
+    lib = Path("/lib")
+    for p in ("/lib/Urlaub/Rom/a.jpg", "/lib/a.jpg", "/other/x/a.jpg"):
+        assert query.album_of(p, lib) == pa.album_label(Path(p), lib), p
+
+
+def test_fit_label_left_truncates():
+    assert pa.fit_label("short", 10) == "short"
+    assert pa.fit_label("Urlaub/Somewhere/Rom", 8) == "…ere/Rom"
+    assert len(pa.fit_label("Urlaub/Somewhere/Rom", 8)) == 8

tests/characterization/test_pa_config.py

@@ -0,0 +1,46 @@
+"""Characterize configuration behavior (donor: load_env_file, _env_str,
+_env_int, _env_bool)."""
+import os
+
+import photo_analyzer as pa
+
+
+def test_load_env_file_parsing(tmp_path, monkeypatch):
+    env = tmp_path / pa.ENV_FILE
+    env.write_text(
+        "# comment\n"
+        "PLAIN=value\n"
+        "export EXPORTED=yes\n"
+        'QUOTED="hash # kept"\n'
+        "INLINE=val # comment stripped\n"
+        "TOKEN=pass#word\n"
+        "PRESET=file-loses\n"
+        "NOEQUALS\n"
+    )
+    monkeypatch.chdir(tmp_path)
+    for k in ("PLAIN", "EXPORTED", "QUOTED", "INLINE", "TOKEN"):
+        monkeypatch.delenv(k, raising=False)
+    monkeypatch.setenv("PRESET", "shell-wins")
+
+    used = pa.load_env_file()
+    assert used == env
+    assert os.environ["PLAIN"] == "value"
+    assert os.environ["EXPORTED"] == "yes", "'export ' prefix accepted"
+    assert os.environ["QUOTED"] == "hash # kept", "quoted # not a comment"
+    assert os.environ["INLINE"] == "val", "inline comment stripped"
+    assert os.environ["TOKEN"] == "pass#word", "mid-token # kept"
+    assert os.environ["PRESET"] == "shell-wins", "shell env always wins"
+
+
+def test_env_helpers(monkeypatch):
+    monkeypatch.setenv("S", "")
+    assert pa._env_str("S", "fb") == "fb", "empty string falls back"
+    monkeypatch.setenv("I", "not-int")
+    assert pa._env_int("I", 7) == 7, "bad int falls back with warning"
+    monkeypatch.setenv("I", "42")
+    assert pa._env_int("I", 7) == 42
+    for truthy in ("1", "true", "YES", "On"):
+        monkeypatch.setenv("B", truthy)
+        assert pa._env_bool("B") is True
+    monkeypatch.setenv("B", "0")
+    assert pa._env_bool("B") is False

tests/characterization/test_pa_db.py

@@ -0,0 +1,68 @@
+"""Characterize DB layer (donor: get_db, _migrate_schema, status transitions,
+get_pending, purge_excluded, prune_missing, FTS sync)."""
+import photo_analyzer as pa
+from conftest import seed_analyzed, write_fixture
+
+
+def test_schema_and_migration_idempotent(tmp_path):
+    db_path = str(tmp_path / "x.db")
+    conn = pa.get_db(db_path)
+    conn.close()
+    conn = pa.get_db(db_path)  # re-open runs migration again — must not fail
+    cols = {r["name"] for r in conn.execute("PRAGMA table_info(photos)")}
+    assert {"path", "status", "phash", "file_sha1", "dup_of", "description",
+            "tags", "raw_response", "analyzed_at", "exif_written_at"} <= cols
+    conn.close()
+
+
+def test_status_lifecycle_pending_analyzed_exif_written(db):
+    seed_analyzed(db, "/x/a.jpg")
+    assert db.execute("SELECT status FROM photos").fetchone()["status"] == "analyzed"
+    pa.mark_exif_written(db, "/x/a.jpg")
+    assert db.execute("SELECT status FROM photos").fetchone()["status"] == "exif_written"
+
+
+def test_mark_error_and_retry_via_get_pending(db):
+    pa.upsert_pending(db, "/x/a.jpg")
+    pa.mark_error(db, "/x/a.jpg", "boom")
+    row = db.execute("SELECT status, error_message FROM photos").fetchone()
+    assert (row["status"], row["error_message"]) == ("error", "boom")
+    # error rows are retried on the next run
+    assert pa.get_pending(db, reanalyze=False) == ["/x/a.jpg"]
+
+
+def test_mark_analyzed_clears_error(db):
+    pa.upsert_pending(db, "/x/a.jpg")
+    pa.mark_error(db, "/x/a.jpg", "boom")
+    seed_analyzed(db, "/x/a.jpg")
+    assert db.execute("SELECT error_message FROM photos").fetchone()["error_message"] is None
+
+
+def test_upsert_pending_never_downgrades(db):
+    seed_analyzed(db, "/x/a.jpg")
+    pa.upsert_pending(db, "/x/a.jpg")  # INSERT OR IGNORE — no reset to pending
+    assert db.execute("SELECT status FROM photos").fetchone()["status"] == "analyzed"
+
+
+def test_fts_kept_in_sync_by_triggers(db):
+    seed_analyzed(db, "/x/beach.jpg", description="A sunny beach with palm trees.")
+    hit = db.execute(
+        "SELECT path FROM photos_fts WHERE photos_fts MATCH 'beach'").fetchall()
+    assert [r["path"] for r in hit] == ["/x/beach.jpg"]
+
+
+def test_purge_excluded_removes_ignore_and_thumb_rows(db):
+    for p in ("/lib/_IGNORE/a.jpg", "/lib/x/.@__thumb/b.jpg", "/lib/keep.jpg"):
+        pa.upsert_pending(db, p)
+    assert pa.purge_excluded(db) == 2
+    assert [r["path"] for r in db.execute("SELECT path FROM photos")] == ["/lib/keep.jpg"]
+
+
+def test_prune_missing_guards_unmounted_library(db, tmp_path):
+    pa.upsert_pending(db, "/gone/a.jpg")
+    # library root missing → refuses to prune (protects against unmounted volume)
+    assert pa.prune_missing(db, tmp_path / "not-there") == 0
+    assert db.execute("SELECT COUNT(*) FROM photos").fetchone()[0] == 1
+    # existing root → stale row is pruned
+    assert pa.prune_missing(db, tmp_path) == 1
+    assert db.execute("SELECT COUNT(*) FROM photos").fetchone()[0] == 0

tests/characterization/test_pa_discovery.py

@@ -0,0 +1,27 @@
+"""Characterize photo_analyzer discovery + path policy (donor: discover_photos)."""
+import photo_analyzer as pa
+
+
+def test_discover_excludes_ignore_and_thumbs(library):
+    found = pa.discover_photos(library)
+    names = [p.name for p in found]
+    assert "secret.jpg" not in names, "_IGNORE/ content must never be discovered"
+    assert "thumb.jpg" not in names, ".@__thumb/ content must never be discovered"
+    assert "notes.txt" not in names
+
+
+def test_discover_finds_supported_including_uppercase(library):
+    names = sorted(p.name for p in pa.discover_photos(library))
+    assert names == ["fx-blocks-01.jpg", "fx-blocks-02.PNG",
+                     "fx-blocks-02.jpg", "fx-blocks-03.jpg"]
+
+
+def test_discover_sorted_and_deduplicated(library):
+    found = pa.discover_photos(library)
+    assert found == sorted(set(found))
+
+
+def test_supported_extensions_contract():
+    # Donor contract: these exact extensions (nsfwtag EXTS differs — no .tiff/.tif).
+    assert pa.SUPPORTED_EXTENSIONS == {
+        ".jpg", ".jpeg", ".png", ".webp", ".heic", ".heif", ".tiff", ".tif"}

tests/characterization/test_pa_exif.py

@@ -0,0 +1,87 @@
+"""Characterize EXIF behavior (donor: build_exif_caption_from_result,
+read_existing_exif, write_exif merge/idempotency, _rec_has_nsfw,
+filter_nsfw_tagged). Uses the real exiftool on synthetic temp files."""
+import subprocess
+
+import pytest
+
+import photo_analyzer as pa
+from conftest import EXIFTOOL
+
+needs_exiftool = pytest.mark.skipif(not EXIFTOOL, reason="exiftool not on PATH")
+
+RESULT = {
+    "description": "A dog on a beach.",
+    "tags": ["dog", "beach"],
+    "mood": "joyful",
+    "location_hint": "Rome, Italy",
+    "approx_year": 2015,
+}
+
+
+def test_caption_golden_full():
+    assert pa.build_exif_caption_from_result(RESULT) == (
+        "A dog on a beach. | Tags: dog, beach | Mood: joyful | "
+        "Location: Rome, Italy | ~2015")
+
+
+def test_caption_golden_sparse():
+    assert pa.build_exif_caption_from_result(
+        {"description": "X.", "tags": [], "location_hint": None}) == "X."
+
+
+def test_rec_has_nsfw_list_scalar_case():
+    assert pa._rec_has_nsfw({"Keywords": ["beach", "NSFW "]})
+    assert pa._rec_has_nsfw({"Subject": "nsfw"})
+    assert not pa._rec_has_nsfw({"Keywords": ["sfw"], "Subject": ["beach"]})
+    assert not pa._rec_has_nsfw({})
+
+
+@needs_exiftool
+def test_write_exif_roundtrip_and_idempotent(img):
+    caption = pa.build_exif_caption_from_result(RESULT)
+    assert pa.write_exif(str(img), caption, RESULT["tags"])
+    rec = pa.read_existing_exif(str(img))
+    assert rec["ImageDescription"] == caption
+    assert rec["XPComment"] == caption
+    assert rec["Subject"] == RESULT["tags"]
+    assert rec["Keywords"] == RESULT["tags"]
+
+    # second write with the same caption must not duplicate anything
+    assert pa.write_exif(str(img), caption, RESULT["tags"])
+    rec2 = pa.read_existing_exif(str(img))
+    assert rec2["ImageDescription"] == caption
+    assert rec2["Keywords"] == RESULT["tags"]
+
+
+@needs_exiftool
+def test_write_exif_preserves_existing_metadata(img):
+    subprocess.run(
+        ["exiftool", "-m", "-overwrite_original", "-Artist=Original Artist",
+         "-ImageDescription=User caption", "-Keywords=userkw", str(img)],
+        check=True, capture_output=True)
+    assert pa.write_exif(str(img), "AI caption", ["aitag"])
+    rec = pa.read_existing_exif(str(img))
+    # donor behavior: AI caption prepended, user caption kept after ' | '
+    assert rec["ImageDescription"] == "AI caption | User caption"
+    kws = rec["Keywords"] if isinstance(rec["Keywords"], list) else [rec["Keywords"]]
+    assert "userkw" in kws and "aitag" in kws, "keywords merge, never replace"
+    full = subprocess.run(["exiftool", "-j", "-Artist", str(img)],
+                          capture_output=True, text=True)
+    assert "Original Artist" in full.stdout, "non-owned field preserved"
+
+
+@needs_exiftool
+def test_filter_nsfw_tagged_splits_and_normalizes(img, tmp_path):
+    from conftest import write_fixture
+    clean = write_fixture("fx-blocks-02", tmp_path / "clean.jpg")
+    subprocess.run(["exiftool", "-m", "-overwrite_original",
+                    "-Keywords=nsfw", "-Subject=nsfw", str(img)],
+                   check=True, capture_output=True)
+    kept, skipped = pa.filter_nsfw_tagged([str(img), str(clean)])
+    assert kept == [str(clean)]
+    assert skipped == [str(img)]
+
+
+def test_filter_nsfw_tagged_empty_list_noop():
+    assert pa.filter_nsfw_tagged([]) == ([], [])

tests/characterization/test_pa_hashing_dedup.py

@@ -0,0 +1,109 @@
+"""Characterize hashing + duplicate ledger (donor: _sha1_file, _phash_image,
+ensure_hashes, cluster_duplicates, mark_duplicates, reconcile_moved).
+
+Complements the pre-existing root-level self-check test_dedup.py.
+"""
+import hashlib
+
+import photo_analyzer as pa
+from conftest import write_fixture
+
+# Captured goldens (Pillow/scipy in current env; a change here means the
+# decoder/algorithm changed — exactly what characterization should catch).
+GOLDEN_PHASH = {
+    "fx-blocks-01": "851e7ae08fa4347b",
+    "fx-blocks-02": "84333bcce3338ce9",
+    "fx-blocks-03": "841f32f069cfcd38",
+    "fx-blocks-04": "864c3c7073b761b9",
+}
+
+
+def test_phash_goldens(tmp_path):
+    for fid, expected in GOLDEN_PHASH.items():
+        p = write_fixture(fid, tmp_path / f"{fid}.jpg")
+        assert pa._phash_image(p) == expected, fid
+
+
+def test_phash_survives_resize_and_recompress(tmp_path, img):
+    from PIL import Image
+    small = tmp_path / "small.jpg"
+    with Image.open(img) as im:
+        im.resize((400, 300)).save(small, quality=70)
+    a = int(pa._phash_image(img), 16)
+    b = int(pa._phash_image(small), 16)
+    assert bin(a ^ b).count("1") <= pa.PHASH_THRESHOLD
+
+
+def test_phash_format_16_hex_chars(img):
+    ph = pa._phash_image(img)
+    assert len(ph) == 16
+    int(ph, 16)
+
+
+def test_sha1_matches_hashlib(img):
+    assert pa._sha1_file(img) == hashlib.sha1(img.read_bytes()).hexdigest()
+
+
+def test_sha1_unreadable_returns_none(tmp_path):
+    assert pa._sha1_file(tmp_path / "missing.jpg") is None
+
+
+def test_ensure_hashes_skips_hashed_and_is_resume_safe(db, tmp_path):
+    p1 = write_fixture("fx-blocks-01", tmp_path / "a.jpg")
+    p2 = write_fixture("fx-blocks-02", tmp_path / "b.jpg")
+    for p in (p1, p2):
+        pa.upsert_pending(db, str(p))
+    assert pa.ensure_hashes(db, [p1, p2]) == 2
+    # second run: nothing to do
+    assert pa.ensure_hashes(db, [p1, p2]) == 0
+    row = db.execute("SELECT phash, file_sha1 FROM photos WHERE path=?",
+                     (str(p1),)).fetchone()
+    assert row["phash"] == GOLDEN_PHASH["fx-blocks-01"]
+    assert row["file_sha1"] == pa._sha1_file(p1)
+
+
+def test_cluster_and_mark_duplicates_largest_is_canonical(db, tmp_path):
+    from PIL import Image
+    big = write_fixture("fx-blocks-01", tmp_path / "big.jpg", quality=95)
+    small = tmp_path / "small.jpg"
+    with Image.open(big) as im:
+        im.resize((400, 300)).save(small, quality=60)
+    other = write_fixture("fx-blocks-02", tmp_path / "other.jpg")
+    for p in (big, small, other):
+        pa.upsert_pending(db, str(p))
+    pa.ensure_hashes(db, [big, small, other])
+
+    clusters = pa.cluster_duplicates(db, pa.PHASH_THRESHOLD)
+    assert len(clusters) == 1
+    assert clusters[0][0]["path"] == str(big), "canonical = largest file"
+    assert {c["path"] for c in clusters[0]} == {str(big), str(small)}
+
+    marked, n = pa.mark_duplicates(db, pa.PHASH_THRESHOLD)
+    assert (marked, n) == (1, 1)
+    row = db.execute("SELECT status, dup_of FROM photos WHERE path=?",
+                     (str(small),)).fetchone()
+    assert row["status"] == "duplicate"
+    assert row["dup_of"] == str(big)
+    # duplicates never re-enter the pending queue
+    assert str(small) not in pa.get_pending(db, reanalyze=False)
+    assert str(small) not in pa.get_pending(db, reanalyze=True)
+
+
+def test_reconcile_moved_preserves_row_by_sha1(db, tmp_path):
+    lib = tmp_path / "lib"
+    old = write_fixture("fx-blocks-03", lib / "album" / "pic.jpg")
+    pa.upsert_pending(db, str(old))
+    pa.ensure_hashes(db, [old])
+    db.execute("UPDATE photos SET status='analyzed', description='kept' WHERE path=?",
+               (str(old),))
+    db.commit()
+
+    new = lib / "renamed" / "pic_new.jpg"
+    new.parent.mkdir(parents=True)
+    old.rename(new)
+
+    assert pa.reconcile_moved(db, lib) == 1
+    row = db.execute("SELECT path, status, description FROM photos").fetchone()
+    assert row["path"] == str(new)
+    assert row["status"] == "analyzed"
+    assert row["description"] == "kept", "analysis survives the move"

tests/characterization/test_pa_imaging.py

@@ -0,0 +1,38 @@
+"""Characterize image preparation (donor: prepare_image)."""
+import base64
+import io
+
+from PIL import Image
+
+import photo_analyzer as pa
+from conftest import make_image_array
+
+
+def _decode(b64: str) -> Image.Image:
+    return Image.open(io.BytesIO(base64.b64decode(b64)))
+
+
+def test_prepare_image_small_passthrough_jpeg(img):
+    b64, mime = pa.prepare_image(img)
+    assert mime == "image/jpeg"
+    out = _decode(b64)
+    assert out.size == (800, 600), "under MAX_LONG_EDGE → no resize"
+    assert out.format == "JPEG"
+
+
+def test_prepare_image_resizes_to_max_long_edge(tmp_path):
+    big = tmp_path / "big.jpg"
+    arr = make_image_array("fx-blocks-01")
+    Image.fromarray(arr).resize((4096, 3072)).save(big, quality=90)
+    b64, _ = pa.prepare_image(big)
+    out = _decode(b64)
+    assert max(out.size) == pa.MAX_LONG_EDGE
+    assert out.size == (2048, 1536), "aspect ratio preserved"
+
+
+def test_prepare_image_converts_png_alpha_to_rgb_jpeg(tmp_path):
+    p = tmp_path / "alpha.png"
+    Image.new("RGBA", (100, 80), (255, 0, 0, 128)).save(p)
+    b64, mime = pa.prepare_image(p)
+    assert mime == "image/jpeg"
+    assert _decode(b64).mode == "RGB"

tests/characterization/test_pa_variants.py

@@ -0,0 +1,45 @@
+"""Characterize variant grouping (donor: _strip_variant_markers, _is_variant,
+_variant_key, build_variant_groups)."""
+from pathlib import Path
+
+import photo_analyzer as pa
+from conftest import write_fixture
+
+
+def test_strip_variant_markers_goldens():
+    cases = {
+        "_MG_1432": "_MG_1432",
+        "_MG_1432-bearbeitet": "_MG_1432",
+        "_MG_1432 (640x427)": "_MG_1432",
+        "_MG_1432 (640x427)-bearbeitet": "_MG_1432",
+        "IMG (1920 x 1080)": "IMG",
+        "party_Bearbeitet": "party",
+        "no-markers-here": "no-markers-here",
+    }
+    for stem, expected in cases.items():
+        assert pa._strip_variant_markers(stem) == expected, stem
+
+
+def test_is_variant():
+    assert pa._is_variant(Path("/a/x (640x427).jpg"))
+    assert pa._is_variant(Path("/a/x-bearbeitet.jpg"))
+    assert not pa._is_variant(Path("/a/x.jpg"))
+
+
+def test_variant_key_folder_scoped_case_insensitive():
+    assert pa._variant_key(Path("/a/IMG (640x427).jpg")) == ("/a", "img")
+    assert pa._variant_key(Path("/b/IMG.jpg")) != pa._variant_key(Path("/a/IMG.jpg"))
+
+
+def test_build_variant_groups_primary_is_largest_unmarked(tmp_path):
+    # base (small), edited (large): primary must be the UNMARKED base even
+    # though the edited file is bigger.
+    base = write_fixture("fx-blocks-03", tmp_path / "pic.jpg", quality=30)
+    edited = write_fixture("fx-blocks-01", tmp_path / "pic-bearbeitet.jpg", quality=95)
+    resized = write_fixture("fx-blocks-03", tmp_path / "pic (640x480).jpg", quality=30)
+    lone = write_fixture("fx-blocks-02", tmp_path / "other.jpg")
+    assert edited.stat().st_size > base.stat().st_size
+
+    groups = pa.build_variant_groups([base, edited, resized, lone])
+    assert groups == {str(base): sorted([str(edited), str(resized)])}
+    assert str(lone) not in groups, "standalone files form no group"

tests/characterization/test_traceability.json

@@ -1,15 +0,0 @@
-{
-  "CHAR-001": "DonorCharacterizationTests.test_discovery_outputs",
-  "CHAR-002": "DonorCharacterizationTests.test_image_preparation_and_hash_outputs",
-  "CHAR-003": "DonorCharacterizationTests.test_database_status_and_fts_outputs",
-  "CHAR-004": "DonorCharacterizationTests.test_caption_and_variant_outputs",
-  "CHAR-005": "DonorCharacterizationTests.test_vision_dry_run_output",
-  "CHAR-006": "DonorCharacterizationTests.test_nsfw_cache_output",
-  "CHAR-007": "DonorCharacterizationTests.test_review_html_output",
-  "CHAR-008": "DonorCharacterizationTests.test_cli_entry_point_help",
-  "CHAR-009": "DonorCharacterizationTests.test_exif_command_contracts",
-  "CHAR-010": "DonorCharacterizationTests.test_error_and_configuration_fallbacks",
-  "CHAR-011": "DonorCharacterizationTests.test_history_log_shape",
-  "CHAR-012": "DonorCharacterizationTests.test_cooperative_cancellation_contract",
-  "LEDGER-001": "DonorLedgerLintTests.test_ledger_is_complete_and_resolvable"
-}

tests/characterization/test_webapp_query.py

@@ -0,0 +1,105 @@
+"""Characterize webapp read layer (donor: query._fts_query, _where, search,
+facets, stats, photo, all_paths) against a seeded temp DB."""
+from pathlib import Path
+
+import pytest
+
+import photo_analyzer as pa
+from webapp import query
+from conftest import seed_analyzed
+
+
+def test_fts_query_sanitization_goldens():
+    assert query._fts_query("beach sun") == '"beach"* "sun"*'
+    assert query._fts_query('AND OR "quoted"') == '"AND"* "OR"* "quoted"*'
+    assert query._fts_query("!!! ???") is None, "pure punctuation can't crash MATCH"
+    assert query._fts_query("") is None
+    assert query._fts_query("Straße") == '"Straße"*', "unicode words survive"
+
+
+def test_where_clause_goldens():
+    clauses, params = query._where({
+        "setting": "indoor", "people": "3+", "year_min": "2010",
+        "has_location": "1"})
+    assert clauses == [
+        "p.setting = ?", "p.people_count >= 3", "p.approx_year >= ?",
+        "p.location_hint IS NOT NULL AND TRIM(p.location_hint) != '' "
+        "AND LOWER(p.location_hint) != 'null'"]
+    assert params == ["indoor", 2010]
+    assert query._where({}) == ([], [])
+
+
+@pytest.fixture
+def seeded(db):
+    seed_analyzed(db, "/lib/album_a/beach.jpg",
+                  description="A sunny beach with palm trees.",
+                  tags=["beach", "palm"], setting="outdoor", people_count=2,
+                  approx_year=2015, location_hint="Rome, Italy")
+    seed_analyzed(db, "/lib/album_a/city.jpg",
+                  description="A night city street.", tags=["city"],
+                  setting="outdoor", time_of_day="night", people_count=0,
+                  approx_year=2011)
+    seed_analyzed(db, "/lib/album_b/dinner.jpg",
+                  description="Family dinner at home.", tags=["family", "dinner"],
+                  setting="indoor", people_count=4)
+    pa.mark_exif_written(db, "/lib/album_b/dinner.jpg")
+    pa.upsert_pending(db, "/lib/album_b/err.jpg")
+    pa.mark_error(db, "/lib/album_b/err.jpg", "timeout")
+    return db
+
+
+def test_search_fts_prefix_match(seeded):
+    out = query.search(seeded, q="beach")
+    assert out["total"] == 1
+    assert out["rows"][0]["path"] == "/lib/album_a/beach.jpg"
+    assert out["rows"][0]["tags"] == ["beach", "palm"], "tags decoded from JSON"
+    assert query.search(seeded, q="bea")["total"] == 1, "prefix term"
+
+
+def test_search_filters_and_browse(seeded):
+    assert query.search(seeded)["total"] == 4, "empty query = browse all"
+    assert query.search(seeded, filters={"setting": "indoor"})["total"] == 1
+    assert query.search(seeded, filters={"people": "3+"})["total"] == 1
+    assert query.search(seeded, filters={"year_min": 2012})["total"] == 1
+    assert query.search(seeded, filters={"has_location": "1"})["total"] == 1
+    assert query.search(seeded, q="city", filters={"setting": "indoor"})["total"] == 0, \
+        "search ANDs with filters"
+
+
+def test_search_paging(seeded):
+    page = query.search(seeded, limit=2, offset=2)
+    assert page["total"] == 4
+    assert len(page["rows"]) == 2
+    assert page["offset"] == 2
+
+
+def test_photo_omits_raw_response(seeded):
+    d = query.photo(seeded, "/lib/album_a/beach.jpg")
+    assert d["description"] == "A sunny beach with palm trees."
+    assert "raw_response" not in d
+    assert query.photo(seeded, "/nope.jpg") is None
+
+
+def test_facets_and_albums(seeded):
+    f = query.facets(seeded, Path("/lib"))
+    assert (f["year_min"], f["year_max"]) == (2011, 2015)
+    assert {a["album"]: a["count"] for a in f["albums"]} == {
+        "album_a": 2, "album_b": 2}
+    assert {v["value"]: v["count"] for v in f["setting"]} == {
+        "outdoor": 2, "indoor": 1}
+
+
+def test_stats_album_progress_and_errors(seeded):
+    s = query.stats(seeded, Path("/lib"))
+    assert s["total"] == 4
+    assert s["status"] == {"analyzed": 2, "exif_written": 1, "error": 1}
+    albums = {a["album"]: (a["done"], a["total"]) for a in s["albums"]}
+    assert albums == {"album_a": (2, 2), "album_b": (1, 2)}, \
+        "done counts analyzed + exif_written"
+    assert s["errors"] == [{"path": "/lib/album_b/err.jpg", "error": "timeout"}]
+
+
+def test_all_paths_is_the_image_endpoint_allowlist(seeded):
+    assert query.all_paths(seeded) == {
+        "/lib/album_a/beach.jpg", "/lib/album_a/city.jpg",
+        "/lib/album_b/dinner.jpg", "/lib/album_b/err.jpg"}