bot-bottle/tests/unit/test_dashboard.py

"""Unit: dashboard headless paths (PRD 0013 phase 4, PRD 0014).

The curses TUI itself isn't exercised here — these tests cover the
discovery + approve/reject + audit-write paths that the TUI's key
handlers call into.

apply_routes_change is stubbed at the dashboard module level so the
tests don't need a running cred-proxy sidecar; the real docker
exec/cp/SIGHUP plumbing is covered by the integration test.
"""

import os
import tempfile
import unittest
from datetime import datetime, timezone
from pathlib import Path

from claude_bottle import supervise
from claude_bottle.backend.docker.cred_proxy_apply import CredProxyApplyError
from claude_bottle.cli import dashboard
from claude_bottle.supervise import (
    Proposal,
    STATUS_APPROVED,
    STATUS_MODIFIED,
    STATUS_REJECTED,
    TOOL_CAPABILITY_BLOCK,
    TOOL_CRED_PROXY_BLOCK,
    TOOL_PIPELOCK_BLOCK,
    read_audit_entries,
    read_response,
    sha256_hex,
)


FIXED = datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc)


def _proposal(slug: str = "dev", tool: str = TOOL_CRED_PROXY_BLOCK) -> Proposal:
    return Proposal.new(
        bottle_slug=slug, tool=tool,
        proposed_file='{"routes": []}\n',
        justification=f"needed for {slug}",
        current_file_hash=sha256_hex("{}"),
        now=FIXED,
    )


class _FakeHomeMixin:
    """Patch supervise.claude_bottle_root to a temp dir for the test."""

    def _setup_fake_home(self):
        self._tmp = tempfile.TemporaryDirectory(prefix="dashboard-test.")
        original = supervise.claude_bottle_root

        def fake_root() -> Path:
            return Path(self._tmp.name) / ".claude-bottle"

        supervise.claude_bottle_root = fake_root  # type: ignore[assignment]
        self._restore_home = lambda: setattr(supervise, "claude_bottle_root", original)

    def _teardown_fake_home(self):
        self._restore_home()
        self._tmp.cleanup()


class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
    def setUp(self):
        self._setup_fake_home()

    def tearDown(self):
        self._teardown_fake_home()

    def test_empty_when_no_queues(self):
        self.assertEqual([], dashboard.discover_pending())

    def test_walks_all_slug_subdirs(self):
        for slug in ("dev", "api"):
            qdir = supervise.queue_dir_for_slug(slug)
            qdir.mkdir(parents=True)
            supervise.write_proposal(qdir, _proposal(slug=slug))
        pending = dashboard.discover_pending()
        self.assertEqual({"dev", "api"}, {qp.proposal.bottle_slug for qp in pending})

    def test_sorted_by_arrival_across_bottles(self):
        early = Proposal.new(
            bottle_slug="api", tool=TOOL_CRED_PROXY_BLOCK,
            proposed_file="{}", justification="early",
            current_file_hash="h",
            now=datetime(2026, 5, 25, 10, 0, 0, tzinfo=timezone.utc),
        )
        late = Proposal.new(
            bottle_slug="dev", tool=TOOL_CRED_PROXY_BLOCK,
            proposed_file="{}", justification="late",
            current_file_hash="h",
            now=datetime(2026, 5, 25, 14, 0, 0, tzinfo=timezone.utc),
        )
        for p in (late, early):
            qdir = supervise.queue_dir_for_slug(p.bottle_slug)
            qdir.mkdir(parents=True, exist_ok=True)
            supervise.write_proposal(qdir, p)
        pending = dashboard.discover_pending()
        self.assertEqual([early.id, late.id], [qp.proposal.id for qp in pending])

    def test_excludes_already_responded(self):
        p = _proposal()
        qdir = supervise.queue_dir_for_slug("dev")
        qdir.mkdir(parents=True)
        supervise.write_proposal(qdir, p)
        supervise.write_response(qdir, supervise.Response(
            proposal_id=p.id, status=STATUS_APPROVED, notes="",
        ))
        self.assertEqual([], dashboard.discover_pending())


class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
    def setUp(self):
        self._setup_fake_home()
        self._original_apply = dashboard.apply_routes_change
        # Default stub: succeed with deterministic before/after so the
        # audit log shows a non-empty diff.
        dashboard.apply_routes_change = lambda slug, content: (
            '{"routes": []}\n', content,
        )

    def tearDown(self):
        dashboard.apply_routes_change = self._original_apply
        self._teardown_fake_home()

    def _enqueue(self, tool: str = TOOL_CRED_PROXY_BLOCK):
        p = _proposal(tool=tool)
        qdir = supervise.queue_dir_for_slug("dev")
        qdir.mkdir(parents=True, exist_ok=True)
        supervise.write_proposal(qdir, p)
        return dashboard.QueuedProposal(proposal=p, queue_dir=qdir)

    def test_approve_writes_response_and_audit(self):
        qp = self._enqueue()
        dashboard.approve(qp)
        resp = read_response(qp.queue_dir, qp.proposal.id)
        self.assertEqual(STATUS_APPROVED, resp.status)
        self.assertIsNone(resp.final_file)
        entries = read_audit_entries("cred-proxy", "dev")
        self.assertEqual(1, len(entries))
        self.assertEqual("approved", entries[0].operator_action)

    def test_approve_with_final_file_marks_modified(self):
        qp = self._enqueue()
        dashboard.approve(qp, final_file='{"routes": [{"path": "/x/"}]}\n', notes="tweaked")
        resp = read_response(qp.queue_dir, qp.proposal.id)
        self.assertEqual(STATUS_MODIFIED, resp.status)
        self.assertEqual('{"routes": [{"path": "/x/"}]}\n', resp.final_file)
        self.assertEqual("tweaked", resp.notes)
        entries = read_audit_entries("cred-proxy", "dev")
        self.assertEqual("modified", entries[0].operator_action)

    def test_reject_writes_rejection(self):
        qp = self._enqueue()
        dashboard.reject(qp, reason="nope")
        resp = read_response(qp.queue_dir, qp.proposal.id)
        self.assertEqual(STATUS_REJECTED, resp.status)
        self.assertEqual("nope", resp.notes)
        entries = read_audit_entries("cred-proxy", "dev")
        self.assertEqual("rejected", entries[0].operator_action)
        self.assertEqual("nope", entries[0].operator_notes)

    def test_capability_block_skips_audit_log(self):
        qp = self._enqueue(tool=TOOL_CAPABILITY_BLOCK)
        dashboard.approve(qp)
        # No audit log for capability-block (per PRD 0013 / 0016).
        # cred-proxy and pipelock logs both empty.
        self.assertEqual([], read_audit_entries("cred-proxy", "dev"))
        self.assertEqual([], read_audit_entries("pipelock", "dev"))

    def test_pipelock_audit_distinct_from_cred_proxy(self):
        qp = self._enqueue(tool=TOOL_PIPELOCK_BLOCK)
        dashboard.approve(qp)
        self.assertEqual(1, len(read_audit_entries("pipelock", "dev")))
        self.assertEqual(0, len(read_audit_entries("cred-proxy", "dev")))


class TestCredProxyApplyWiring(_FakeHomeMixin, unittest.TestCase):
    """PRD 0014 Phase 3: approve() on a cred-proxy-block proposal
    must call apply_routes_change with the right args and surface
    its failures."""

    def setUp(self):
        self._setup_fake_home()
        self._original_apply = dashboard.apply_routes_change

    def tearDown(self):
        dashboard.apply_routes_change = self._original_apply
        self._teardown_fake_home()

    def _enqueue_cred_proxy(self, proposed: str = '{"routes": []}\n'):
        p = Proposal.new(
            bottle_slug="dev", tool=TOOL_CRED_PROXY_BLOCK,
            proposed_file=proposed,
            justification="need a route",
            current_file_hash=sha256_hex(proposed),
            now=FIXED,
        )
        qdir = supervise.queue_dir_for_slug("dev")
        qdir.mkdir(parents=True, exist_ok=True)
        supervise.write_proposal(qdir, p)
        return dashboard.QueuedProposal(proposal=p, queue_dir=qdir)

    def test_cred_proxy_block_calls_apply_with_proposed_file(self):
        calls = []
        dashboard.apply_routes_change = lambda slug, content: (
            calls.append((slug, content)) or ("before", content)
        )
        qp = self._enqueue_cred_proxy(proposed='{"routes": [{"path": "/new/"}]}\n')
        dashboard.approve(qp)
        self.assertEqual(1, len(calls))
        slug, content = calls[0]
        self.assertEqual("dev", slug)
        self.assertEqual('{"routes": [{"path": "/new/"}]}\n', content)

    def test_modify_passes_final_file_to_apply(self):
        calls = []
        dashboard.apply_routes_change = lambda slug, content: (
            calls.append(content) or ("before", content)
        )
        qp = self._enqueue_cred_proxy()
        dashboard.approve(qp, final_file='{"routes": [{"path": "/edited/"}]}\n', notes="tweaked")
        self.assertEqual(['{"routes": [{"path": "/edited/"}]}\n'], calls)

    def test_apply_failure_blocks_response_and_audit(self):
        dashboard.apply_routes_change = lambda slug, content: (_ for _ in ()).throw(
            CredProxyApplyError("docker exec failed")
        )
        qp = self._enqueue_cred_proxy()
        with self.assertRaises(CredProxyApplyError):
            dashboard.approve(qp)
        # No response file (proposal stays pending).
        self.assertEqual(
            [qp.proposal.id],
            [p.id for p in supervise.list_pending_proposals(qp.queue_dir)],
        )
        # No audit entry.
        self.assertEqual([], read_audit_entries("cred-proxy", "dev"))

    def test_real_diff_lands_in_audit(self):
        dashboard.apply_routes_change = lambda slug, content: (
            '{"routes": []}\n',  # before
            '{"routes": [{"path": "/new/"}]}\n',  # after
        )
        qp = self._enqueue_cred_proxy(proposed='{"routes": [{"path": "/new/"}]}\n')
        dashboard.approve(qp)
        entries = read_audit_entries("cred-proxy", "dev")
        self.assertEqual(1, len(entries))
        self.assertIn('+{"routes": [{"path": "/new/"}]}', entries[0].diff)
        self.assertIn('-{"routes": []}', entries[0].diff)

    def test_reject_does_not_call_apply(self):
        called = []
        dashboard.apply_routes_change = lambda slug, content: (
            called.append(True) or ("", content)
        )
        qp = self._enqueue_cred_proxy()
        dashboard.reject(qp, reason="no thanks")
        self.assertEqual([], called)
        # Reject still writes a response + audit entry with empty diff.
        resp = read_response(qp.queue_dir, qp.proposal.id)
        self.assertEqual(STATUS_REJECTED, resp.status)
        entries = read_audit_entries("cred-proxy", "dev")
        self.assertEqual(1, len(entries))
        self.assertEqual("", entries[0].diff)


class TestOperatorEditRoutes(_FakeHomeMixin, unittest.TestCase):
    """PRD 0014 Phase 4: operator-initiated routes edit (not gated
    on a pending proposal)."""

    def setUp(self):
        self._setup_fake_home()
        self._original_apply = dashboard.apply_routes_change

    def tearDown(self):
        dashboard.apply_routes_change = self._original_apply
        self._teardown_fake_home()

    def test_writes_audit_with_operator_edit_action(self):
        dashboard.apply_routes_change = lambda slug, content: (
            '{"routes": []}\n', content,
        )
        dashboard.operator_edit_routes("dev", '{"routes": [{"path": "/x/"}]}\n')
        entries = read_audit_entries("cred-proxy", "dev")
        self.assertEqual(1, len(entries))
        self.assertEqual(supervise.ACTION_OPERATOR_EDIT, entries[0].operator_action)
        self.assertEqual("", entries[0].justification)
        self.assertIn("+", entries[0].diff)

    def test_failure_does_not_write_audit(self):
        dashboard.apply_routes_change = lambda slug, content: (_ for _ in ()).throw(
            CredProxyApplyError("nope")
        )
        with self.assertRaises(CredProxyApplyError):
            dashboard.operator_edit_routes("dev", '{"routes": []}\n')
        self.assertEqual([], read_audit_entries("cred-proxy", "dev"))


class TestDiscoverCredProxySlugs(unittest.TestCase):
    """Slug-extraction parsing — exercises only the parsing path; the
    docker ps invocation itself is environment-dependent (and tested
    implicitly by the integration test)."""

    def test_returns_empty_when_docker_unavailable(self):
        # Force a failure by setting PATH to a dir with no docker
        # binary. The discover helper swallows the non-zero rc.
        import os
        original = os.environ.get("PATH", "")
        os.environ["PATH"] = "/nonexistent-no-docker-here"
        try:
            self.assertEqual([], dashboard.discover_cred_proxy_slugs())
        finally:
            os.environ["PATH"] = original


class TestEditInEditor(unittest.TestCase):
    def test_runs_editor_returns_edited_content(self):
        # Fake "editor" is /bin/sh -c 'cat <<EOF > $1 ... EOF'
        original_editor = os.environ.get("EDITOR")
        try:
            # Use a fake editor that overwrites the file with a known
            # marker. EDITOR is split with shlex equivalence by
            # subprocess.run when invoked as a list — keep it as a
            # single program path that takes the file as argv[1].
            os.environ["EDITOR"] = (
                "/bin/sh -c 'printf %s \"edited\" > \"$0\"'"
            )
            # subprocess.run with the str as the first list element
            # would try to find a binary literally named "/bin/sh -c ..."
            # — that won't work. Use shell mode trick: wrap in a script.
            # Easier: build a tiny helper script.
            with tempfile.NamedTemporaryFile(
                mode="w", suffix=".sh", delete=False, prefix="fake-editor.",
            ) as script:
                script.write('#!/bin/sh\nprintf "%s" "edited" > "$1"\n')
                editor_script = script.name
            os.chmod(editor_script, 0o755)
            os.environ["EDITOR"] = editor_script
            try:
                result = dashboard.edit_in_editor("original")
                self.assertEqual("edited", result)
            finally:
                os.unlink(editor_script)
        finally:
            if original_editor is None:
                os.environ.pop("EDITOR", None)
            else:
                os.environ["EDITOR"] = original_editor

    def test_returns_none_when_unchanged(self):
        original_editor = os.environ.get("EDITOR")
        try:
            # No-op editor: touch the file (leaves it unchanged).
            with tempfile.NamedTemporaryFile(
                mode="w", suffix=".sh", delete=False, prefix="noop-editor.",
            ) as script:
                script.write('#!/bin/sh\n: $1\n')
                editor_script = script.name
            os.chmod(editor_script, 0o755)
            os.environ["EDITOR"] = editor_script
            try:
                result = dashboard.edit_in_editor("original")
                self.assertIsNone(result)
            finally:
                os.unlink(editor_script)
        finally:
            if original_editor is None:
                os.environ.pop("EDITOR", None)
            else:
                os.environ["EDITOR"] = original_editor


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