test(cli): update supervise triage coverage
This commit is contained in:
@@ -277,51 +277,5 @@ class TestBottleMetadataBackend(_FakeHomeMixin, unittest.TestCase):
|
||||
self.assertEqual("", loaded.backend)
|
||||
|
||||
|
||||
class TestBottleForSlugBackend(_FakeHomeMixin, unittest.TestCase):
|
||||
"""PRD 0040: _bottle_for_slug constructs the right bottle type."""
|
||||
|
||||
def setUp(self):
|
||||
self._setup_fake_home()
|
||||
|
||||
def tearDown(self):
|
||||
self._teardown_fake_home()
|
||||
|
||||
def test_docker_metadata_returns_docker_bottle(self):
|
||||
from bot_bottle.backend.docker.bottle import DockerBottle
|
||||
from bot_bottle.cli.dashboard import _bottle_for_slug
|
||||
write_metadata(BottleMetadata(
|
||||
identity="dev-d1",
|
||||
agent_name="dev",
|
||||
cwd="",
|
||||
copy_cwd=False,
|
||||
started_at="2026-06-02T00:00:00+00:00",
|
||||
compose_project="bot-bottle-dev-d1",
|
||||
backend="docker",
|
||||
))
|
||||
bottle, _ = _bottle_for_slug("dev-d1", {}, None)
|
||||
self.assertIsInstance(bottle, DockerBottle)
|
||||
|
||||
def test_smolmachines_metadata_returns_smolmachines_bottle(self):
|
||||
from bot_bottle.backend.smolmachines.bottle import SmolmachinesBottle
|
||||
from bot_bottle.cli.dashboard import _bottle_for_slug
|
||||
write_metadata(BottleMetadata(
|
||||
identity="dev-s1",
|
||||
agent_name="dev",
|
||||
cwd="",
|
||||
copy_cwd=False,
|
||||
started_at="2026-06-02T00:00:00+00:00",
|
||||
compose_project="",
|
||||
backend="smolmachines",
|
||||
))
|
||||
bottle, _ = _bottle_for_slug("dev-s1", {}, None)
|
||||
self.assertIsInstance(bottle, SmolmachinesBottle)
|
||||
|
||||
def test_no_metadata_defaults_to_docker_bottle(self):
|
||||
from bot_bottle.backend.docker.bottle import DockerBottle
|
||||
from bot_bottle.cli.dashboard import _bottle_for_slug
|
||||
bottle, _ = _bottle_for_slug("unknown-slug", {}, None)
|
||||
self.assertIsInstance(bottle, DockerBottle)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -1,492 +0,0 @@
|
||||
"""Unit: dashboard's row-formatting + selection helpers (PRD 0019)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
from bot_bottle import supervise
|
||||
from bot_bottle.cli import dashboard
|
||||
|
||||
|
||||
class _FakeHomeMixin:
|
||||
def _setup_fake_home(self) -> None:
|
||||
self._tmp = tempfile.TemporaryDirectory(prefix="dashboard-aa-test.")
|
||||
original = supervise.bot_bottle_root
|
||||
|
||||
def fake_root() -> Path:
|
||||
return Path(self._tmp.name) / ".bot-bottle"
|
||||
|
||||
supervise.bot_bottle_root = fake_root # type: ignore[assignment]
|
||||
self._restore_home = lambda: setattr(supervise, "bot_bottle_root", original)
|
||||
|
||||
def _teardown_fake_home(self) -> None:
|
||||
self._restore_home()
|
||||
self._tmp.cleanup()
|
||||
|
||||
|
||||
class TestFormatAgentRow(unittest.TestCase):
|
||||
"""One-line row formatting for the agents pane (PRD 0019 chunk 2)."""
|
||||
|
||||
def _agent(self, **overrides) -> dashboard.ActiveAgent:
|
||||
defaults = dict(
|
||||
backend_name="docker",
|
||||
slug="dev-abc12",
|
||||
agent_name="implementer",
|
||||
started_at="2026-05-26T02:55:01+00:00",
|
||||
services=("egress", "git-gate", "pipelock", "supervise"),
|
||||
)
|
||||
defaults.update(overrides)
|
||||
return dashboard.ActiveAgent(**defaults)
|
||||
|
||||
def test_renders_slug_name_time_services(self):
|
||||
s = dashboard._format_agent_row(self._agent(), 200)
|
||||
self.assertIn("dev-abc12", s)
|
||||
self.assertIn("implementer", s)
|
||||
self.assertIn("02:55:01", s)
|
||||
self.assertIn("egress,git-gate,pipelock,supervise", s)
|
||||
|
||||
def test_starting_label_when_no_services(self):
|
||||
# Race window: compose project is up but containers haven't
|
||||
# been picked up by `docker ps` yet.
|
||||
s = dashboard._format_agent_row(self._agent(services=()), 200)
|
||||
self.assertIn("(starting)", s)
|
||||
|
||||
def test_filters_agent_service_from_display(self):
|
||||
# The `agent` service is always present for an active bottle;
|
||||
# listing it is noise. The row should show only the sidecars.
|
||||
s = dashboard._format_agent_row(
|
||||
self._agent(services=("agent", "pipelock", "supervise")), 200,
|
||||
)
|
||||
self.assertIn("[pipelock,supervise]", s)
|
||||
self.assertNotIn("agent,", s)
|
||||
self.assertNotIn(",agent", s)
|
||||
|
||||
def test_only_agent_service_shows_starting(self):
|
||||
# A bottle whose only running service is `agent` (sidecars
|
||||
# still warming up) renders as `(starting)`.
|
||||
s = dashboard._format_agent_row(self._agent(services=("agent",)), 200)
|
||||
self.assertIn("(starting)", s)
|
||||
|
||||
def test_question_mark_when_no_started_at(self):
|
||||
s = dashboard._format_agent_row(self._agent(started_at=""), 200)
|
||||
self.assertIn("started ?", s)
|
||||
|
||||
def test_truncates_to_maxw(self):
|
||||
s = dashboard._format_agent_row(self._agent(), 30)
|
||||
self.assertLessEqual(len(s), 30)
|
||||
self.assertTrue(s.endswith("…"))
|
||||
|
||||
|
||||
class TestSelectionStatus(unittest.TestCase):
|
||||
"""Idle-state status-line text for the agents-pane focus
|
||||
(PRD 0019 chunk 3). Empty when the proposals pane is focused;
|
||||
surfaces the selected agent (or a clear placeholder) when the
|
||||
agents pane is focused."""
|
||||
|
||||
def _agent(self, slug: str) -> dashboard.ActiveAgent:
|
||||
return dashboard.ActiveAgent(
|
||||
backend_name="docker",
|
||||
slug=slug, agent_name="x", started_at="", services=(),
|
||||
)
|
||||
|
||||
def test_empty_when_proposals_focused(self):
|
||||
s = dashboard._selection_status(
|
||||
dashboard.PANE_PROPOSALS, [self._agent("a-1")], 0,
|
||||
)
|
||||
self.assertEqual("", s)
|
||||
|
||||
def test_no_agents_message_when_agents_pane_empty(self):
|
||||
s = dashboard._selection_status(dashboard.PANE_AGENTS, [], 0)
|
||||
self.assertEqual("[no active agents]", s)
|
||||
|
||||
def test_shows_selected_slug(self):
|
||||
agents = [self._agent("a-1"), self._agent("b-2"), self._agent("c-3")]
|
||||
s = dashboard._selection_status(dashboard.PANE_AGENTS, agents, 1)
|
||||
self.assertEqual("[selected: b-2]", s)
|
||||
|
||||
def test_out_of_bounds_falls_back_to_no_selection(self):
|
||||
agents = [self._agent("only")]
|
||||
s = dashboard._selection_status(dashboard.PANE_AGENTS, agents, 99)
|
||||
self.assertEqual("[no agent selected]", s)
|
||||
|
||||
|
||||
class TestFilterAgents(unittest.TestCase):
|
||||
"""Pure-function picker filter (PRD 0020 chunk 2). Curses-free
|
||||
so we can exercise the substring + case-insensitivity rules
|
||||
directly."""
|
||||
|
||||
NAMES = ["implementer", "researcher", "triage-bot", "ImplDeluxe"]
|
||||
|
||||
def test_empty_query_returns_all(self):
|
||||
self.assertEqual(self.NAMES, dashboard._filter_agents("", self.NAMES))
|
||||
|
||||
def test_substring_match(self):
|
||||
self.assertEqual(
|
||||
["implementer", "ImplDeluxe"],
|
||||
dashboard._filter_agents("impl", self.NAMES),
|
||||
)
|
||||
|
||||
def test_case_insensitive(self):
|
||||
self.assertEqual(
|
||||
["implementer", "ImplDeluxe"],
|
||||
dashboard._filter_agents("IMPL", self.NAMES),
|
||||
)
|
||||
|
||||
def test_no_match_returns_empty(self):
|
||||
self.assertEqual([], dashboard._filter_agents("zzz", self.NAMES))
|
||||
|
||||
def test_preserves_input_order(self):
|
||||
# Filtering should never re-sort; the picker draws in the
|
||||
# order the manifest exposed.
|
||||
out = dashboard._filter_agents("e", ["beta", "alpha", "echo"])
|
||||
self.assertEqual(["beta", "echo"], out)
|
||||
|
||||
|
||||
class TestDashboardManifestLoading(unittest.TestCase):
|
||||
def test_new_agent_flow_empty_manifest_has_no_picker_entries(self):
|
||||
manifest = dashboard.Manifest.from_json_obj({"bottles": {}, "agents": {}})
|
||||
with mock.patch("bot_bottle.cli.dashboard._picker_modal", return_value=None) as picker:
|
||||
status = dashboard._new_agent_flow(
|
||||
None, manifest, {}, [], tmux_state=None, # type: ignore[arg-type]
|
||||
)
|
||||
picker.assert_called_once()
|
||||
self.assertEqual([], picker.call_args.args[1])
|
||||
self.assertIn("no agents configured", status)
|
||||
|
||||
|
||||
class TestRunningCounts(unittest.TestCase):
|
||||
"""Per-agent running-count surfaced in the picker so the
|
||||
operator sees `(N running)` before picking. Counts come from
|
||||
the dashboard's current `discover_active_agents` snapshot."""
|
||||
|
||||
def _agent(self, agent_name: str) -> dashboard.ActiveAgent:
|
||||
return dashboard.ActiveAgent(
|
||||
backend_name="docker",
|
||||
slug=f"{agent_name}-abc",
|
||||
agent_name=agent_name,
|
||||
started_at="",
|
||||
services=(),
|
||||
)
|
||||
|
||||
def test_empty_when_no_active_agents(self):
|
||||
self.assertEqual({}, dashboard._running_counts({}, []))
|
||||
|
||||
def test_one_per_unique_agent_name(self):
|
||||
agents = [self._agent("a"), self._agent("b"), self._agent("c")]
|
||||
self.assertEqual(
|
||||
{"a": 1, "b": 1, "c": 1},
|
||||
dashboard._running_counts({}, agents),
|
||||
)
|
||||
|
||||
def test_counts_collisions(self):
|
||||
agents = [
|
||||
self._agent("implementer"),
|
||||
self._agent("implementer"),
|
||||
self._agent("researcher"),
|
||||
]
|
||||
self.assertEqual(
|
||||
{"implementer": 2, "researcher": 1},
|
||||
dashboard._running_counts({}, agents),
|
||||
)
|
||||
|
||||
|
||||
class TestSelectedAgent(unittest.TestCase):
|
||||
"""`_selected_agent` is what chunk 4's e/p key handlers use to
|
||||
decide whether to fire and which agent to target."""
|
||||
|
||||
def _agent(self, slug: str, services: tuple[str, ...] = ()) -> dashboard.ActiveAgent:
|
||||
return dashboard.ActiveAgent(
|
||||
backend_name="docker",
|
||||
slug=slug, agent_name="x", started_at="", services=services,
|
||||
)
|
||||
|
||||
def test_none_when_proposals_focused(self):
|
||||
agents = [self._agent("a-1")]
|
||||
self.assertIsNone(
|
||||
dashboard._selected_agent(dashboard.PANE_PROPOSALS, agents, 0),
|
||||
)
|
||||
|
||||
def test_none_when_no_agents(self):
|
||||
self.assertIsNone(
|
||||
dashboard._selected_agent(dashboard.PANE_AGENTS, [], 0),
|
||||
)
|
||||
|
||||
def test_returns_indexed_agent_when_in_range(self):
|
||||
agents = [self._agent("a-1"), self._agent("b-2")]
|
||||
result = dashboard._selected_agent(dashboard.PANE_AGENTS, agents, 1)
|
||||
self.assertIsNotNone(result)
|
||||
assert result is not None # for type checker
|
||||
self.assertEqual("b-2", result.slug)
|
||||
|
||||
def test_none_when_index_out_of_range(self):
|
||||
agents = [self._agent("only")]
|
||||
self.assertIsNone(
|
||||
dashboard._selected_agent(dashboard.PANE_AGENTS, agents, 99),
|
||||
)
|
||||
|
||||
|
||||
class TestBottleForSlug(unittest.TestCase):
|
||||
"""Re-attach target resolution (PRD 0020 chunk 3). Dashboard-
|
||||
owned bottles return the stored handle as-is; non-owned bottles
|
||||
get a synthesized DockerBottle backed by the slug-derived
|
||||
container name."""
|
||||
|
||||
def test_owned_bottle_returns_held_handle(self):
|
||||
sentinel = object()
|
||||
bottles = {"dev-abc": (None, sentinel, "dev-abc")}
|
||||
bottle, _ = dashboard._bottle_for_slug("dev-abc", bottles, None)
|
||||
self.assertIs(sentinel, bottle)
|
||||
|
||||
def test_unowned_synthesizes_docker_bottle(self):
|
||||
bottle, _ = dashboard._bottle_for_slug("dev-xyz", {}, None)
|
||||
# The synth wraps the slug-derived container name.
|
||||
self.assertEqual("bot-bottle-dev-xyz", bottle.name)
|
||||
|
||||
def test_unowned_without_manifest_omits_prompt_path(self):
|
||||
bottle, hint = dashboard._bottle_for_slug("dev-xyz", {}, None)
|
||||
self.assertEqual("", hint)
|
||||
|
||||
|
||||
class TestPickNextAfterStop(unittest.TestCase):
|
||||
"""After `x` stops a bottle, the dashboard slides focus to
|
||||
the next agent — the one filling the stopped row, or the
|
||||
new last row if the stopped was last. Pure helper, easy
|
||||
to unit-test."""
|
||||
|
||||
def _agent(self, slug: str) -> dashboard.ActiveAgent:
|
||||
return dashboard.ActiveAgent(
|
||||
backend_name="docker",
|
||||
slug=slug, agent_name=slug, started_at="", services=(),
|
||||
)
|
||||
|
||||
def test_empty_list_returns_none(self):
|
||||
self.assertIsNone(
|
||||
dashboard._pick_next_after_stop([], 0, "anything"),
|
||||
)
|
||||
|
||||
def test_only_agent_being_stopped_returns_none(self):
|
||||
# Stopping the last agent → nothing to focus.
|
||||
agents = [self._agent("only")]
|
||||
self.assertIsNone(
|
||||
dashboard._pick_next_after_stop(agents, 0, "only"),
|
||||
)
|
||||
|
||||
def test_middle_row_slides_up_to_same_index(self):
|
||||
agents = [self._agent("a"), self._agent("b"), self._agent("c")]
|
||||
# Cursor was on "b" at index 1; stopping "b" → "c" now sits
|
||||
# at index 1 and takes focus.
|
||||
out = dashboard._pick_next_after_stop(agents, 1, "b")
|
||||
self.assertEqual((1, self._agent("c")), out)
|
||||
|
||||
def test_last_row_wraps_to_new_last(self):
|
||||
agents = [self._agent("a"), self._agent("b"), self._agent("c")]
|
||||
# Cursor on "c" at index 2; stopping "c" leaves a 2-agent
|
||||
# list — index 2 is out of bounds, fall back to new last (1).
|
||||
out = dashboard._pick_next_after_stop(agents, 2, "c")
|
||||
self.assertEqual((1, self._agent("b")), out)
|
||||
|
||||
def test_first_row(self):
|
||||
agents = [self._agent("a"), self._agent("b")]
|
||||
out = dashboard._pick_next_after_stop(agents, 0, "a")
|
||||
self.assertEqual((0, self._agent("b")), out)
|
||||
|
||||
def test_clamps_negative_selection(self):
|
||||
# Defensive: a stale negative index doesn't crash.
|
||||
agents = [self._agent("a"), self._agent("b")]
|
||||
out = dashboard._pick_next_after_stop(agents, -1, "a")
|
||||
self.assertEqual((0, self._agent("b")), out)
|
||||
|
||||
|
||||
class TestTmuxPaneArgvBuilders(unittest.TestCase):
|
||||
"""Pure argv builders for the tmux split-pane integration
|
||||
(PRD 0021 chunk 2). The subprocess invocation itself is
|
||||
environment-dependent; here we lock the wrapping shape so
|
||||
a regression surfaces in CI without needing a real tmux."""
|
||||
|
||||
DOCKER_ARGV = [
|
||||
"docker", "exec", "-it",
|
||||
"bot-bottle-dev-abc",
|
||||
"claude", "--dangerously-skip-permissions", "--continue",
|
||||
]
|
||||
|
||||
def test_split_pane_argv_horizontal_with_pane_id_capture(self):
|
||||
argv = dashboard._build_split_pane_argv(self.DOCKER_ARGV)
|
||||
self.assertEqual(
|
||||
["tmux", "split-window", "-h",
|
||||
"-P", "-F", "#{pane_id}",
|
||||
*self.DOCKER_ARGV],
|
||||
argv,
|
||||
)
|
||||
|
||||
def test_respawn_pane_argv_kills_existing_process(self):
|
||||
argv = dashboard._build_respawn_pane_argv("%12", self.DOCKER_ARGV)
|
||||
self.assertEqual(
|
||||
["tmux", "respawn-pane", "-k", "-t", "%12", *self.DOCKER_ARGV],
|
||||
argv,
|
||||
)
|
||||
|
||||
def test_respawn_pane_argv_threads_pane_id_unmodified(self):
|
||||
# Pane ids contain `%`; make sure we pass them straight
|
||||
# through to `-t` without quoting or substitution surprises.
|
||||
argv = dashboard._build_respawn_pane_argv("%abc.123", ["sh"])
|
||||
self.assertIn("%abc.123", argv)
|
||||
|
||||
|
||||
class TestResumeArgvWithFallback(unittest.TestCase):
|
||||
"""The `claude --continue || claude` shell fallback for the
|
||||
tmux re-attach path. Without it, an agent that's been spun
|
||||
up but never typed at crashes the pane on Enter because
|
||||
--continue has no session to resume."""
|
||||
|
||||
def _bottle(self, prompt_path: str | None = None):
|
||||
from bot_bottle.backend.docker.bottle import DockerBottle
|
||||
return DockerBottle(
|
||||
container="bot-bottle-dev-abc",
|
||||
teardown=lambda: None,
|
||||
prompt_path_in_container=prompt_path,
|
||||
)
|
||||
|
||||
def test_wraps_in_sh_c_with_or_fallback(self):
|
||||
argv = dashboard._build_resume_argv_with_fallback(self._bottle())
|
||||
# Must end with `sh -c '<cmd> --continue || <cmd>'`.
|
||||
self.assertEqual(
|
||||
["docker", "exec", "-it", "bot-bottle-dev-abc", "sh", "-c"],
|
||||
argv[:6],
|
||||
)
|
||||
inner = argv[6]
|
||||
self.assertIn("--continue", inner)
|
||||
self.assertIn("||", inner)
|
||||
# Both branches mention claude.
|
||||
self.assertEqual(2, inner.count("claude"))
|
||||
|
||||
def test_inner_args_quoted_safely(self):
|
||||
# Paths with spaces would break naive concatenation.
|
||||
bottle = self._bottle("/home/with space/.prompt")
|
||||
argv = dashboard._build_resume_argv_with_fallback(bottle)
|
||||
inner = argv[-1]
|
||||
# shlex.quote should single-quote any token with a space.
|
||||
self.assertIn("'/home/with space/.prompt'", inner)
|
||||
|
||||
def test_includes_skip_permissions(self):
|
||||
argv = dashboard._build_resume_argv_with_fallback(self._bottle())
|
||||
self.assertIn("--dangerously-skip-permissions", argv[-1])
|
||||
|
||||
def test_includes_prompt_file_flag_when_set(self):
|
||||
bottle = self._bottle("/home/node/.bot-bottle-prompt.txt")
|
||||
argv = dashboard._build_resume_argv_with_fallback(bottle)
|
||||
self.assertIn("--append-system-prompt-file", argv[-1])
|
||||
self.assertIn("/home/node/.bot-bottle-prompt.txt", argv[-1])
|
||||
|
||||
|
||||
class TestClaudeRuntimeArgs(unittest.TestCase):
|
||||
"""The argv passed to `bottle.agent_argv` on each
|
||||
attach. Locked here so the tmux + foreground paths build
|
||||
identical agent invocations."""
|
||||
|
||||
def test_default_skip_permissions_only(self):
|
||||
self.assertEqual(
|
||||
["--dangerously-skip-permissions"],
|
||||
dashboard._agent_runtime_args(resume=False),
|
||||
)
|
||||
|
||||
def test_resume_appends_continue(self):
|
||||
self.assertEqual(
|
||||
["--dangerously-skip-permissions", "--continue"],
|
||||
dashboard._agent_runtime_args(resume=True),
|
||||
)
|
||||
|
||||
def test_remote_control(self):
|
||||
args = dashboard._agent_runtime_args(
|
||||
resume=False, remote_control=True,
|
||||
)
|
||||
self.assertIn("--remote-control", args)
|
||||
|
||||
|
||||
class TestStopBottleFlow(unittest.TestCase):
|
||||
"""Explicit per-bottle stop (PRD 0020 chunk 4). The non-owned
|
||||
path is the one safe to test without curses + docker — the
|
||||
owned path drives `cm.__exit__` against a real launch context
|
||||
and belongs in integration tests."""
|
||||
|
||||
def test_non_owned_returns_cleanup_hint(self):
|
||||
# stdscr is None here on purpose — the non-owned branch
|
||||
# returns before any curses call.
|
||||
msg = dashboard._stop_bottle_flow(
|
||||
stdscr=None, # type: ignore[arg-type]
|
||||
bottles={},
|
||||
slug="ghost-zzz",
|
||||
)
|
||||
self.assertIn("not dashboard-owned", msg)
|
||||
self.assertIn("./cli.py cleanup", msg)
|
||||
|
||||
def test_non_owned_does_not_touch_tmux_state(self):
|
||||
# PRD 0021: a stop on an unknown slug should never clear
|
||||
# the right-pane occupant tracking, even if the slugs
|
||||
# happen to match (defensive — non-owned can't be in the
|
||||
# right pane via the dashboard's normal flow anyway).
|
||||
tmux_state = {"pane_id": "%5", "slug": "live-bbb"}
|
||||
dashboard._stop_bottle_flow(
|
||||
stdscr=None, # type: ignore[arg-type]
|
||||
bottles={},
|
||||
slug="ghost-zzz",
|
||||
tmux_state=tmux_state,
|
||||
)
|
||||
self.assertEqual({"pane_id": "%5", "slug": "live-bbb"}, tmux_state)
|
||||
|
||||
|
||||
class TestOperatorEditFlowGuards(_FakeHomeMixin, unittest.TestCase):
|
||||
"""Chunk-4 contract: the edit flow refuses when the selected
|
||||
agent doesn't have the required sidecar running. The discover-
|
||||
and-prompt scaffolding is gone, so the gating happens here
|
||||
instead of in the key handler."""
|
||||
|
||||
def setUp(self) -> None:
|
||||
self._setup_fake_home()
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self._teardown_fake_home()
|
||||
|
||||
def _agent(self, services: tuple[str, ...]) -> dashboard.ActiveAgent:
|
||||
return dashboard.ActiveAgent(
|
||||
backend_name="docker",
|
||||
slug="dev-abc12",
|
||||
agent_name="impl",
|
||||
started_at="",
|
||||
services=services,
|
||||
)
|
||||
|
||||
def test_routes_edit_refuses_without_egress(self):
|
||||
# Bottle without bottle.egress.routes → no egress sidecar.
|
||||
msg = dashboard._operator_edit_flow(
|
||||
stdscr=None, # type: ignore[arg-type]
|
||||
agent=self._agent(("pipelock", "supervise")),
|
||||
required_service="egress",
|
||||
label="routes",
|
||||
fetch=lambda _: "x",
|
||||
apply=lambda _slug, _content: None,
|
||||
suffix=".yaml",
|
||||
)
|
||||
self.assertIn("no running egress sidecar", msg)
|
||||
self.assertIn("dev-abc12", msg)
|
||||
|
||||
def test_pipelock_edit_refuses_when_pipelock_missing(self):
|
||||
# Belt-and-braces — pipelock should always be there, but
|
||||
# the race window between `compose up` and `docker ps`
|
||||
# update is real.
|
||||
msg = dashboard._operator_edit_flow(
|
||||
stdscr=None, # type: ignore[arg-type]
|
||||
agent=self._agent(()),
|
||||
required_service="pipelock",
|
||||
label="pipelock",
|
||||
fetch=lambda _: "x",
|
||||
apply=lambda _slug, _content: None,
|
||||
suffix=".txt",
|
||||
)
|
||||
self.assertIn("no running pipelock sidecar", msg)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,94 +0,0 @@
|
||||
"""Unit: dashboard_model — state/model layer extracted from dashboard.py.
|
||||
|
||||
Tests for functions that were previously buried in the 2103-line
|
||||
dashboard.py and had no coverage: _approval_status,
|
||||
_proposed_payload_label, and _suffix_for_tool."""
|
||||
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from bot_bottle.cli.dashboard_model import (
|
||||
QueuedProposal,
|
||||
_approval_status,
|
||||
_proposed_payload_label,
|
||||
_suffix_for_tool,
|
||||
)
|
||||
from bot_bottle.supervise import (
|
||||
Proposal,
|
||||
TOOL_CAPABILITY_BLOCK,
|
||||
TOOL_EGRESS_BLOCK,
|
||||
TOOL_PIPELOCK_BLOCK,
|
||||
sha256_hex,
|
||||
)
|
||||
from datetime import datetime, timezone
|
||||
|
||||
|
||||
def _qp(tool: str, slug: str = "dev") -> QueuedProposal:
|
||||
payload = "x"
|
||||
p = Proposal.new(
|
||||
bottle_slug=slug,
|
||||
tool=tool,
|
||||
proposed_file=payload,
|
||||
justification="test",
|
||||
current_file_hash=sha256_hex(payload),
|
||||
now=datetime(2026, 6, 1, 0, 0, 0, tzinfo=timezone.utc),
|
||||
)
|
||||
return QueuedProposal(proposal=p, queue_dir=Path("/tmp/q"))
|
||||
|
||||
|
||||
class TestApprovalStatus(unittest.TestCase):
|
||||
def test_egress_block_base_message(self):
|
||||
qp = _qp(TOOL_EGRESS_BLOCK, slug="my-bot")
|
||||
msg = _approval_status(qp, "approved")
|
||||
self.assertEqual("approved egress-block for [my-bot]", msg)
|
||||
|
||||
def test_modified_verb(self):
|
||||
qp = _qp(TOOL_PIPELOCK_BLOCK, slug="dev")
|
||||
msg = _approval_status(qp, "modified+approved")
|
||||
self.assertEqual("modified+approved pipelock-block for [dev]", msg)
|
||||
|
||||
def test_capability_block_appends_resume_hint(self):
|
||||
qp = _qp(TOOL_CAPABILITY_BLOCK, slug="alpha")
|
||||
msg = _approval_status(qp, "approved")
|
||||
self.assertIn("resume: ./cli.py resume alpha", msg)
|
||||
self.assertIn("approved capability-block for [alpha]", msg)
|
||||
|
||||
def test_egress_block_has_no_resume_hint(self):
|
||||
qp = _qp(TOOL_EGRESS_BLOCK)
|
||||
self.assertNotIn("resume", _approval_status(qp, "approved"))
|
||||
|
||||
def test_pipelock_block_has_no_resume_hint(self):
|
||||
qp = _qp(TOOL_PIPELOCK_BLOCK)
|
||||
self.assertNotIn("resume", _approval_status(qp, "approved"))
|
||||
|
||||
|
||||
class TestProposedPayloadLabel(unittest.TestCase):
|
||||
def test_pipelock_returns_failed_url(self):
|
||||
self.assertEqual("failed URL", _proposed_payload_label(TOOL_PIPELOCK_BLOCK))
|
||||
|
||||
def test_egress_returns_proposed_file(self):
|
||||
self.assertEqual("proposed file", _proposed_payload_label(TOOL_EGRESS_BLOCK))
|
||||
|
||||
def test_capability_returns_proposed_file(self):
|
||||
self.assertEqual("proposed file", _proposed_payload_label(TOOL_CAPABILITY_BLOCK))
|
||||
|
||||
def test_unknown_tool_returns_proposed_file(self):
|
||||
self.assertEqual("proposed file", _proposed_payload_label("unknown-tool"))
|
||||
|
||||
|
||||
class TestSuffixForTool(unittest.TestCase):
|
||||
def test_capability_block_returns_dockerfile_suffix(self):
|
||||
self.assertEqual(".dockerfile", _suffix_for_tool(TOOL_CAPABILITY_BLOCK))
|
||||
|
||||
def test_egress_block_returns_txt(self):
|
||||
self.assertEqual(".txt", _suffix_for_tool(TOOL_EGRESS_BLOCK))
|
||||
|
||||
def test_pipelock_block_returns_txt(self):
|
||||
self.assertEqual(".txt", _suffix_for_tool(TOOL_PIPELOCK_BLOCK))
|
||||
|
||||
def test_unknown_tool_returns_txt(self):
|
||||
self.assertEqual(".txt", _suffix_for_tool("whatever"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,10 +1,10 @@
|
||||
"""Unit: dashboard headless paths (PRD 0013 phase 4, PRD 0014).
|
||||
"""Unit: supervise 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
|
||||
apply_routes_change is stubbed at the supervise 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.
|
||||
"""
|
||||
@@ -19,7 +19,7 @@ from bot_bottle import supervise
|
||||
from bot_bottle.backend.docker.capability_apply import CapabilityApplyError
|
||||
from bot_bottle.backend.docker.egress_apply import EgressApplyError
|
||||
from bot_bottle.backend.docker.pipelock_apply import PipelockApplyError
|
||||
from bot_bottle.cli import dashboard
|
||||
from bot_bottle.cli import supervise as dashboard
|
||||
from bot_bottle.supervise import (
|
||||
Proposal,
|
||||
STATUS_APPROVED,
|
||||
@@ -61,7 +61,7 @@ class _FakeHomeMixin:
|
||||
"""Patch supervise.bot_bottle_root to a temp dir for the test."""
|
||||
|
||||
def _setup_fake_home(self):
|
||||
self._tmp = tempfile.TemporaryDirectory(prefix="dashboard-test.")
|
||||
self._tmp = tempfile.TemporaryDirectory(prefix="supervise-test.")
|
||||
original = supervise.bot_bottle_root
|
||||
|
||||
def fake_root() -> Path:
|
||||
@@ -306,7 +306,7 @@ class TestEgressApplyWiring(_FakeHomeMixin, unittest.TestCase):
|
||||
|
||||
class TestPipelockApplyWiring(_FakeHomeMixin, unittest.TestCase):
|
||||
"""PRD 0015 Phase 2 + PR #25 follow-up: approve() on a
|
||||
pipelock-block proposal carries the failed URL; the dashboard
|
||||
pipelock-block proposal carries the failed URL; the supervise TUI
|
||||
extracts the host, merges it into the running allowlist, and
|
||||
calls apply_allowlist_change with the merged content."""
|
||||
|
||||
@@ -383,7 +383,7 @@ class TestPipelockApplyWiring(_FakeHomeMixin, unittest.TestCase):
|
||||
def test_url_without_host_raises(self):
|
||||
dashboard.fetch_current_allowlist = lambda slug: ""
|
||||
# supervise_server's validator would catch this; if a broken
|
||||
# URL ever makes it through, the dashboard surfaces it too.
|
||||
# URL ever makes it through, the supervise TUI surfaces it too.
|
||||
qp = self._enqueue_pipelock("https:///nohost")
|
||||
with self.assertRaises(PipelockApplyError):
|
||||
dashboard.approve(qp)
|
||||
@@ -458,68 +458,6 @@ class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
|
||||
self.assertEqual(2, len(processed))
|
||||
|
||||
|
||||
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("egress", "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(
|
||||
EgressApplyError("nope")
|
||||
)
|
||||
with self.assertRaises(EgressApplyError):
|
||||
dashboard.operator_edit_routes("dev", '{"routes": []}\n')
|
||||
self.assertEqual([], read_audit_entries("egress", "dev"))
|
||||
|
||||
|
||||
class TestOperatorEditAllowlist(_FakeHomeMixin, unittest.TestCase):
|
||||
"""PRD 0015 Phase 3: operator-initiated pipelock allowlist edit."""
|
||||
|
||||
def setUp(self):
|
||||
self._setup_fake_home()
|
||||
self._original = dashboard.apply_allowlist_change
|
||||
|
||||
def tearDown(self):
|
||||
dashboard.apply_allowlist_change = self._original
|
||||
self._teardown_fake_home()
|
||||
|
||||
def test_writes_audit_with_operator_edit_action(self):
|
||||
dashboard.apply_allowlist_change = lambda slug, content: (
|
||||
"old.example\n", content,
|
||||
)
|
||||
dashboard.operator_edit_allowlist("dev", "old.example\nnew.example\n")
|
||||
entries = read_audit_entries("pipelock", "dev")
|
||||
self.assertEqual(1, len(entries))
|
||||
self.assertEqual(supervise.ACTION_OPERATOR_EDIT, entries[0].operator_action)
|
||||
self.assertIn("+new.example", entries[0].diff)
|
||||
|
||||
def test_failure_does_not_write_audit(self):
|
||||
dashboard.apply_allowlist_change = lambda slug, content: (_ for _ in ()).throw(
|
||||
PipelockApplyError("nope")
|
||||
)
|
||||
with self.assertRaises(PipelockApplyError):
|
||||
dashboard.operator_edit_allowlist("dev", "x.example\n")
|
||||
self.assertEqual([], read_audit_entries("pipelock", "dev"))
|
||||
|
||||
|
||||
class TestEditInEditor(unittest.TestCase):
|
||||
def test_runs_editor_returns_edited_content(self):
|
||||
# Fake "editor" is /bin/sh -c 'cat <<EOF > $1 ... EOF'
|
||||
+12
-12
@@ -1,4 +1,4 @@
|
||||
"""Unit: dashboard launch/crash failure logging (issue #100).
|
||||
"""Unit: supervise launch/crash failure logging (issue #100).
|
||||
|
||||
The dashboard runs under curses, so anything written to stderr while the
|
||||
TUI owns the terminal is wiped when the terminal is restored. These
|
||||
@@ -17,7 +17,7 @@ from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
from bot_bottle import supervise
|
||||
from bot_bottle.cli import dashboard
|
||||
from bot_bottle.cli import supervise as dashboard
|
||||
from bot_bottle.log import Die, die
|
||||
|
||||
|
||||
@@ -44,7 +44,7 @@ class _FakeHomeMixin:
|
||||
~/.bot-bottle."""
|
||||
|
||||
def _setup_fake_home(self):
|
||||
self._tmp = tempfile.TemporaryDirectory(prefix="dash-crash-test.")
|
||||
self._tmp = tempfile.TemporaryDirectory(prefix="supervise-crash-test.")
|
||||
self._orig_root = supervise.bot_bottle_root
|
||||
self._root = Path(self._tmp.name) / ".bot-bottle"
|
||||
supervise.bot_bottle_root = lambda: self._root # type: ignore[assignment]
|
||||
@@ -54,7 +54,7 @@ class _FakeHomeMixin:
|
||||
self._tmp.cleanup()
|
||||
|
||||
|
||||
class TestCmdDashboardErrorPaths(_FakeHomeMixin, unittest.TestCase):
|
||||
class TestCmdSuperviseErrorPaths(_FakeHomeMixin, unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._setup_fake_home()
|
||||
|
||||
@@ -65,7 +65,7 @@ class TestCmdDashboardErrorPaths(_FakeHomeMixin, unittest.TestCase):
|
||||
with mock.patch.object(
|
||||
dashboard.curses, "wrapper", side_effect=KeyboardInterrupt
|
||||
):
|
||||
self.assertEqual(130, dashboard.cmd_dashboard([]))
|
||||
self.assertEqual(130, dashboard.cmd_supervise([]))
|
||||
|
||||
def test_die_resurfaces_message_after_curses(self):
|
||||
buf = io.StringIO()
|
||||
@@ -74,7 +74,7 @@ class TestCmdDashboardErrorPaths(_FakeHomeMixin, unittest.TestCase):
|
||||
side_effect=Die(1, "manifest parse error at line 3"),
|
||||
):
|
||||
with contextlib.redirect_stderr(buf):
|
||||
rc = dashboard.cmd_dashboard([])
|
||||
rc = dashboard.cmd_supervise([])
|
||||
self.assertEqual(1, rc)
|
||||
self.assertIn("manifest parse error at line 3", buf.getvalue())
|
||||
|
||||
@@ -82,7 +82,7 @@ class TestCmdDashboardErrorPaths(_FakeHomeMixin, unittest.TestCase):
|
||||
buf = io.StringIO()
|
||||
with mock.patch.object(dashboard.curses, "wrapper", side_effect=Die(1)):
|
||||
with contextlib.redirect_stderr(buf):
|
||||
rc = dashboard.cmd_dashboard([])
|
||||
rc = dashboard.cmd_supervise([])
|
||||
self.assertEqual(1, rc)
|
||||
self.assertIn("fatal error", buf.getvalue())
|
||||
|
||||
@@ -93,12 +93,12 @@ class TestCmdDashboardErrorPaths(_FakeHomeMixin, unittest.TestCase):
|
||||
side_effect=ValueError("kaboom in render"),
|
||||
):
|
||||
with contextlib.redirect_stderr(buf):
|
||||
rc = dashboard.cmd_dashboard([])
|
||||
rc = dashboard.cmd_supervise([])
|
||||
self.assertEqual(1, rc)
|
||||
out = buf.getvalue()
|
||||
self.assertIn("dashboard crashed: ValueError: kaboom in render", out)
|
||||
self.assertIn("supervise crashed: ValueError: kaboom in render", out)
|
||||
self.assertIn("full traceback written to", out)
|
||||
log_path = self._root / "logs" / "dashboard-crash.log"
|
||||
log_path = self._root / "logs" / "supervise-crash.log"
|
||||
self.assertTrue(log_path.exists())
|
||||
content = log_path.read_text()
|
||||
self.assertIn("kaboom in render", content)
|
||||
@@ -117,9 +117,9 @@ class TestWriteCrashLog(_FakeHomeMixin, unittest.TestCase):
|
||||
raise RuntimeError("explode")
|
||||
except RuntimeError as e:
|
||||
path = dashboard._write_crash_log(e)
|
||||
self.assertEqual(self._root / "logs" / "dashboard-crash.log", path)
|
||||
self.assertEqual(self._root / "logs" / "supervise-crash.log", path)
|
||||
text = path.read_text()
|
||||
self.assertIn("=== dashboard crash", text)
|
||||
self.assertIn("=== supervise crash", text)
|
||||
self.assertIn("RuntimeError: explode", text)
|
||||
|
||||
def test_falls_back_to_tempfile_when_home_unwritable(self):
|
||||
+3
-3
@@ -1,4 +1,4 @@
|
||||
"""Unit: dashboard's detail-view line builder.
|
||||
"""Unit: supervise's detail-view line builder.
|
||||
|
||||
_detail_lines returns (text, attr) tuples. Most are plain; for
|
||||
pipelock-block proposals it appends a "→ would allow host: <host>"
|
||||
@@ -8,7 +8,7 @@ which hostname will land in pipelock's allowlist on approval."""
|
||||
import unittest
|
||||
|
||||
from bot_bottle import supervise
|
||||
from bot_bottle.cli import dashboard
|
||||
from bot_bottle.cli import supervise as dashboard
|
||||
from bot_bottle.supervise import (
|
||||
Proposal,
|
||||
TOOL_CAPABILITY_BLOCK,
|
||||
@@ -63,7 +63,7 @@ class TestPipelockHostHighlight(unittest.TestCase):
|
||||
def test_skips_host_line_when_url_unparseable(self):
|
||||
# Shouldn't happen in production — supervise_server validates
|
||||
# the URL before queuing — but if a malformed payload ever
|
||||
# reaches the dashboard, don't render a misleading host line.
|
||||
# reaches the supervise TUI, don't render a misleading host line.
|
||||
lines = dashboard._detail_lines(
|
||||
_qp(TOOL_PIPELOCK_BLOCK, "garbage-not-a-url"),
|
||||
green_attr=self.GREEN,
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Unit: dashboard's new-proposal highlight window.
|
||||
"""Unit: supervise's new-proposal highlight window.
|
||||
|
||||
The curses rendering itself is exercised manually; this isolates
|
||||
the pure decision `is the proposal still in its post-arrival
|
||||
@@ -6,7 +6,7 @@ highlight window?`"""
|
||||
|
||||
import unittest
|
||||
|
||||
from bot_bottle.cli import dashboard
|
||||
from bot_bottle.cli import supervise as dashboard
|
||||
|
||||
|
||||
class TestIsRecent(unittest.TestCase):
|
||||
Reference in New Issue
Block a user