test: add characterization tests for core modules

33 tests covering env/io helpers, config parsers, pipeline meta I/O,
worker inbox logic, and FastAPI smoke routes. No gmic/rembg subprocesses.

Co-authored-by: Cursor <cursoragent@cursor.com>

tests/conftest.py

@@ -0,0 +1,26 @@
+"""Shared test fixtures."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def data_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    """Point config path constants at a temporary data directory."""
+    variants = tmp_path / "variants"
+    intermediates = tmp_path / "intermediates"
+    meta = tmp_path / "meta"
+    inbox = tmp_path / "incoming"
+    for directory in (variants, intermediates, meta, inbox):
+        directory.mkdir()
+
+    monkeypatch.setattr("app.config.DATA_DIR", tmp_path)
+    monkeypatch.setattr("app.config.VARIANTS_DIR", variants)
+    monkeypatch.setattr("app.config.INTERMEDIATES_DIR", intermediates)
+    monkeypatch.setattr("app.config.META_DIR", meta)
+    monkeypatch.setattr("app.config.INBOX_DIR", inbox)
+    monkeypatch.setattr("app.config.PROCESSED_FILE", tmp_path / "processed.json")
+    return tmp_path

tests/test_config.py

@@ -0,0 +1,30 @@
+"""Tests for config helpers."""
+
+from __future__ import annotations
+
+from app import config
+from app.config import _float_env, _int_env
+
+
+def test_int_env_valid(monkeypatch) -> None:
+    monkeypatch.setenv("TEST_INT", "42")
+    assert _int_env("TEST_INT", 0) == 42
+
+
+def test_int_env_invalid_falls_back(monkeypatch) -> None:
+    monkeypatch.setenv("TEST_INT", "nope")
+    assert _int_env("TEST_INT", 7) == 7
+
+
+def test_float_env_valid(monkeypatch) -> None:
+    monkeypatch.setenv("TEST_FLOAT", "2.5")
+    assert _float_env("TEST_FLOAT", 0.0) == 2.5
+
+
+def test_supported_extensions() -> None:
+    assert ".jpg" in config.SUPPORTED_EXTENSIONS
+    assert ".jpeg" in config.SUPPORTED_EXTENSIONS
+
+
+def test_opacity_choices_percentages() -> None:
+    assert all(choice.endswith("%") for choice in config.OPACITY_CHOICES)

tests/test_env_utils.py

@@ -0,0 +1,22 @@
+"""Tests for env_utils."""
+
+from __future__ import annotations
+
+from app.env_utils import env_bool
+
+
+def test_env_bool_default(monkeypatch) -> None:
+    monkeypatch.delenv("TEST_FLAG", raising=False)
+    assert env_bool("TEST_FLAG", True) is True
+    assert env_bool("TEST_FLAG", False) is False
+
+
+def test_env_bool_truthy(monkeypatch) -> None:
+    for value in ("1", "true", "TRUE", "yes", "on"):
+        monkeypatch.setenv("TEST_FLAG", value)
+        assert env_bool("TEST_FLAG", False) is True
+
+
+def test_env_bool_falsy(monkeypatch) -> None:
+    monkeypatch.setenv("TEST_FLAG", "0")
+    assert env_bool("TEST_FLAG", True) is False

tests/test_io_utils.py

@@ -0,0 +1,26 @@
+"""Tests for io_utils."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from app.io_utils import read_json, write_json_atomic
+
+
+def test_read_json_missing_returns_default(tmp_path: Path) -> None:
+    assert read_json(tmp_path / "missing.json", default={}) == {}
+
+
+def test_read_json_corrupt_returns_default(tmp_path: Path) -> None:
+    path = tmp_path / "bad.json"
+    path.write_text("not json", encoding="utf-8")
+    assert read_json(path, default=None) is None
+
+
+def test_write_json_atomic_roundtrip(tmp_path: Path) -> None:
+    path = tmp_path / "data.json"
+    payload = {"job_id": "photo", "variants": []}
+    write_json_atomic(path, payload)
+    assert json.loads(path.read_text(encoding="utf-8")) == payload
+    assert not path.with_suffix(path.suffix + ".tmp").exists()

tests/test_main.py

@@ -0,0 +1,42 @@
+"""Tests for FastAPI routes and validation."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from app import config, pipeline
+from app.main import _safe_job_file, _validate_job_id, app
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture
+def client(data_dir: Path) -> TestClient:
+    return TestClient(app)
+
+
+def test_validate_job_id_sanitizes_input() -> None:
+    assert _validate_job_id("foo@bar") == "foo_bar"
+    assert _validate_job_id("photo01") == "photo01"
+
+
+def test_safe_job_file_rejects_wrong_prefix(data_dir: Path) -> None:
+    stem = "photo01"
+    paths = pipeline.job_paths(stem)
+    paths.variants.mkdir(parents=True, exist_ok=True)
+    target = paths.variants / f"{stem}_v1.png"
+    target.write_bytes(b"x")
+    with pytest.raises(HTTPException):
+        _safe_job_file(stem, "other_v1.png")
+
+
+def test_index_empty(client: TestClient) -> None:
+    response = client.get("/")
+    assert response.status_code == 200
+    assert config.SITE_TITLE in response.text
+
+
+def test_job_not_found(client: TestClient) -> None:
+    response = client.get("/jobs/does-not-exist")
+    assert response.status_code == 404

tests/test_pipeline.py

@@ -0,0 +1,80 @@
+"""Tests for pipeline pure helpers and meta I/O."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from app import pipeline
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [
+        ("photo.jpg", "photo"),
+        ("../evil", "evil"),
+        ("", "photo"),
+        ("  spaced  ", "spaced"),
+        ("foo@bar", "foo_bar"),
+    ],
+)
+def test_sanitize_stem(raw: str, expected: str) -> None:
+    assert pipeline.sanitize_stem(raw) == expected
+
+
+def test_variant_filename() -> None:
+    assert pipeline.variant_filename("photo", "v1") == "photo_v1.png"
+
+
+@pytest.mark.parametrize(
+    ("command", "expected_tail"),
+    [
+        ("fx_blur 3", ["fx_blur", "3"]),
+        ("normalize", ["normalize"]),
+    ],
+)
+def test_gmic_filter_args(tmp_path: Path, command: str, expected_tail: list[str]) -> None:
+    image = tmp_path / "in.png"
+    image.touch()
+    args = pipeline.gmic_filter_args(image, command)
+    assert args[0] == str(image)
+    assert args[1:] == expected_tail
+
+
+def test_next_variant_id() -> None:
+    manifest = {"variants": [{"id": "v1"}, {"id": "v2"}]}
+    assert pipeline.next_variant_id(manifest) == "v3"
+
+
+def test_next_variant_id_after_gap() -> None:
+    manifest = {"variants": [{"id": "v1"}, {"id": "v3"}]}
+    assert pipeline.next_variant_id(manifest) == "v4"
+
+
+def test_strip_ansi() -> None:
+    assert pipeline.strip_ansi("\x1b[31merror\x1b[0m") == "error"
+
+
+def test_read_write_meta_roundtrip(data_dir: Path) -> None:
+    paths = pipeline.job_paths("event01")
+    pipeline.write_status(paths, "processing", source_file="event01.jpg")
+    meta = pipeline.read_meta("event01")
+    assert meta is not None
+    assert meta["status"] == "processing"
+    assert meta["source_file"] == "event01.jpg"
+
+    status = pipeline.read_status("event01")
+    assert status is not None
+    assert status["status"] == "processing"
+
+    manifest = pipeline.read_manifest("event01")
+    assert manifest is not None
+    assert manifest["variants"] == []
+
+
+def test_list_job_ids(data_dir: Path) -> None:
+    paths = pipeline.job_paths("a")
+    pipeline.write_status(paths, "done")
+    paths = pipeline.job_paths("b")
+    pipeline.write_status(paths, "processing")
+    assert sorted(pipeline.list_job_ids()) == ["a", "b"]

tests/test_worker.py

@@ -0,0 +1,46 @@
+"""Tests for worker inbox helpers."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from app import config
+from app.worker import _file_key, _is_ignored_name, find_new_files, load_processed, save_processed
+
+
+def test_is_ignored_name() -> None:
+    assert _is_ignored_name(".stfolder") is True
+    assert _is_ignored_name(".syncthing.tmp") is True
+    assert _is_ignored_name("photo.jpg") is False
+
+
+def test_find_new_files_skips_processed(data_dir: Path) -> None:
+    photo = config.INBOX_DIR / "party.jpg"
+    photo.write_bytes(b"fake")
+    stat = photo.stat()
+    processed = {
+        _file_key(photo): {
+            "size": stat.st_size,
+            "mtime": stat.st_mtime,
+        }
+    }
+    assert find_new_files(processed) == []
+
+
+def test_find_new_files_picks_supported(data_dir: Path) -> None:
+    (config.INBOX_DIR / "party.jpg").write_bytes(b"fake")
+    (config.INBOX_DIR / "notes.txt").write_text("nope", encoding="utf-8")
+    assert len(find_new_files({})) == 1
+
+
+def test_load_processed_corrupt_warns(data_dir: Path, caplog) -> None:
+    config.PROCESSED_FILE.write_text("{bad", encoding="utf-8")
+    with caplog.at_level("WARNING"):
+        assert load_processed() == {}
+    assert "processed.json unreadable" in caplog.text
+
+
+def test_save_processed_roundtrip(data_dir: Path) -> None:
+    payload = {"incoming/party.jpg": {"size": 1, "mtime": 2}}
+    save_processed(payload)
+    assert load_processed() == payload