US01-01: Inventory and Characterize Donor Code
This commit is contained in:
295
tests/characterization/test_donors.py
Normal file
295
tests/characterization/test_donors.py
Normal file
@@ -0,0 +1,295 @@
|
||||
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("<portrait>.jpg", page)
|
||||
self.assertIn("album & 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()
|
||||
62
tests/characterization/test_ledger.py
Normal file
62
tests/characterization/test_ledger.py
Normal file
@@ -0,0 +1,62 @@
|
||||
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()
|
||||
15
tests/characterization/test_traceability.json
Normal file
15
tests/characterization/test_traceability.json
Normal file
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"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"
|
||||
}
|
||||
Reference in New Issue
Block a user