rfc-app/backend/tests/test_prune_absent_projects.py

"""§9 hardening — the registry reconcile prunes projects the registry no longer
declares (was additive-only).

A re-pin to a new registry used to strand the old projects' rows (PPE's
stale-ecomm 1238-row entry cache starving writes). `prune_absent_projects` now
deletes a removed project together with its collections and every project-scoped
row, at whole-project granularity, with a `foreign_key_check` backstop. The
safety guard is structural: `refresh_registry` raises on a read/transport error
and `parse_registry` rejects an empty project list, so the prune is reached only
after a real, non-empty registry parse — a transient failure can't wipe data.
"""
from __future__ import annotations

import tempfile
from pathlib import Path

import pytest

import app.db as db
from app import projects, registry


class _Cfg:
    def __init__(self, path, default_id):
        self.database_path = path
        self.default_project_id = default_id


def _two_project_db(monkeypatch, default_id="ohm"):
    """A multi-project DB: default project 'ohm' (collection 'default') + a second
    project 'ecomm' (collection 'ecomm'), each with a rich set of project- and
    collection-keyed rows, plus the non-keyed `thread_messages` descendant."""
    path = str(Path(tempfile.mkdtemp()) / "t.db")
    cfg = _Cfg(path, default_id)
    db.run_migrations(cfg)
    monkeypatch.setattr(db, "_CONN", db.connect(path))
    conn = db.conn()
    conn.execute("DELETE FROM collections")
    conn.execute("DELETE FROM projects")
    conn.execute("INSERT INTO projects (id,name,content_repo,visibility) VALUES ('ohm','OHM','ohm-content','public')")
    conn.execute("INSERT INTO projects (id,name,content_repo,visibility) VALUES ('ecomm','Ecomm','ecomm-content','public')")
    conn.execute("INSERT INTO collections (id,project_id,type,subfolder,initial_state,visibility,name) "
                 "VALUES ('default','ohm','document','','super-draft','public','OHM')")
    conn.execute("INSERT INTO collections (id,project_id,type,subfolder,initial_state,visibility,name) "
                 "VALUES ('ecomm','ecomm','bdd','ecomm','super-draft','public','Ecomm')")
    conn.execute("INSERT INTO users (id,gitea_login,display_name,role) VALUES (1,'a','A','contributor')")

    def seed(proj, coll, slug):
        conn.execute("INSERT INTO cached_rfcs (slug,title,state,collection_id) VALUES (?,?,'active',?)",
                     (slug, slug.title(), coll))
        conn.execute("INSERT INTO cached_branches (rfc_slug,branch_name,collection_id) VALUES (?, 'main', ?)",
                     (slug, coll))
        conn.execute("INSERT INTO cached_prs (rfc_slug,pr_kind,repo,pr_number,title,state,project_id) "
                     "VALUES (?, 'rfc_branch', ?, 1, 't', 'open', ?)", (slug, proj, proj))
        conn.execute("INSERT INTO stars (user_id,rfc_slug,collection_id) VALUES (1,?,?)", (slug, coll))
        conn.execute("INSERT INTO watches (user_id,rfc_slug,state,set_by,collection_id) VALUES (1,?,'watching','explicit',?)",
                     (slug, coll))
        conn.execute("INSERT INTO changes (rfc_slug,branch_name,kind,original,proposed,project_id) "
                     "VALUES (?, 'main', 'ai', 'o', 'p', ?)", (slug, proj))
        conn.execute("INSERT INTO notifications (recipient_user_id,event_kind,project_id) VALUES (1,'x',?)", (proj,))
        cur = conn.execute("INSERT INTO threads (rfc_slug,anchor_kind,thread_kind,project_id) "
                           "VALUES (?, 'whole-doc', 'chat', ?)", (slug, proj))
        conn.execute("INSERT INTO thread_messages (thread_id,role,text) VALUES (?, 'user', 'hi')", (cur.lastrowid,))

    seed("ohm", "default", "human")
    seed("ecomm", "ecomm", "cart")
    assert conn.execute("PRAGMA foreign_key_check").fetchall() == []  # seed is FK-clean
    return cfg, conn


_PRUNED_TABLES = [
    "cached_rfcs", "cached_branches", "cached_prs", "stars", "watches",
    "changes", "notifications", "threads",
]


def test_prune_removes_absent_project_and_all_its_data(monkeypatch):
    cfg, conn = _two_project_db(monkeypatch, default_id="ohm")
    pruned = projects.prune_absent_projects(cfg, {"ohm"})  # ecomm absent
    assert pruned == ["ecomm"]

    # ecomm: project, collection, and every keyed row gone.
    assert conn.execute("SELECT 1 FROM projects WHERE id='ecomm'").fetchone() is None
    assert conn.execute("SELECT 1 FROM collections WHERE id='ecomm'").fetchone() is None
    for t in _PRUNED_TABLES:
        col = "collection_id" if t in ("cached_rfcs", "cached_branches", "stars", "watches") else None
        if col:
            assert conn.execute(f"SELECT COUNT(*) n FROM {t} WHERE collection_id='ecomm'").fetchone()["n"] == 0, t
        else:
            assert conn.execute(f"SELECT COUNT(*) n FROM {t} WHERE project_id='ecomm'").fetchone()["n"] == 0, t
    # The non-keyed descendant (thread_messages on ecomm's thread) is gone too.
    assert conn.execute("SELECT COUNT(*) n FROM thread_messages").fetchone()["n"] == 1  # only ohm's remains

    # ohm (the default) is fully intact.
    assert conn.execute("SELECT 1 FROM projects WHERE id='ohm'").fetchone() is not None
    assert conn.execute("SELECT 1 FROM collections WHERE id='default'").fetchone() is not None
    assert conn.execute("SELECT COUNT(*) n FROM cached_rfcs WHERE slug='human'").fetchone()["n"] == 1
    assert conn.execute("SELECT COUNT(*) n FROM threads WHERE rfc_slug='human'").fetchone()["n"] == 1

    # FK integrity intact after the prune.
    assert conn.execute("PRAGMA foreign_key_check").fetchall() == []


def test_prune_is_noop_when_all_projects_present(monkeypatch):
    cfg, conn = _two_project_db(monkeypatch, default_id="ohm")
    assert projects.prune_absent_projects(cfg, {"ohm", "ecomm"}) == []
    assert conn.execute("SELECT COUNT(*) n FROM projects").fetchone()["n"] == 2
    assert conn.execute("SELECT COUNT(*) n FROM cached_rfcs").fetchone()["n"] == 2


def test_prune_never_removes_the_default_project(monkeypatch):
    # Even with an EMPTY present set (which the real caller can't produce —
    # parse_registry rejects an empty registry), the default project survives.
    cfg, conn = _two_project_db(monkeypatch, default_id="ohm")
    pruned = projects.prune_absent_projects(cfg, set())
    assert "ohm" not in pruned and "ecomm" in pruned
    assert conn.execute("SELECT 1 FROM projects WHERE id='ohm'").fetchone() is not None
    assert conn.execute("SELECT 1 FROM collections WHERE id='default'").fetchone() is not None
    assert conn.execute("PRAGMA foreign_key_check").fetchall() == []


def test_empty_registry_is_rejected_so_prune_is_never_reached():
    # The structural safety guard: a parse that would yield no projects raises,
    # so refresh_registry never reaches the prune with an empty present set.
    with pytest.raises(registry.RegistryError):
        registry.parse_registry("deployment:\n  name: x\nprojects: []\n")


def test_incidental_refresh_does_not_prune(app_with_fake_gitea):  # noqa: F811
    """An incidental re-mirror (the registry webhook, prune=False) must NOT prune
    a project absent from projects.yaml — only the full-reconcile triggers
    (startup + periodic sweep) prune. Guards the scoping decision so an in-app
    refresh can't wipe a project."""
    import hashlib
    import hmac
    import json as _json

    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        # A project present in the DB but not in the fake registry's projects.yaml.
        db.conn().execute(
            "INSERT INTO projects (id,name,content_repo,visibility) "
            "VALUES ('ghost','Ghost','ghost-content','public')"
        )
        body = _json.dumps({"repository": {"full_name": "wiggleverse/registry"}}).encode()
        secret = "test-webhook-secret-for-signature-verification"
        sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
        r = client.post(
            "/api/webhooks/gitea",
            content=body,
            headers={"X-Gitea-Event": "push", "X-Gitea-Signature": sig,
                     "Content-Type": "application/json"},
        )
        assert r.status_code == 200
        # The incidental re-mirror left the ghost in place (prune is off here).
        assert db.conn().execute("SELECT 1 FROM projects WHERE id='ghost'").fetchone() is not None


# app_with_fake_gitea fixture import for the integration test above.
from test_propose_vertical import app_with_fake_gitea, tmp_env  # noqa: E402,F401