booth/tests/test_blur.py

"""Per-item blur storage — `.blurred` round-trips any rel, whoever writes it.

`.blurred` was one stripped rel per line, so a rel with a leading space could
not survive a write: blurring " a.png" stored "a.png", and toggled the
neighbour instead (heid bug-hunt on r2b merge 1, reported to booth-dev). The set
now lives in `.blurred.json`, a JSON array (the `.seen` shape), read without
following a link or blocking on a FIFO. The legacy `.blurred` is still READ, as
lines, while no `.blurred.json` exists; the first write retires it. Two names,
so neither format is ever sniffed (heid bug-hunt on this change, 3 of 3 arms).

Two writers share the file: the service (the operator's per-item control) and
`scripts/booth blur` (a session at post time). Both go through `booth.blur`,
which is stdlib-only so the CLI can import it under the system python3.
"""

from __future__ import annotations

import json
import os
import pathlib
import subprocess
import sys
import threading

import pytest

from fastapi.testclient import TestClient

sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))

from booth.app import BLUR_FILE, create_app, read_blurred, set_blurred  # noqa: E402
from booth.blur import BLUR_MAX_BYTES, LEGACY_BLUR_FILE, BlurUnwritable  # noqa: E402
from booth.items import booth_items  # noqa: E402

PNG = b"\x89PNG\r\n\x1a\n"
SCRIPT = pathlib.Path(__file__).parent.parent / "scripts" / "booth"


def _booth(root: pathlib.Path, name: str, files: dict[str, bytes]) -> pathlib.Path:
    b = root / name
    b.mkdir()
    for rel, data in files.items():
        (b / rel).write_bytes(data)
    return b


def _within(seconds: float, fn):
    """Run fn in a thread and fail, rather than hang the suite, if it blocks."""
    out: dict = {}
    t = threading.Thread(target=lambda: out.setdefault("v", fn()), daemon=True)
    t.start()
    t.join(seconds)
    assert not t.is_alive(), f"{fn} blocked for over {seconds}s"
    return out["v"]


# ---- the round-trip: the defect ---------------------------------------------


def test_a_leading_space_rel_round_trips(tmp_path):
    """Defeating change: storing rels line-stripped (the old format)."""
    set_blurred(tmp_path, " a.png", True)
    assert read_blurred(tmp_path) == {" a.png"}


def test_unblurring_a_leading_space_rel_leaves_its_neighbour_blurred(tmp_path):
    """The reported wrong-item write: " a.png" and "a.png" are two items, and
    toggling one must never move the other. (Unblurring the SPACED one would
    pass under the old format too — it was a no-op there — so this unblurs the
    plain one and asks whether the spaced one survived.)"""
    set_blurred(tmp_path, "a.png", True)
    set_blurred(tmp_path, " a.png", True)
    set_blurred(tmp_path, "a.png", False)
    assert read_blurred(tmp_path) == {" a.png"}


def test_a_newline_in_a_rel_round_trips(tmp_path):
    """A line format cannot hold one at all."""
    set_blurred(tmp_path, "two\nlines.png", True)
    assert read_blurred(tmp_path) == {"two\nlines.png"}


def test_the_file_is_a_json_array_in_sorted_order(tmp_path):
    """The `.seen` shape, and a stated order (invariant 6) so two writes of the
    same set are byte-identical."""
    set_blurred(tmp_path, "b.png", True)
    set_blurred(tmp_path, "a.png", True)
    assert json.loads((tmp_path / BLUR_FILE).read_text("utf-8")) == ["a.png", "b.png"]


def test_emptying_the_set_removes_the_file(tmp_path):
    """Unchanged: an empty marker is a lie by omission."""
    set_blurred(tmp_path, "a.png", True)
    set_blurred(tmp_path, "a.png", False)
    assert not (tmp_path / BLUR_FILE).exists()


# ---- the legacy format: nothing live changes until it is written ------------


def test_the_legacy_line_format_still_reads(tmp_path):
    """Six live booths hold line-format files. Defeating change: a JSON-only
    reader, which would un-blur every one of them on deploy."""
    (tmp_path / LEGACY_BLUR_FILE).write_text("a.png\nsub/b.png\n\n")
    assert read_blurred(tmp_path) == {"a.png", "sub/b.png"}


def test_a_legacy_rel_that_starts_with_a_bracket_still_reads(tmp_path):
    """A line-format file whose first rel happens to begin with "[" is not
    JSON, and must fall back to lines rather than read as nothing."""
    (tmp_path / LEGACY_BLUR_FILE).write_text("[draft] a.png\nb.png\n")
    assert read_blurred(tmp_path) == {"[draft] a.png", "b.png"}


def test_a_write_upgrades_a_legacy_file_and_keeps_its_rels(tmp_path):
    """And retires the legacy file, so it can never speak again."""
    (tmp_path / LEGACY_BLUR_FILE).write_text("a.png\n")
    set_blurred(tmp_path, "b.png", True)
    assert json.loads((tmp_path / BLUR_FILE).read_text("utf-8")) == ["a.png", "b.png"]
    assert not (tmp_path / LEGACY_BLUR_FILE).exists()


# ---- a planted file: never blocks, never follows ----------------------------


def test_a_fifo_blur_file_does_not_block_the_read(tmp_path):
    """read_blurred runs for every booth the Desk renders; a FIFO with no writer
    used to hang it — the outage class `.seen` was built against."""
    os.mkfifo(tmp_path / BLUR_FILE)
    assert _within(5, lambda: read_blurred(tmp_path)) == set()


def test_a_symlinked_blur_file_is_not_followed_on_read(tmp_path):
    outside = tmp_path / "outside.json"
    outside.write_text('["a.png"]')
    b = tmp_path / "b"
    b.mkdir()
    (b / BLUR_FILE).symlink_to(outside)
    assert read_blurred(b) == set()


def test_a_write_replaces_a_planted_symlink_rather_than_writing_through_it(tmp_path):
    outside = tmp_path / "outside.txt"
    outside.write_text("untouched")
    b = tmp_path / "b"
    b.mkdir()
    (b / BLUR_FILE).symlink_to(outside)
    set_blurred(b, "a.png", True)
    assert outside.read_text() == "untouched"
    assert not (b / BLUR_FILE).is_symlink()
    assert read_blurred(b) == {"a.png"}


def test_malformed_json_array_contents_are_skipped_not_fatal(tmp_path):
    (tmp_path / BLUR_FILE).write_text('["a.png", 3, null, ["x"]]')
    assert read_blurred(tmp_path) == {"a.png"}


# ---- the route: the operator's per-item control -----------------------------


def test_the_blur_route_blurs_exactly_the_item_it_names(tmp_path):
    """The route stripped `f` before writing, so the form for " a.png" blurred
    "a.png". Defeating change: `f.strip()` back in the route."""
    b = _booth(tmp_path, "g", {" a.png": PNG, "a.png": PNG})
    c = TestClient(create_app(tmp_path, ttl_hours=24, start_sweeper=False))
    r = c.post("/b/g/blur", data={"f": " a.png", "on": "1"}, follow_redirects=False)
    assert r.status_code == 303
    blurred = {it.rel: it.blurred for it in booth_items(b)}
    assert blurred == {" a.png": True, "a.png": False}


# ---- the CLI: the other writer ----------------------------------------------


def _cli(data: pathlib.Path, *args: str) -> subprocess.CompletedProcess:
    env = {**os.environ, "BOOTH_DATA_DIR": str(data), "BOOTH_URL": "http://booth.invalid"}
    return subprocess.run([str(SCRIPT), *args], capture_output=True, text=True, env=env, timeout=30)


def test_the_cli_writes_the_format_the_service_reads(tmp_path):
    """Both writers, one format. Defeating change: the CLI keeping its own
    grep/printf line writer, which appends a line to a JSON array."""
    b = _booth(tmp_path, "g", {" a.png": PNG, "a.png": PNG})
    set_blurred(b, "a.png", True)                     # the service wrote first
    r = _cli(tmp_path, "blur", "g", " a.png")
    assert r.returncode == 0, r.stderr
    assert read_blurred(b) == {"a.png", " a.png"}
    r = _cli(tmp_path, "unblur", "g", " a.png")
    assert r.returncode == 0, r.stderr
    assert read_blurred(b) == {"a.png"}


def test_the_cli_unblurring_the_last_item_removes_the_file(tmp_path):
    b = _booth(tmp_path, "g", {"a.png": PNG})
    assert _cli(tmp_path, "blur", "g", "a.png").returncode == 0
    assert _cli(tmp_path, "unblur", "g", "a.png").returncode == 0
    assert not (b / BLUR_FILE).exists()


def test_the_cli_still_refuses_a_dotdot_path(tmp_path):
    _booth(tmp_path, "g", {"a.png": PNG})
    r = _cli(tmp_path, "blur", "g", "../escape.png")
    assert r.returncode == 2
    assert not (tmp_path / "g" / BLUR_FILE).exists()


# ---- one read of blur state per render (invariant 3) ------------------------


def test_the_item_record_carries_its_own_blur_apart_from_the_booths(tmp_path):
    """r2b's per-item control needs the item's OWN blur as well as the composed
    one. It came from a second `read_blurred` in build_gallery — a second reader
    of one file, which a write between the two could split. It is now resolved
    in `booth_items`, from the one read the composed fact already uses."""
    b = _booth(tmp_path, "g", {"a.png": PNG, "b.png": PNG})
    set_blurred(b, "a.png", True)
    (b / ".blurbooth").write_bytes(b"")
    got = {it.rel: (it.blurred, it.blurred_self) for it in booth_items(b)}
    assert got == {"a.png": (True, True), "b.png": (True, False)}


def test_app_py_never_reads_the_blur_file_itself():
    """Invariant 3, extended from route bodies to the whole module: blur state
    is read in `booth_items` and nowhere in app.py. Defeating change: the
    second `read_blurred` in build_gallery."""
    import ast
    src = pathlib.Path(__file__).parent.parent / "booth" / "app.py"
    calls = [
        n for n in ast.walk(ast.parse(src.read_text()))
        if isinstance(n, ast.Call) and getattr(n.func, "id", getattr(n.func, "attr", None)) == "read_blurred"
    ]
    assert calls == []


# ---- the heid bug-hunt on this change (3 arms), folded -------------------------


@pytest.mark.parametrize("line", ['["a.png"]', "[]", "[1,2]"])
def test_a_legacy_line_that_is_valid_json_still_reads_as_a_line(tmp_path, line):
    """3 of 3 arms. Sniffing one file for two formats misread a legacy file
    whose ONE line is an item literally named like a JSON array: `["a.png"]`
    read as {"a.png"}, un-blurring the item and blurring its neighbour — the bug
    this change exists to fix, recreated by its migration. Defeating change:
    trying JSON on the legacy file."""
    (tmp_path / LEGACY_BLUR_FILE).write_text(line + "\n")
    assert read_blurred(tmp_path) == {line}


def test_a_stale_legacy_file_is_silent_once_the_current_one_exists(tmp_path):
    (tmp_path / LEGACY_BLUR_FILE).write_text("old.png\n")
    (tmp_path / BLUR_FILE).write_text('["new.png"]')
    assert read_blurred(tmp_path) == {"new.png"}


def test_a_planted_directory_at_the_blur_file_is_a_refusal_not_a_crash(tmp_path):
    """2 of 3 arms plus a third from another angle: the reader was hardened
    against a planted directory, the writer was not, and `os.replace` onto a
    directory raised IsADirectoryError through the route. Defeating change:
    letting the OSError out of set_blurred."""
    (tmp_path / BLUR_FILE).mkdir()
    with pytest.raises(BlurUnwritable):
        set_blurred(tmp_path, "a.png", True)
    assert (tmp_path / BLUR_FILE).is_dir(), "a planted directory is not ours to remove"


def test_unblurring_under_a_planted_directory_is_not_an_error(tmp_path):
    """Nothing reads as blurred and nothing was asked to be: the reader agrees
    with the request, so there is nothing to refuse. A directory holds no set,
    so strict writes (below) have nothing to protect here."""
    (tmp_path / BLUR_FILE).mkdir()
    assert set_blurred(tmp_path, "a.png", False) == set()


def test_a_planted_directory_at_the_legacy_name_does_not_block_a_write(tmp_path):
    (tmp_path / LEGACY_BLUR_FILE).mkdir()
    set_blurred(tmp_path, "a.png", True)
    assert read_blurred(tmp_path) == {"a.png"}


def test_the_route_answers_a_planted_directory_with_409(tmp_path):
    b = _booth(tmp_path, "g", {"a.png": PNG})
    (b / BLUR_FILE).mkdir()
    c = TestClient(create_app(tmp_path, ttl_hours=24, start_sweeper=False))
    r = c.post("/b/g/blur", data={"f": "a.png", "on": "1"}, follow_redirects=False)
    assert r.status_code == 409


def test_a_lone_surrogate_in_the_file_is_skipped_and_writes_still_work(tmp_path):
    """hulda, execution-verified: `"\\ud800"` is a valid JSON string no filename
    can produce, and the writer's UTF-8 encode raised on it, so one planted
    escape froze the booth's blur. Defeating change: keeping every str member."""
    (tmp_path / BLUR_FILE).write_text('["\\ud800", "a.png"]')
    assert read_blurred(tmp_path) == {"a.png"}
    assert set_blurred(tmp_path, "b.png", True) == {"a.png", "b.png"}


@pytest.mark.parametrize("rel", ["", "/abs.png", "a/../b.png", "..", "\ud800.png"])
def test_a_rel_that_is_not_an_item_path_is_refused(tmp_path, rel):
    with pytest.raises(ValueError):
        set_blurred(tmp_path, rel, True)
    assert not (tmp_path / BLUR_FILE).exists()


def test_a_double_dot_inside_a_name_is_an_item_path(tmp_path):
    """A `..` COMPONENT is an escape; `a..b.png` is a filename."""
    assert set_blurred(tmp_path, "a..b.png", True) == {"a..b.png"}


def test_the_route_refuses_an_empty_rel(tmp_path):
    """kimi: `f="/"` stripped to "" and was stored as a member no item can have."""
    _booth(tmp_path, "g", {"a.png": PNG})
    c = TestClient(create_app(tmp_path, ttl_hours=24, start_sweeper=False))
    r = c.post("/b/g/blur", data={"f": "/", "on": "1"}, follow_redirects=False)
    assert r.status_code == 400
    assert not (tmp_path / "g" / BLUR_FILE).exists()


def test_the_writer_never_writes_a_set_the_reader_would_refuse(tmp_path, monkeypatch):
    """2 of 3 arms: nothing capped the writer, the reader refuses a file over
    the cap and reads it as EMPTY, so the write that crossed it revealed every
    item. Defeating change: no size check before the write."""
    import booth.blur as blur
    set_blurred(tmp_path, "a.png", True)
    monkeypatch.setattr(blur, "BLUR_MAX_BYTES", len(b'["a.png"]') + 3)
    with pytest.raises(BlurUnwritable):
        set_blurred(tmp_path, "bbbbbbbb.png", True)
    monkeypatch.setattr(blur, "BLUR_MAX_BYTES", BLUR_MAX_BYTES)
    assert read_blurred(tmp_path) == {"a.png"}, "a refused write changed the set"


def test_the_cli_accepts_a_double_dot_inside_a_name(tmp_path):
    """2 of 3 arms: the CLI's `*..*` substring guard refused `a..b.png`, which
    the route accepts. One predicate now serves both."""
    b = _booth(tmp_path, "g", {"a..b.png": PNG})
    r = _cli(tmp_path, "blur", "g", "a..b.png")
    assert r.returncode == 0, r.stderr
    assert read_blurred(b) == {"a..b.png"}


def test_the_cli_refuses_an_empty_item_path_before_writing(tmp_path):
    """regin: `booth blur g /` stored an empty member. Refused, and a valid
    item named alongside it is not written either."""
    b = _booth(tmp_path, "g", {"a.png": PNG})
    r = _cli(tmp_path, "blur", "g", "a.png", "/")
    assert r.returncode == 2
    assert read_blurred(b) == set()


def test_the_cli_refuses_a_planted_directory_with_a_message(tmp_path):
    b = _booth(tmp_path, "g", {"a.png": PNG})
    (b / BLUR_FILE).mkdir()
    r = _cli(tmp_path, "blur", "g", "a.png")
    assert r.returncode == 3
    assert "Traceback" not in r.stderr and BLUR_FILE in r.stderr


def test_the_cli_fails_closed_without_its_package(tmp_path):
    """kimi: the `link` verb says why and exits 3 when booth/ is missing; the
    `blur` verb died with a bare traceback. Same deployment shape as
    test_cli's link test: the script alone, no package beside it."""
    b = _booth(tmp_path, "g", {"a.png": PNG})
    lone = tmp_path / "lone" / "scripts"
    lone.mkdir(parents=True)
    (lone / "booth").write_text(SCRIPT.read_text())
    (lone / "booth").chmod(0o755)
    env = {k: v for k, v in os.environ.items() if k != "PYTHONPATH"}
    env.update(BOOTH_DATA_DIR=str(tmp_path), BOOTH_URL="http://booth.invalid")
    r = subprocess.run([str(lone / "booth"), "blur", "g", "a.png"], capture_output=True,
                       text=True, env=env, cwd="/tmp", timeout=30)
    assert r.returncode == 3
    assert "Traceback" not in r.stderr
    assert read_blurred(b) == set()


def test_a_fifo_at_the_legacy_name_does_not_block_the_read(tmp_path):
    os.mkfifo(tmp_path / LEGACY_BLUR_FILE)
    assert _within(5, lambda: read_blurred(tmp_path)) == set()


# ---- reads lenient, writes strict (groa's retry, and marks' lesson) ------------
#
# The reader turns anything it cannot read into an EMPTY set, which is right for
# rendering: a damaged file costs the blur, never the page. A writer that builds
# on that empty set then replaces the file, and whatever it could not read is
# gone. That is the `.marks.json` wipe of 2026-09-21
# (persistent-memory.d/2026-09-21-marks-write-wiped-judgment.md), and a
# cross-uid reader that got EACCES would do it here (groa).


def test_an_unreadable_blur_file_is_never_overwritten(tmp_path):
    """Defeating change: set_blurred building on the lenient reader."""
    set_blurred(tmp_path, "a.png", True)
    before = (tmp_path / BLUR_FILE).read_bytes()
    os.chmod(tmp_path / BLUR_FILE, 0)
    try:
        with pytest.raises(BlurUnwritable):
            set_blurred(tmp_path, "b.png", True)
    finally:
        os.chmod(tmp_path / BLUR_FILE, 0o644)
    assert (tmp_path / BLUR_FILE).read_bytes() == before


def test_a_malformed_blur_file_is_never_overwritten(tmp_path):
    (tmp_path / BLUR_FILE).write_text("not json at all")
    with pytest.raises(BlurUnwritable):
        set_blurred(tmp_path, "a.png", True)
    assert (tmp_path / BLUR_FILE).read_text() == "not json at all"


def test_an_oversized_blur_file_is_never_overwritten(tmp_path, monkeypatch):
    import booth.blur as blur
    set_blurred(tmp_path, "a.png", True)
    before = (tmp_path / BLUR_FILE).read_bytes()
    monkeypatch.setattr(blur, "BLUR_MAX_BYTES", 4)
    with pytest.raises(BlurUnwritable):
        set_blurred(tmp_path, "b.png", False)
    assert (tmp_path / BLUR_FILE).read_bytes() == before


def test_the_blur_file_is_world_readable_as_it_always_was(tmp_path):
    """groa: mkstemp creates 0600, where the line-format writer left 0644, so a
    reader under another uid saw nothing. Defeating change: no chmod."""
    set_blurred(tmp_path, "a.png", True)
    assert (tmp_path / BLUR_FILE).stat().st_mode & 0o777 == 0o644


# The open flags are the SECOND layer: `_load` lstat-checks for a regular file
# first, so a FIFO or a link never reaches `os.open` through the public API, and
# a mutation run found the flags VACUOUS there. They still close the race (a
# file swapped for a FIFO or a link between the lstat and the open), so they
# are held to account directly, where nothing stands in front of them.


def test_the_raw_read_never_blocks_on_a_fifo(tmp_path):
    from booth.blur import _read_capped
    os.mkfifo(tmp_path / "f")
    assert _within(5, lambda: _read_capped(tmp_path / "f")) is None


def test_the_raw_read_never_follows_a_link(tmp_path):
    from booth.blur import _read_capped
    (tmp_path / "real.json").write_text('["a.png"]')
    (tmp_path / "link").symlink_to(tmp_path / "real.json")
    assert _read_capped(tmp_path / "link") is None