diff --git a/postpeer_pilot/config.py b/postpeer_pilot/config.py index c8967d0..e5e6400 100644 --- a/postpeer_pilot/config.py +++ b/postpeer_pilot/config.py @@ -12,6 +12,7 @@ scheduled.jsonl local ledger of what this tool scheduled (series caps) """ import json +from . import safe_read import os from pathlib import Path @@ -51,8 +52,9 @@ def load() -> dict: cfg = dict(DEFAULTS) f = HOME / "config.json" - if f.exists(): - cfg.update(json.loads(f.read_text())) + raw = safe_read.read_text_if_present(f) + if raw is not None: + cfg.update(json.loads(raw)) return cfg @@ -60,8 +62,7 @@ def api_key() -> str: if os.environ.get("POSTPEER_API_KEY"): return os.environ["POSTPEER_API_KEY"] env = HOME / ".env" - if env.exists(): - for line in env.read_text().splitlines(): + for line in safe_read.read_lines_if_present(env): if line.startswith("POSTPEER_API_KEY="): return line.split("=", 1)[1].strip() raise RuntimeError(f"POSTPEER_API_KEY not set (env var or {env})") diff --git a/postpeer_pilot/perf.py b/postpeer_pilot/perf.py index 9f37792..6923afb 100644 --- a/postpeer_pilot/perf.py +++ b/postpeer_pilot/perf.py @@ -20,7 +20,7 @@ from datetime import datetime, timezone from pathlib import Path -from . import config +from . import config, safe_read STORE = config.HOME / "performance.jsonl" @@ -42,10 +42,8 @@ def _append(rows: list): def latest() -> list: """Latest snapshot per (source, title): [{'title', 'views', 'source'}].""" - if not STORE.exists(): - return [] best = {} - for ln in STORE.read_text().splitlines(): + for ln in safe_read.read_lines_if_present(STORE): try: r = json.loads(ln) best[(r.get("source"), r["title"])] = r # append-only: later line wins @@ -128,8 +126,7 @@ def pull_meta(limit: int = 50) -> int: return 0 token = None env = Path(m.get("env", "")).expanduser() - if env.exists(): - for ln in env.read_text().splitlines(): + for ln in safe_read.read_lines_if_present(env): if ln.startswith("META_PAGE_TOKEN="): token = ln.split("=", 1)[1].strip() if not token: diff --git a/postpeer_pilot/plan.py b/postpeer_pilot/plan.py index 0eb7ddc..3dcf4a4 100644 --- a/postpeer_pilot/plan.py +++ b/postpeer_pilot/plan.py @@ -7,7 +7,7 @@ from collections import defaultdict from datetime import date, datetime, timedelta -from . import api, config +from . import api, config, safe_read PLAN_FILE = config.HOME / "plan.json" LEDGER = config.HOME / "scheduled.jsonl" @@ -19,8 +19,9 @@ def active() -> dict: plan = {"per_day_by_wd": {int(k): int(v) for k, v in cfg["per_day_by_wd"].items()}, "slot_hours": list(cfg["slot_hours"]), "series_day_cap": int(cfg["series_day_cap"]), "source": "config defaults"} - if PLAN_FILE.exists(): - p = json.loads(PLAN_FILE.read_text()) + raw = safe_read.read_text_if_present(PLAN_FILE) + if raw is not None: + p = json.loads(raw) plan.update({"per_day_by_wd": {int(k): int(v) for k, v in p["per_day_by_wd"].items()}, "slot_hours": list(p.get("slot_hours", plan["slot_hours"])), "source": "plan.json"}) @@ -49,8 +50,7 @@ def _series_by_day(exclude_slots: set | None = None) -> dict: handed out AND ledger-recorded in the same run must not count twice toward the cap.""" out = defaultdict(list) exclude_slots = exclude_slots or set() - if LEDGER.exists(): - for ln in LEDGER.read_text().splitlines(): + for ln in safe_read.read_lines_if_present(LEDGER): try: r = json.loads(ln) if r.get("series") and r["when"] not in exclude_slots: @@ -70,10 +70,8 @@ def record(when: str, video: str, series: str | None, post_id: str = "", def ledger_entries() -> list: - if not LEDGER.exists(): - return [] out = [] - for ln in LEDGER.read_text().splitlines(): + for ln in safe_read.read_lines_if_present(LEDGER): try: out.append(json.loads(ln)) except json.JSONDecodeError: @@ -86,8 +84,7 @@ def ledger_by_post_id() -> dict: performance matching: a published post whose id is in here needs no fuzzy text match to know its canonical caption.""" out = {} - if LEDGER.exists(): - for ln in LEDGER.read_text().splitlines(): + for ln in safe_read.read_lines_if_present(LEDGER): try: r = json.loads(ln) if r.get("post_id"): diff --git a/postpeer_pilot/safe_read.py b/postpeer_pilot/safe_read.py new file mode 100644 index 0000000..0cc0213 --- /dev/null +++ b/postpeer_pilot/safe_read.py @@ -0,0 +1,44 @@ +"""Absent and unreadable are not the same thing. + +`Path.exists()` returns **False** when the OS refuses access — macOS TCC, unix +permissions, an unmounted network share. Used as a guard before reading a file, +it silently turns "I am not allowed to read this" into "this does not exist", +and the caller falls back to defaults without anyone noticing. + +For a publishing tool that failure is not cosmetic. A plan file that cannot be +read looks like *no plan*, so the scheduler quietly books against built-in +defaults; a store that cannot be read looks like *no history*, so the damped +planner sees an empty performance record. Both produce confident, wrong output. + +The rule this module enforces: a read either returns content, returns ``None`` +because the file genuinely is not there, or raises. It never degrades quietly. +""" +from __future__ import annotations + +from pathlib import Path + + +class Unreadable(OSError): + """The path exists (or its status is unknowable) but could not be read.""" + + +def read_text_if_present(path: Path) -> str | None: + """Return file contents, or ``None`` if the file genuinely does not exist. + + Raises :class:`Unreadable` for every other failure — permission denied, + unreadable mount, a directory where a file was expected. Callers that want + a default must choose it explicitly after catching, so the fallback is a + decision in the code rather than an accident of the filesystem. + """ + try: + return Path(path).read_text() + except FileNotFoundError: + return None + except OSError as exc: # PermissionError, NotADirectoryError, EIO … + raise Unreadable(f"{path} exists but could not be read: {exc}") from exc + + +def read_lines_if_present(path: Path) -> list[str]: + """``read_text_if_present`` split into lines; ``[]`` when the file is absent.""" + raw = read_text_if_present(path) + return raw.splitlines() if raw is not None else [] diff --git a/tests/test_safe_read.py b/tests/test_safe_read.py new file mode 100644 index 0000000..9e9c992 --- /dev/null +++ b/tests/test_safe_read.py @@ -0,0 +1,67 @@ +"""Unreadable must never look like absent. + +Regression for the failure found on 2026-07-20: `queue_status` reported an empty +review folder while nine finished videos sat in it. The guard was +`if PATH.exists()` — and `exists()` returns False when the OS denies access, so a +permission problem was indistinguishable from "nothing there". Anything that +schedules against such a reading books work it cannot see. +""" +import json +import os +import stat + +import pytest + +from postpeer_pilot import config, plan, safe_read + + +def test_absent_file_reads_as_none(tmp_path): + assert safe_read.read_text_if_present(tmp_path / "nope.json") is None + assert safe_read.read_lines_if_present(tmp_path / "nope.jsonl") == [] + + +def test_present_file_reads_through(tmp_path): + f = tmp_path / "there.txt" + f.write_text("a\nb\n") + assert safe_read.read_text_if_present(f) == "a\nb\n" + assert safe_read.read_lines_if_present(f) == ["a", "b"] + + +@pytest.mark.skipif(os.geteuid() == 0, reason="root bypasses file permissions") +def test_unreadable_file_raises_instead_of_defaulting(tmp_path): + f = tmp_path / "locked.json" + f.write_text("{}") + f.chmod(0) + try: + with pytest.raises(safe_read.Unreadable): + safe_read.read_text_if_present(f) + finally: + f.chmod(stat.S_IRUSR | stat.S_IWUSR) + + +@pytest.mark.skipif(os.geteuid() == 0, reason="root bypasses file permissions") +def test_unreadable_plan_does_not_silently_fall_back(tmp_path, monkeypatch): + """The important one: an unreadable plan must NOT masquerade as 'no plan', + because the scheduler would then quietly book against built-in defaults.""" + pf = tmp_path / "plan.json" + pf.write_text(json.dumps({"per_day_by_wd": {str(i): 1 for i in range(7)}})) + pf.chmod(0) + monkeypatch.setattr(plan, "PLAN_FILE", pf) + try: + with pytest.raises(safe_read.Unreadable): + plan.active() + finally: + pf.chmod(stat.S_IRUSR | stat.S_IWUSR) + + +@pytest.mark.skipif(os.geteuid() == 0, reason="root bypasses file permissions") +def test_unreadable_config_does_not_silently_fall_back(tmp_path, monkeypatch): + f = tmp_path / "config.json" + f.write_text("{}") + f.chmod(0) + monkeypatch.setattr(config, "HOME", tmp_path) + try: + with pytest.raises(safe_read.Unreadable): + config.load() + finally: + f.chmod(stat.S_IRUSR | stat.S_IWUSR)