didericis
0aecb41e33
Phase 4 of PRD 0013. Adds `claude-bottle dashboard` subcommand:
- discover_pending() walks ~/.claude-bottle/queue/* and gathers
pending proposals across all bottles, sorted FIFO by arrival.
- approve / approve-with-final-file / reject helpers write the
Response file the sidecar polls, and append an AuditEntry for
cred-proxy and pipelock tools. capability-block proposals don't
write to an audit log here (PRD 0016 captures via rebuild record).
- Stdlib-curses TUI: list view, detail view, $EDITOR shellout for
modify-then-approve, inline prompt for reject reason.
- `dashboard --once` dumps pending proposals to stdout without
bringing up curses — useful for scripted checks and tests.
For 0013 the audit entry's diff field is render_diff("", proposed)
because we don't yet have access to the live on-disk current file;
PRDs 0014 / 0015 fill in real before→after diffs once they own the
host-side config writes.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
228 lines
8.5 KiB
Python
228 lines
8.5 KiB
Python
"""Unit: dashboard headless paths (PRD 0013 phase 4).
|
|
|
|
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.
|
|
"""
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from claude_bottle import supervise
|
|
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()
|
|
|
|
def tearDown(self):
|
|
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 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()
|