diff --git a/bcc.py b/bcc.py index b0d598f..6f33f93 100644 --- a/bcc.py +++ b/bcc.py @@ -72,6 +72,12 @@ WARN = "#fbbf24" STATUS_COLORS = {"ok": GOOD, "missing": BAD, "warn": WARN, "remote": "#60a5fa", "unknown": WARN} STATUS_GLYPH = {"ok": "●", "missing": "●", "warn": "▲", "remote": "◆", "unknown": "○"} +# Health dot (spawn-test outcome, see core.HealthStatus) shown per row in the +# server tables' "Health" column -- distinct from the PATH-dependency Status +# column above. +HEALTH_COLORS = {"ok": GOOD, "failed": BAD, "untested": MUTED} +HEALTH_GLYPH = {"ok": "●", "failed": "●", "untested": "○"} + STYLESHEET = f""" /* No font-family here on purpose: Qt already uses the native system UI font on every platform (San Francisco / Segoe UI / desktop default). Naming @@ -1416,6 +1422,12 @@ class MainWindow(QMainWindow): self._focused_table = None self._row_of_index: dict[int, tuple] = {} self._undo_stack: list[list] = [] # each entry: snapshot of self.servers + self._filter_query = "" + self._health: dict[str, tuple[str, str]] = {} # server name -> (HealthStatus, summary) + self._test_all_queue: list[core.ServerEntry] = [] + self._test_all_total = 0 + self._test_all_done = 0 + self._health_tester: SpawnTester | None = None central = QWidget() self.setCentralWidget(central) @@ -1533,10 +1545,10 @@ class MainWindow(QMainWindow): # --- left (server table) -------------------------------------------- # def _make_server_table(self, object_name=None): - t = QTableWidget(0, 4) + t = QTableWidget(0, 5) if object_name: t.setObjectName(object_name) - t.setHorizontalHeaderLabels(["On", "Name", "Type", "Status"]) + t.setHorizontalHeaderLabels(["On", "Name", "Type", "Status", "Health"]) t.verticalHeader().setVisible(False) t.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows) t.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection) @@ -1546,6 +1558,7 @@ class MainWindow(QMainWindow): h.setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch) h.setSectionResizeMode(2, QHeaderView.ResizeMode.ResizeToContents) h.setSectionResizeMode(3, QHeaderView.ResizeMode.ResizeToContents) + h.setSectionResizeMode(4, QHeaderView.ResizeMode.ResizeToContents) t.itemSelectionChanged.connect(lambda tbl=t: self._on_selection(tbl)) t.itemChanged.connect(self._table_item_changed) return t @@ -1560,6 +1573,12 @@ class MainWindow(QMainWindow): head.setObjectName("h1") v.addWidget(head) + self.search_box = QLineEdit() + self.search_box.setPlaceholderText("Search servers by name, command, or url…") + self.search_box.setClearButtonEnabled(True) + self.search_box.textChanged.connect(self._on_search_changed) + v.addWidget(self.search_box) + # Active and Disabled sections live in a vertical splitter so the user # can drag the divider instead of being stuck with a fixed-height # disabled list. @@ -1616,12 +1635,17 @@ class MainWindow(QMainWindow): self.undo_btn = QPushButton("Undo") self.undo_btn.setEnabled(False) self.undo_btn.setToolTip("Undo last change (Ctrl+Z)") + self.test_all_btn = QPushButton("Test all") + self.test_all_btn.setToolTip( + "Spawn-test every enabled local server, one at a time, and fill in the Health column" + ) self.add_btn.clicked.connect(self.add_server) self.dup_btn.clicked.connect(self.duplicate_server) self.del_btn.clicked.connect(self.delete_server) self.paste_btn.clicked.connect(self.paste_json) self.copy_btn.clicked.connect(self.copy_to_menu) self.undo_btn.clicked.connect(self._undo) + self.test_all_btn.clicked.connect(self._test_all_servers) for b in ( self.add_btn, self.dup_btn, @@ -1629,6 +1653,7 @@ class MainWindow(QMainWindow): self.paste_btn, self.copy_btn, self.undo_btn, + self.test_all_btn, ): bar.addWidget(b) # Ctrl+Z shortcut @@ -1749,6 +1774,7 @@ class MainWindow(QMainWindow): self.restart_btn.hide() self._undo_stack.clear() self.undo_btn.setEnabled(False) + self._health.clear() # health results are per-profile; a fresh load invalidates them self._refresh_tables(select_index=0 if self.servers else -1) self._update_status(saved=False) if repaired: @@ -1785,6 +1811,14 @@ class MainWindow(QMainWindow): QColor(STATUS_COLORS.get(dep["status"], MUTED)) ) # status stays colored even when off table.setItem(r, 3, st) + health_status, health_summary = self._health.get(s.name, (core.HealthStatus.UNTESTED, "")) + if s.kind == "remote" and s.name not in self._health: + health_summary = "remote server — use “Test connection” in the editor" + health_item = QTableWidgetItem(HEALTH_GLYPH.get(health_status, "○")) + health_item.setForeground(QColor(HEALTH_COLORS.get(health_status, MUTED))) + health_item.setTextAlignment(Qt.AlignmentFlag.AlignCenter) + health_item.setToolTip(health_summary or health_status) + table.setItem(r, 4, health_item) self._row_of_index[master_idx] = (table, r) def _refresh_tables(self, select_index=None): @@ -1793,8 +1827,16 @@ class MainWindow(QMainWindow): self._row_of_index = {} self.active_table.setRowCount(0) self.disabled_table.setRowCount(0) + query = self._filter_query n_active = n_disabled = 0 + total_active = total_disabled = 0 for i, s in enumerate(self.servers): + if s.enabled: + total_active += 1 + else: + total_disabled += 1 + if not core.server_matches_filter(s, query): + continue if s.enabled: self._add_row(self.active_table, i, s) n_active += 1 @@ -1803,8 +1845,18 @@ class MainWindow(QMainWindow): n_disabled += 1 self.active_table.setVisible(n_active > 0) self.active_empty.setVisible(n_active == 0) + self.active_empty.setText( + "No matches." + if query.strip() and total_active and not n_active + else "No active servers. Add one, or Paste JSON." + ) self.disabled_table.setVisible(n_disabled > 0) self.disabled_empty.setVisible(n_disabled == 0) + self.disabled_empty.setText( + "No matches." + if query.strip() and total_disabled and not n_disabled + else "Nothing disabled." + ) self._refresh_badges() self._suppress_table = False self._suppress_sel = False @@ -1816,6 +1868,58 @@ class MainWindow(QMainWindow): self._load_editor_from_selection() self._validate() + # --- search / filter -------------------------------------------------- # + def _on_search_changed(self, text): + self._filter_query = text + cur = self._current_index() + self._refresh_tables(select_index=cur if cur >= 0 else None) + + # --- test all (spawn-test every enabled local server) ---------------- # + def _test_all_servers(self): + targets = [s for s in self.servers if s.enabled and s.kind == "stdio"] + if not targets: + self.status.setText("No enabled local servers to test.") + return + self._test_all_queue = list(targets) + self._test_all_total = len(targets) + self._test_all_done = 0 + self.test_all_btn.setEnabled(False) + self.test_all_btn.setText(f"Testing 0/{self._test_all_total}…") + self._run_next_health_test() + + def _run_next_health_test(self): + if not self._test_all_queue: + self.test_all_btn.setEnabled(True) + self.test_all_btn.setText("Test all") + self.status.setText(f"Tested {self._test_all_done} server(s).") + return + entry = self._test_all_queue.pop(0) + self._health_tester = SpawnTester(dict(entry.data), timeout=3.0) + self._health_tester.done.connect( + lambda result, name=entry.name: self._on_health_test_done(name, result) + ) + self._health_tester.start() + + def _on_health_test_done(self, name, result): + status, summary = core.health_from_spawn_result(result) + self._health[name] = (status, summary) + self._test_all_done += 1 + self.test_all_btn.setText(f"Testing {self._test_all_done}/{self._test_all_total}…") + self._update_health_cell(name, status, summary) + self._run_next_health_test() + + def _update_health_cell(self, name, status, summary): + idx = next((i for i, s in enumerate(self.servers) if s.name == name), None) + if idx is None or idx not in self._row_of_index: + return + table, row = self._row_of_index[idx] + item = table.item(row, 4) + if item is None: + return + item.setText(HEALTH_GLYPH.get(status, "○")) + item.setForeground(QColor(HEALTH_COLORS.get(status, MUTED))) + item.setToolTip(summary or status) + def _section_html(self, title, n, miss, warn): base = f"{title} · {n}" if miss: @@ -1922,6 +2026,9 @@ class MainWindow(QMainWindow): entry.data = self.editor.dump_data() # The server stays in its section (enable state unchanged), so update # its existing row in place rather than re-rendering. + # An edit invalidates any cached "Test all" result -- the server that + # was spawn-tested no longer matches what's on disk once saved. + self._health.pop(entry.name, None) loc = self._row_of_index.get(idx) if loc: table, row = loc @@ -1932,6 +2039,13 @@ class MainWindow(QMainWindow): st = table.item(row, 3) st.setText(f"{STATUS_GLYPH.get(dep['status'], '○')} {dep['label']}") st.setForeground(QColor(STATUS_COLORS.get(dep["status"], MUTED))) + health_item = table.item(row, 4) + if health_item is not None: + health_item.setText(HEALTH_GLYPH.get(core.HealthStatus.UNTESTED, "○")) + health_item.setForeground( + QColor(HEALTH_COLORS.get(core.HealthStatus.UNTESTED, MUTED)) + ) + health_item.setToolTip("Not tested since last edit.") self._suppress_table = False self._refresh_badges() self._mark_dirty() diff --git a/bcc_core.py b/bcc_core.py index bc8d603..7a8574f 100644 --- a/bcc_core.py +++ b/bcc_core.py @@ -1213,6 +1213,32 @@ def validate_servers(servers: list[ServerEntry]) -> list[str]: return problems +# --------------------------------------------------------------------------- # +# Search / filter +# --------------------------------------------------------------------------- # +def server_matches_filter(entry: ServerEntry, query: str) -> bool: + """ + Case-insensitive substring match against a server's name, and its + command (stdio) or url (remote). An empty/whitespace-only query matches + everything -- that's what lets the search box double as "no filter". + """ + q = (query or "").strip().lower() + if not q: + return True + if q in entry.name.lower(): + return True + if entry.kind == "remote": + haystack = str(entry.data.get("url", "")) + else: + haystack = str(entry.data.get("command", "")) + return q in haystack.lower() + + +def filter_servers(entries: list[ServerEntry], query: str) -> list[ServerEntry]: + """Return only the entries that match `query` (see server_matches_filter).""" + return [e for e in entries if server_matches_filter(e, query)] + + # --------------------------------------------------------------------------- # # Dependency / PATH checking # --------------------------------------------------------------------------- # @@ -1682,6 +1708,53 @@ def spawn_test(data: dict, timeout: float = 3.0) -> dict: } +# --------------------------------------------------------------------------- # +# Health status (maps a spawn_test() result to a simple tri-state for the +# server-list UI's per-row status dot; see "Test all" in bcc.py) +# --------------------------------------------------------------------------- # +class HealthStatus: + """ + Tri-state health for the server-list status dot. A plain class of string + constants -- not an Enum -- to match the plain-string status values used + elsewhere in this module (see check_dependency's 'status'). + """ + + UNTESTED = "untested" + OK = "ok" + FAILED = "failed" + + +def health_from_spawn_result(result: dict) -> tuple[str, str]: + """ + Map a spawn_test() result dict to (HealthStatus, short_summary) for the + server-list status column. Reuses spawn_test's own outcome classification + rather than re-deriving pass/fail from returncode/stderr: + + outcome "ok" -> OK (server started and kept running) + outcome "not_applicable" -> UNTESTED (remote server, or no command set) + anything else -> FAILED (exited, crashed, or not found) + + The summary is short enough for a table cell/tooltip; when the process + wrote to stderr before dying, its first line is appended for context. + """ + outcome = result.get("outcome", "") + detail = result.get("detail", "") or "" + stderr = (result.get("stderr") or "").strip() + + if outcome == "ok": + return HealthStatus.OK, detail or "started" + if outcome == "not_applicable": + return HealthStatus.UNTESTED, detail or "not applicable" + + # exited / crashed / not_found: the server didn't come up cleanly. + summary = detail or outcome + if stderr: + first_line = stderr.splitlines()[0].strip() + if first_line: + summary = f"{summary} — {first_line}" + return HealthStatus.FAILED, summary + + def test_remote(url: str, timeout: float = 5.0) -> tuple[bool, str]: """ Reachability check for a url-based MCP server. ANY HTTP response (even 4xx/5xx) diff --git a/tests/test_core.py b/tests/test_core.py index 90dfd6f..578387b 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -1200,3 +1200,129 @@ def test_fetch_latest_release_malformed_json_returns_none(monkeypatch): urllib.request, "urlopen", lambda req, timeout=None: _FakeHTTPResponse(b"not json") ) assert c.fetch_latest_release() is None + + +# --------------------------------------------------------------------------- # +# Server search / filter (#27) +# --------------------------------------------------------------------------- # +def test_filter_matches_name(): + e = c.ServerEntry("brave-search", {"command": "npx", "args": ["-y", "@x/brave"]}, True) + assert c.server_matches_filter(e, "brave") + + +def test_filter_matches_stdio_command(): + e = c.ServerEntry("filesystem", {"command": "uvx", "args": []}, True) + assert c.server_matches_filter(e, "uvx") + + +def test_filter_matches_remote_url(): + e = c.ServerEntry("hosted", {"url": "https://mcp.example.com/sse"}, True) + assert c.server_matches_filter(e, "example.com") + + +def test_filter_is_case_insensitive(): + e = c.ServerEntry("BraveSearch", {"command": "NPX", "args": []}, True) + assert c.server_matches_filter(e, "bravesearch") + assert c.server_matches_filter(e, "npx") + + +def test_filter_empty_query_matches_everything(): + e = c.ServerEntry("anything", {"command": "node", "args": []}, True) + assert c.server_matches_filter(e, "") + assert c.server_matches_filter(e, " ") + + +def test_filter_no_match_returns_false(): + e = c.ServerEntry("filesystem", {"command": "uvx", "args": []}, True) + assert not c.server_matches_filter(e, "nonexistent") + + +def test_filter_remote_query_does_not_match_stdio_command_field(): + e = c.ServerEntry("hosted", {"url": "https://mcp.example.com/sse"}, True) + assert not c.server_matches_filter(e, "npx") + + +def test_filter_servers_returns_only_matches(): + servers = [ + c.ServerEntry("brave-search", {"command": "npx", "args": []}, True), + c.ServerEntry("filesystem", {"command": "uvx", "args": []}, True), + c.ServerEntry("hosted", {"url": "https://mcp.example.com/sse"}, True), + ] + assert [s.name for s in c.filter_servers(servers, "brave")] == ["brave-search"] + assert [s.name for s in c.filter_servers(servers, "")] == [s.name for s in servers] + assert c.filter_servers(servers, "zzz-nope") == [] + + +# --------------------------------------------------------------------------- # +# Health status mapping (#28) +# --------------------------------------------------------------------------- # +def test_health_from_spawn_result_ok(): + result = { + "outcome": "ok", + "returncode": None, + "stderr": "", + "detail": "still running after 3s — server started successfully", + } + status, summary = c.health_from_spawn_result(result) + assert status == c.HealthStatus.OK + assert "started" in summary + + +def test_health_from_spawn_result_not_applicable_is_untested(): + result = { + "outcome": "not_applicable", + "returncode": None, + "stderr": "", + "detail": "remote server", + } + status, summary = c.health_from_spawn_result(result) + assert status == c.HealthStatus.UNTESTED + assert summary == "remote server" + + +def test_health_from_spawn_result_crashed_is_failed(): + result = { + "outcome": "crashed", + "returncode": 1, + "stderr": "Traceback: boom\nsecond line", + "detail": "process exited with code 1", + } + status, summary = c.health_from_spawn_result(result) + assert status == c.HealthStatus.FAILED + assert "process exited with code 1" in summary + assert "Traceback: boom" in summary + + +def test_health_from_spawn_result_exited_is_failed(): + result = { + "outcome": "exited", + "returncode": 0, + "stderr": "", + "detail": "process exited cleanly (code 0) — unusual", + } + status, summary = c.health_from_spawn_result(result) + assert status == c.HealthStatus.FAILED + assert "exited cleanly" in summary + + +def test_health_from_spawn_result_not_found_is_failed(): + result = { + "outcome": "not_found", + "returncode": None, + "stderr": "", + "detail": "command not found: totallybogus", + } + status, summary = c.health_from_spawn_result(result) + assert status == c.HealthStatus.FAILED + assert "not found" in summary + + +def test_health_from_spawn_result_failed_without_stderr_has_no_dash(): + result = { + "outcome": "not_found", + "returncode": None, + "stderr": "", + "detail": "command not found: x", + } + _, summary = c.health_from_spawn_result(result) + assert "—" not in summary