Gitea-Tools/tests/test_webui_inventory.py

"""Tests for the unified web-console inventory API (#636).

Covers the four cases the issue names — empty, populated, partial failure, and
the no-false-unowned invariant — plus redaction, collision detection, the
resource-split routes, and read-only guarantees against a real control-plane
database and real durable lock files.
"""
import json
import os
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path

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

from starlette.testclient import TestClient

import control_plane_db
from webui.app import create_app
from webui import inventory


def _iso(dt: datetime) -> str:
    return dt.astimezone(timezone.utc).isoformat()


def _write_lock(lock_dir: str, name: str, payload: dict) -> str:
    path = os.path.join(lock_dir, name)
    with open(path, "w", encoding="utf-8") as handle:
        json.dump(payload, handle)
    return path


def _live_lock_payload(
    *,
    issue_number: int,
    branch: str,
    worktree_path: str,
    pid: int,
    username: str = "jcwalker3",
    profile: str = "prgs-author",
) -> dict:
    now = datetime.now(timezone.utc)
    future = now + timedelta(hours=2)
    return {
        "branch_name": branch,
        "issue_number": issue_number,
        "org": "Scaled-Tech-Consulting",
        "repo": "Gitea-Tools",
        "remote": "prgs",
        "pid": pid,
        "session_pid": pid,
        "lock_generation": 1,
        "worktree_path": worktree_path,
        "claimant": {"username": username, "profile": profile},
        "work_lease": {
            "branch": branch,
            "issue_number": issue_number,
            "operation_type": "author_issue_work",
            "created_at": _iso(now),
            "expires_at": _iso(future),
            "last_heartbeat_at": _iso(now),
            "claimant": {"username": username, "profile": profile},
        },
    }


class _FixtureMixin(unittest.TestCase):
    def setUp(self) -> None:
        self._tmp = tempfile.TemporaryDirectory()
        self.tmp = self._tmp.name
        self.lock_dir = os.path.join(self.tmp, "locks")
        os.makedirs(self.lock_dir, mode=0o700)
        self.db_path = os.path.join(self.tmp, "control_plane.db")
        self.addCleanup(self._tmp.cleanup)

    def _seed_db(self) -> control_plane_db.ControlPlaneDB:
        db = control_plane_db.ControlPlaneDB(self.db_path)
        db.upsert_session(
            session_id="prgs-author-1",
            role="author",
            profile="prgs-author",
            namespace="gitea-author",
            pid=os.getpid(),
        )
        db.upsert_work_item(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            kind="issue",
            number=636,
        )
        db.assign_and_lease(
            session_id="prgs-author-1",
            role="author",
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            kind="issue",
            number=636,
        )
        return db


class TestRedaction(unittest.TestCase):
    def test_redact_path_collapses_home(self):
        home = os.path.expanduser("~")
        self.assertEqual(
            inventory.redact_path(f"{home}/Development/Gitea-Tools"),
            "~/Development/Gitea-Tools",
        )

    def test_redact_url_strips_userinfo_and_query(self):
        self.assertEqual(
            inventory.redact_url("https://user:[email protected]/api?token=abc"),
            "https://gitea.prgs.cc/api",
        )

    def test_scrub_drops_credential_keys(self):
        scrubbed = inventory.scrub(
            {"token": "abc123", "api_key": "k", "profile": "prgs-author"}
        )
        self.assertEqual(scrubbed["token"], "[redacted]")
        self.assertEqual(scrubbed["api_key"], "[redacted]")
        self.assertEqual(scrubbed["profile"], "prgs-author")

    def test_scrub_is_recursive_and_never_raises(self):
        class Weird:
            def __repr__(self) -> str:
                return "weird-obj"

        out = inventory.scrub({"nested": [{"password": "p", "obj": Weird()}]})
        self.assertEqual(out["nested"][0]["password"], "[redacted]")
        self.assertEqual(out["nested"][0]["obj"], "weird-obj")


class TestEmptyInventory(_FixtureMixin):
    def test_empty_db_and_locks_degrade_without_raising(self):
        # No DB file, no locks: sessions/leases unavailable, locks ok+empty.
        snap = inventory.load_inventory_snapshot(
            db_path=self.db_path,
            lock_dir=self.lock_dir,
            load_hygiene=lambda: _StubHygiene(entries=()),
        )
        sessions = snap.section("sessions")
        leases = snap.section("leases")
        locks = snap.section("locks")
        self.assertEqual(sessions.status, inventory.STATUS_UNAVAILABLE)
        self.assertEqual(leases.status, inventory.STATUS_UNAVAILABLE)
        self.assertEqual(locks.status, inventory.STATUS_OK)
        self.assertEqual(len(locks.items), 0)
        # Ownership authority is incomplete because the DB is missing.
        self.assertFalse(snap.ownership_authority_complete)
        self.assertEqual(snap.collisions, ())

    def test_empty_db_present_but_unpopulated(self):
        control_plane_db.ControlPlaneDB(self.db_path)  # creates schema, no rows
        snap = inventory.load_inventory_snapshot(
            db_path=self.db_path,
            lock_dir=self.lock_dir,
            load_hygiene=lambda: _StubHygiene(entries=()),
        )
        self.assertEqual(snap.section("sessions").status, inventory.STATUS_OK)
        self.assertEqual(len(snap.section("sessions").items), 0)
        self.assertEqual(snap.section("leases").status, inventory.STATUS_OK)
        self.assertTrue(snap.ownership_authority_complete)


class TestPopulatedInventory(_FixtureMixin):
    def test_sections_populated_and_correlated(self):
        self._seed_db()
        wt = f"{self.tmp}/branches/issue-636-inventory-api"
        _write_lock(
            self.lock_dir,
            "prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
            _live_lock_payload(
                issue_number=636,
                branch="feat/issue-636-inventory-api",
                worktree_path=wt,
                pid=os.getpid(),
            ),
        )
        hygiene = _StubHygiene(
            entries=(
                _StubEntry(
                    rel_path="branches/issue-636-inventory-api",
                    branch="feat/issue-636-inventory-api",
                    classification="active-issue",
                ),
            )
        )
        snap = inventory.load_inventory_snapshot(
            db_path=self.db_path,
            lock_dir=self.lock_dir,
            load_hygiene=lambda: hygiene,
        )
        self.assertTrue(snap.ownership_authority_complete)
        self.assertEqual(len(snap.section("sessions").items), 1)
        self.assertEqual(len(snap.section("leases").items), 1)
        self.assertEqual(len(snap.section("locks").items), 1)
        self.assertEqual(len(snap.section("worktrees").items), 1)

        # The lease, lock, and worktree for #636 correlate onto one row.
        row = next(r for r in snap.correlations if r["issue_number"] == 636)
        self.assertEqual(row["branch"], "feat/issue-636-inventory-api")
        self.assertTrue(row["lock_live"])
        self.assertEqual(row["worktree_classification"], "active-issue")
        self.assertEqual(len(row["lease_ids"]), 1)
        # No collision: live lock, live pid, matching worktree.
        self.assertEqual(snap.collisions, ())

    def test_serialized_payload_declares_field_authority(self):
        self._seed_db()
        snap = inventory.load_inventory_snapshot(
            db_path=self.db_path,
            lock_dir=self.lock_dir,
            load_hygiene=lambda: _StubHygiene(entries=()),
        )
        payload = inventory.snapshot_to_dict(snap)
        self.assertEqual(payload["api_version"], "v1")
        self.assertEqual(payload["schema_version"], 1)
        self.assertEqual(payload["field_authority"]["sessions"], "control_plane_db")
        self.assertEqual(payload["field_authority"]["locks"], "filesystem")
        self.assertIn("sessions", payload["sections"])


class TestPartialFailure(_FixtureMixin):
    def test_worktree_scan_failure_degrades_only_that_section(self):
        self._seed_db()

        def _boom():
            raise RuntimeError("git worktree list exploded")

        snap = inventory.load_inventory_snapshot(
            db_path=self.db_path,
            lock_dir=self.lock_dir,
            load_hygiene=_boom,
        )
        self.assertEqual(
            snap.section("worktrees").status, inventory.STATUS_UNAVAILABLE
        )
        self.assertIn("exploded", snap.section("worktrees").reason)
        # DB-backed sections still healthy.
        self.assertEqual(snap.section("sessions").status, inventory.STATUS_OK)
        self.assertIn("worktrees", snap.degraded_sections)

    def test_degraded_ownership_suppresses_unowned_claim(self):
        # DB absent → sessions/leases unavailable → ownership incomplete even
        # though a lock exists and could look "unclaimed" by the DB alone.
        _write_lock(
            self.lock_dir,
            "prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
            _live_lock_payload(
                issue_number=636,
                branch="feat/issue-636-inventory-api",
                worktree_path=f"{self.tmp}/wt",
                pid=os.getpid(),
            ),
        )
        snap = inventory.load_inventory_snapshot(
            db_path=self.db_path,
            lock_dir=self.lock_dir,
            load_hygiene=lambda: _StubHygiene(entries=()),
        )
        self.assertFalse(snap.ownership_authority_complete)
        payload = inventory.snapshot_to_dict(snap)
        self.assertIn("may be treated as unowned", payload["ownership_note"])


class TestCollisionDetection(_FixtureMixin):
    def test_live_lock_dead_owner_flagged(self):
        _write_lock(
            self.lock_dir,
            "prgs-Scaled-Tech-Consulting-Gitea-Tools-700.json",
            _live_lock_payload(
                issue_number=700,
                branch="feat/issue-700-x",
                worktree_path=f"{self.tmp}/wt700",
                pid=999_999_999,  # not a running pid
            ),
        )
        control_plane_db.ControlPlaneDB(self.db_path)  # empty but present
        snap = inventory.load_inventory_snapshot(
            db_path=self.db_path,
            lock_dir=self.lock_dir,
            load_hygiene=lambda: _StubHygiene(entries=()),
        )
        kinds = {c.kind for c in snap.collisions}
        self.assertIn("live-lock-dead-owner", kinds)
        # Also lock-without-worktree, since no worktree carries the branch.
        self.assertIn("lock-without-worktree", kinds)

    def test_duplicate_live_lock_on_same_branch(self):
        for issue in (800, 801):
            _write_lock(
                self.lock_dir,
                f"prgs-Scaled-Tech-Consulting-Gitea-Tools-{issue}.json",
                _live_lock_payload(
                    issue_number=issue,
                    branch="feat/issue-800-shared",
                    worktree_path=f"{self.tmp}/wt{issue}",
                    pid=os.getpid(),
                ),
            )
        control_plane_db.ControlPlaneDB(self.db_path)
        snap = inventory.load_inventory_snapshot(
            db_path=self.db_path,
            lock_dir=self.lock_dir,
            load_hygiene=lambda: _StubHygiene(entries=()),
        )
        self.assertIn(
            "duplicate-live-lock", {c.kind for c in snap.collisions}
        )

    def test_no_collision_when_sections_degraded(self):
        # locks ok but worktrees unavailable → lock-without-worktree must NOT
        # be asserted (a missing scan is not a missing worktree).
        _write_lock(
            self.lock_dir,
            "prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
            _live_lock_payload(
                issue_number=636,
                branch="feat/issue-636-inventory-api",
                worktree_path=f"{self.tmp}/wt",
                pid=os.getpid(),
            ),
        )
        control_plane_db.ControlPlaneDB(self.db_path)

        def _boom():
            raise RuntimeError("scan down")

        snap = inventory.load_inventory_snapshot(
            db_path=self.db_path,
            lock_dir=self.lock_dir,
            load_hygiene=_boom,
        )
        self.assertNotIn(
            "lock-without-worktree", {c.kind for c in snap.collisions}
        )


class TestSectionInclude(_FixtureMixin):
    def test_include_restricts_scanned_sections(self):
        self._seed_db()
        snap = inventory.load_inventory_snapshot(
            db_path=self.db_path,
            lock_dir=self.lock_dir,
            include=("locks",),
        )
        self.assertIsNotNone(snap.section("locks"))
        self.assertIsNone(snap.section("sessions"))
        self.assertIsNone(snap.section("worktrees"))


class TestRoutes(_FixtureMixin):
    def setUp(self) -> None:
        super().setUp()
        # Point the loaders at the fixture DB and lock dir via env, and stub
        # the worktree scan so the route does not shell out to git.
        self._prev_env = {
            "GITEA_CONTROL_PLANE_DB": os.environ.get("GITEA_CONTROL_PLANE_DB"),
            "GITEA_ISSUE_LOCK_DIR": os.environ.get("GITEA_ISSUE_LOCK_DIR"),
            "WEBUI_TEST_OFFLINE": os.environ.get("WEBUI_TEST_OFFLINE"),
        }
        os.environ["GITEA_CONTROL_PLANE_DB"] = self.db_path
        os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir
        os.environ["WEBUI_TEST_OFFLINE"] = "1"
        self._seed_db()
        self.client = TestClient(create_app())

    def tearDown(self) -> None:
        for key, value in self._prev_env.items():
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value

    def test_inventory_route_returns_versioned_payload(self):
        resp = self.client.get("/api/v1/inventory")
        self.assertEqual(resp.status_code, 200)
        body = resp.json()
        self.assertEqual(body["api_version"], "v1")
        self.assertIn("sessions", body["sections"])
        self.assertIn("field_authority", body)

    def test_section_route_restricts_and_labels(self):
        resp = self.client.get("/api/v1/inventory/locks")
        self.assertEqual(resp.status_code, 200)
        body = resp.json()
        self.assertEqual(body["requested_section"], "locks")
        self.assertIn("locks", body["sections"])
        self.assertNotIn("sessions", body["sections"])

    def test_unknown_section_is_404(self):
        resp = self.client.get("/api/v1/inventory/bogus")
        self.assertEqual(resp.status_code, 404)
        self.assertEqual(resp.json()["error"], "unknown_section")

    def test_inventory_route_rejects_post(self):
        resp = self.client.post("/api/v1/inventory")
        self.assertEqual(resp.status_code, 405)


class TestReadOnly(_FixtureMixin):
    def test_snapshot_does_not_create_db_file(self):
        missing = os.path.join(self.tmp, "does-not-exist.db")
        inventory.load_inventory_snapshot(
            db_path=missing,
            lock_dir=self.lock_dir,
            load_hygiene=lambda: _StubHygiene(entries=()),
        )
        self.assertFalse(os.path.exists(missing))

    def test_readonly_connection_refuses_write(self):
        self._seed_db()
        conn = inventory._open_readonly(self.db_path)
        try:
            with self.assertRaises(Exception):
                conn.execute(
                    "INSERT INTO sessions(session_id, role, started_at, "
                    "last_heartbeat_at, status) VALUES ('x','author',"
                    "'t','t','active')"
                )
                conn.commit()
        finally:
            conn.close()


# ── lightweight stand-ins for the #432 hygiene snapshot ──────────────────────


class _StubEntry:
    def __init__(
        self,
        *,
        rel_path: str,
        branch: str | None = None,
        classification: str = "stale-clean",
        head_sha: str | None = "abc123",
        dirty_tracked: int = 0,
        dirty_untracked: bool = False,
        detached: bool = False,
        registered_worktree: bool = True,
        notes: str = "",
    ) -> None:
        self.rel_path = rel_path
        self.folder_name = rel_path.split("/", 1)[-1]
        self.branch = branch
        self.classification = classification
        self.head_sha = head_sha
        self.dirty_tracked = dirty_tracked
        self.dirty_untracked = dirty_untracked
        self.detached = detached
        self.registered_worktree = registered_worktree
        self.notes = notes


class _StubHygiene:
    def __init__(self, *, entries=(), scan_error=None) -> None:
        self.entries = tuple(entries)
        self.scan_error = scan_error


if __name__ == "__main__":
    unittest.main()