"""Read-only restart console: views, gates, and honesty rules (#667).

The console consumes the #655 substrate. These tests hold it to the three
properties that make a status surface trustworthy:

* an unreadable source is reported unavailable, never rendered as green;
* authorization is probed the way execution would probe it, so an allow is
  never shown for something that could not run;
* the surface performs no mutation, including no write to the control-plane DB.
"""

from __future__ import annotations

import os
import sqlite3
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  # noqa: E402

import restart_coordinator  # noqa: E402
from webui import console_authz, restart_console, restart_views  # noqa: E402
from webui.app import create_app  # noqa: E402

NOW = datetime(2026, 7, 25, 21, 0, 0, tzinfo=timezone.utc)


def _principal(role: str) -> console_authz.Principal:
    return console_authz.Principal(
        subject="operator@example.com",
        role=role,
        identity_source=console_authz.IDENTITY_LOCAL_DEV,
        authenticated=True,
    )


def _inventory(*, complete: bool = True, sessions=(), leases=()):
    def _read(**_kwargs):
        return {
            "sessions": list(sessions),
            "leases": list(leases),
            "terminal_lock": None,
            "prior_recovery_attempts": [],
            "inventory_complete": complete,
            "incomplete_reasons": (
                [] if complete else ["fixture: inventory withheld"]
            ),
        }

    return _read


def _live_session(session_id: str = "prgs-author-1234-abcd") -> dict:
    return {
        "session_id": session_id,
        "role": "author",
        "profile": "prgs-author",
        "pid": os.getpid(),
        "status": "active",
        "last_heartbeat_at": (NOW - timedelta(seconds=30)).isoformat(),
    }


def drain_proof_fixture() -> dict:
    """A structurally complete but unsigned drain proof."""
    return {
        "version": "drain-proof/v1",
        "proof_id": "deadbeef" * 8,
        "clean": True,
        "issued_at": (NOW - timedelta(minutes=1)).isoformat(),
        "expires_at": (NOW + timedelta(minutes=5)).isoformat(),
        "requesting_session_id": "s-live",
        "impact_fingerprint": "f" * 64,
        "checks": [],
        "failed_checks": [],
    }


class RestartClassMatrixTest(unittest.TestCase):
    def test_every_policy_class_is_rendered(self) -> None:
        views = restart_console.build_restart_class_views("operator")
        self.assertEqual(len(views), len(restart_coordinator.RESTART_CLASS_POLICIES))

    def test_viewer_capability_is_role_scoped_not_generic(self) -> None:
        """A worker role must not be shown as able to request a full restart."""
        author = {
            v.restart_class: v
            for v in restart_console.build_restart_class_views("author")
        }
        operator = {
            v.restart_class: v
            for v in restart_console.build_restart_class_views("operator")
        }
        full = restart_coordinator.RestartClass.FULL_MCP_RESTART.value

        self.assertFalse(author[full].viewer_may_request)
        self.assertFalse(author[full].viewer_may_execute)
        self.assertTrue(operator[full].viewer_may_request)
        self.assertTrue(operator[full].viewer_may_execute)

    def test_unknown_role_may_do_nothing(self) -> None:
        views = restart_console.build_restart_class_views("not-a-role")
        self.assertTrue(all(not v.viewer_may_request for v in views))
        self.assertTrue(all(not v.viewer_may_execute for v in views))


class AuthorizationProbeTest(unittest.TestCase):
    def test_probe_asks_for_execution_so_phase_gate_is_reported(self) -> None:
        """An admin clears the role bar and still cannot execute in Phase 1.

        This is the case that distinguishes the two probes. Asked without
        ``for_execution`` an admin is *allowed* for ``system.restart_namespace``,
        which on a control surface reads as a live button. Asked the way
        execution asks, the same principal is refused ``phase_not_active``. The
        console must report the second answer.
        """
        by_id = {
            a.action_id: a
            for a in restart_console.build_action_authorizations(
                _principal(console_authz.ADMIN)
            )
        }
        restart = by_id["system.restart_namespace"]

        self.assertFalse(restart.execution_enabled)
        self.assertEqual(restart.reason_code, console_authz.DENY_PHASE_NOT_ACTIVE)

        permissive = console_authz.authorize(
            "system.restart_namespace", _principal(console_authz.ADMIN)
        )
        self.assertTrue(
            permissive.allowed,
            "guard precondition: without for_execution an admin is allowed, "
            "which is exactly why the console must not probe that way",
        )

    def test_operator_is_refused_the_admin_only_restart_action(self) -> None:
        """Role refusal precedes the phase gate and is reported as such."""
        by_id = {
            a.action_id: a
            for a in restart_console.build_action_authorizations(
                _principal(console_authz.OPERATOR)
            )
        }
        self.assertEqual(
            by_id["system.restart_namespace"].reason_code,
            console_authz.DENY_INSUFFICIENT_ROLE,
        )

    def test_anonymous_is_denied_unauthenticated(self) -> None:
        by_id = {
            a.action_id: a for a in restart_console.build_action_authorizations(None)
        }
        self.assertEqual(
            by_id["system.restart_namespace"].reason_code,
            console_authz.DENY_UNAUTHENTICATED,
        )

    def test_no_authorization_ever_reports_execution_enabled(self) -> None:
        for role in (
            console_authz.VIEWER,
            console_authz.OPERATOR,
            console_authz.CONTROLLER,
            console_authz.ADMIN,
        ):
            for auth in restart_console.build_action_authorizations(_principal(role)):
                self.assertFalse(
                    auth.execution_enabled,
                    f"{role} reported execution_enabled for {auth.action_id}",
                )


class ImpactPreviewTest(unittest.TestCase):
    def test_impact_renders_from_coordinator_dto(self) -> None:
        impact, source = restart_console.load_impact_report(
            principal=_principal(console_authz.OPERATOR),
            read_inventory=_inventory(sessions=[_live_session()]),
            now=NOW,
        )
        self.assertTrue(source.available)
        self.assertIsNotNone(impact)
        self.assertEqual(
            impact["restart_class"],
            restart_coordinator.RestartClass.FULL_MCP_RESTART.value,
        )
        self.assertIn("verdict", impact)
        self.assertFalse(impact["restart_performed"])
        self.assertTrue(impact["dry_run"])

    def test_incomplete_inventory_is_surfaced_and_denies(self) -> None:
        impact, source = restart_console.load_impact_report(
            principal=_principal(console_authz.OPERATOR),
            read_inventory=_inventory(complete=False),
            now=NOW,
        )
        self.assertFalse(impact["inventory_complete"])
        self.assertFalse(impact["allow_restart"])
        self.assertTrue(source.detail, "incomplete inventory must explain itself")

    def test_inventory_reader_failure_is_unavailable_not_empty(self) -> None:
        """A reader that raises must not be rendered as 'no sessions affected'."""

        def _boom(**_kwargs):
            raise RuntimeError("control-plane unreachable")

        impact, source = restart_console.load_impact_report(
            principal=_principal(console_authz.OPERATOR),
            read_inventory=_boom,
            now=NOW,
        )
        self.assertIsNone(impact)
        self.assertFalse(source.available)
        self.assertIn("control-plane unreachable", source.detail)


class ControlPlaneReadTest(unittest.TestCase):
    def test_missing_database_is_incomplete_not_empty(self) -> None:
        inventory = restart_console.read_control_plane_inventory(
            db_path="/nonexistent/control-plane.sqlite3"
        )
        self.assertFalse(inventory["inventory_complete"])
        self.assertEqual(inventory["sessions"], [])
        self.assertTrue(inventory["incomplete_reasons"])

    def test_reader_never_creates_the_database(self) -> None:
        """Reading status must not bring a control-plane DB into existence.

        The path deliberately sits in a directory that already exists: a
        read-write ``sqlite3.connect`` would happily create the file there, so
        this fails if the reader ever stops opening the database ``mode=ro``.
        A nested-missing-directory path would pass for the wrong reason,
        because sqlite cannot create the parent directory either way.
        """
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "control_plane.sqlite3")
            self.assertTrue(os.path.isdir(os.path.dirname(path)))

            inventory = restart_console.read_control_plane_inventory(db_path=path)

            self.assertFalse(
                os.path.exists(path),
                "reading restart status created a control-plane database",
            )
            self.assertFalse(inventory["inventory_complete"])

    def test_reads_active_sessions_from_a_real_database(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "cp.sqlite3")
            conn = sqlite3.connect(path)
            conn.execute(
                "CREATE TABLE sessions (session_id TEXT, role TEXT, profile TEXT,"
                " pid INTEGER, status TEXT, last_heartbeat_at TEXT)"
            )
            conn.execute(
                "CREATE TABLE work_items (work_item_id INTEGER, kind TEXT,"
                " number INTEGER)"
            )
            conn.execute(
                "CREATE TABLE leases (lease_id TEXT, session_id TEXT, role TEXT,"
                " phase TEXT, status TEXT, worktree_path TEXT,"
                " work_item_id INTEGER, expires_at TEXT)"
            )
            conn.execute(
                "INSERT INTO sessions VALUES (?,?,?,?,?,?)",
                ("s-live", "author", "prgs-author", 4242, "active", NOW.isoformat()),
            )
            conn.execute(
                "INSERT INTO sessions VALUES (?,?,?,?,?,?)",
                ("s-done", "author", "prgs-author", 11, "closed", NOW.isoformat()),
            )
            conn.execute("INSERT INTO work_items VALUES (1, 'issue', 667)")
            conn.execute(
                "INSERT INTO leases VALUES (?,?,?,?,?,?,?,?)",
                (
                    "l-1",
                    "s-live",
                    "author",
                    "allocated",
                    "active",
                    None,
                    1,
                    NOW.isoformat(),
                ),
            )
            conn.commit()
            conn.close()

            inventory = restart_console.read_control_plane_inventory(db_path=path)

        self.assertTrue(inventory["inventory_complete"])
        self.assertEqual([s["session_id"] for s in inventory["sessions"]], ["s-live"])
        self.assertEqual(inventory["leases"][0]["work_number"], 667)


class DrainAndReconcileTest(unittest.TestCase):
    def test_absent_drain_proof_is_not_a_pass(self) -> None:
        drain, source = restart_console.load_drain_status(proof=None, now=NOW)
        self.assertIsNone(drain)
        self.assertFalse(source.available)
        self.assertIn("denies", source.detail)

    def test_tampered_drain_proof_is_reported_invalid(self) -> None:
        proof = drain_proof_fixture()
        proof["clean"] = True
        proof["proof_id"] = "0" * 64
        drain, source = restart_console.load_drain_status(proof=proof, now=NOW)
        self.assertTrue(source.available)
        self.assertFalse(drain["valid"])

    def test_absent_reconcile_proof_is_unavailable(self) -> None:
        reconcile, source = restart_console.load_reconcile_status(load_proof=None)
        self.assertIsNone(reconcile)
        self.assertFalse(source.available)

    def test_reconcile_proof_is_rendered_when_supplied(self) -> None:
        payload = {
            "overall_status": "degraded",
            "mode": "log_only",
            "resolved_count": 3,
            "unresolved_count": 2,
            "items": [
                {
                    "dimension": "leases",
                    "status": "unresolved",
                    "summary": "2 orphaned leases",
                    "follow_up_required": True,
                }
            ],
        }
        reconcile, source = restart_console.load_reconcile_status(
            load_proof=lambda: payload
        )
        self.assertTrue(source.available)
        self.assertEqual(reconcile["unresolved_count"], 2)


class RenderingTest(unittest.TestCase):
    def _snapshot(self, **kwargs):
        params = {
            "principal": _principal(console_authz.OPERATOR),
            "read_inventory": _inventory(sessions=[_live_session()]),
            "now": NOW,
        }
        params.update(kwargs)
        return restart_console.load_restart_console_snapshot(**params)

    def test_page_renders_every_section(self) -> None:
        html = restart_views.render_restart_console_page(self._snapshot())
        for heading in (
            "Impact preview",
            "Drain proof",
            "Post-restart reconcile",
            "Restart classes",
            "Approval controls",
            "Break-glass",
        ):
            self.assertIn(heading, html)

    def test_hostile_session_id_is_escaped(self) -> None:
        hostile = "<script>alert('x')</script>"
        html = restart_views.render_restart_console_page(
            self._snapshot(read_inventory=_inventory(sessions=[_live_session(hostile)]))
        )
        self.assertNotIn("<script>alert", html)
        self.assertIn("&lt;script&gt;", html)

    def test_unavailable_impact_says_unsafe_rather_than_clean(self) -> None:
        def _boom(**_kwargs):
            raise RuntimeError("nope")

        snapshot = self._snapshot(read_inventory=_boom)
        html = restart_views.render_restart_console_page(snapshot)
        self.assertIn("blast radius of a restart is unknown", html)
        self.assertIn("unavailable", html)

    def test_break_glass_is_hidden_from_unprivileged_viewers(self) -> None:
        viewer_html = restart_views.render_restart_console_page(
            self._snapshot(principal=_principal(console_authz.VIEWER))
        )
        self.assertIn("visible to operator-class", viewer_html)
        self.assertNotIn(
            f"#{restart_console.BREAK_GLASS_ISSUE}", viewer_html
        )

    def test_break_glass_shown_to_operator_is_marked_unavailable(self) -> None:
        html = restart_views.render_restart_console_page(self._snapshot())
        self.assertIn("unavailable", html)
        self.assertIn(f"#{restart_console.BREAK_GLASS_ISSUE}", html)

    def test_snapshot_always_declares_itself_read_only(self) -> None:
        self.assertTrue(self._snapshot().read_only)


class RestartConsoleRouteTest(unittest.TestCase):
    def setUp(self) -> None:
        self.client = TestClient(create_app())

    def test_page_route_renders(self) -> None:
        res = self.client.get("/runtime/restart")
        self.assertEqual(res.status_code, 200)
        self.assertIn("Restart status and impact", res.text)

    def test_api_route_exports_snapshot(self) -> None:
        res = self.client.get("/api/v1/system/restart/status")
        self.assertEqual(res.status_code, 200)
        payload = res.json()
        self.assertTrue(payload["read_only"])
        self.assertEqual(payload["links"]["issue"], 667)
        self.assertEqual(
            len(payload["restart_classes"]),
            len(restart_coordinator.RESTART_CLASS_POLICIES),
        )

    def test_restart_class_is_selectable(self) -> None:
        res = self.client.get(
            "/api/v1/system/restart/status?restart_class=client_reconnect"
        )
        self.assertEqual(res.status_code, 200)
        self.assertEqual(res.json()["impact"]["restart_class"], "client_reconnect")

    def test_unknown_restart_class_fails_closed(self) -> None:
        res = self.client.get(
            "/api/v1/system/restart/status?restart_class=obliterate-everything"
        )
        self.assertEqual(res.status_code, 200)
        impact = res.json()["impact"]
        self.assertFalse(impact["allow_restart"])

    def test_anonymous_api_reader_gets_no_execution_grant(self) -> None:
        payload = self.client.get("/api/v1/system/restart/status").json()
        self.assertFalse(payload["break_glass"]["available"])
        for auth in payload["authorizations"]:
            self.assertFalse(auth["execution_enabled"])

    def test_route_is_registered_in_nav(self) -> None:
        from webui.nav import nav_hrefs

        self.assertIn("/runtime/restart", nav_hrefs())

    def test_no_write_method_is_exposed(self) -> None:
        """The surface is read-only: nothing accepts a POST."""
        for path in ("/runtime/restart", "/api/v1/system/restart/status"):
            self.assertEqual(self.client.post(path).status_code, 405, path)


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