generated from coulomb/repo-seed
tegwick
365c0d611a
Implements the full BRIDGE-WP-0003 workplan: 188 tests passing, 0 lint errors. ## What's added **Capability registry** (`src/bridge/capabilities.py`): - 10 capabilities with required_access_modes (cli/mcp/skill) - Single source of truth for what OpsBridge does and where **MCP server** (`src/bridge/mcp_server/server.py`): - 10 FastMCP tools: bridge_up/down/restart/status/logs + 5 catalog_* tools - 3 resources: bridge://status, catalog://domains, catalog://targets - `.mcp.json` for project-scope auto-registration - `scripts/register_mcp.py` for user-scope machine-global registration **Skill** (`~/.claude/plugins/ops-bridge/bridge-status.md`): - /bridge-status: health table with emoji indicators + remediation advice **Cross-mode test coverage enforcement**: - `tests/conftest.py`: capability/access_mode marks + collect_capability_coverage() - `tests/test_mcp.py`: 31 FastMCP in-process client tests (Client(mcp) pattern) - `tests/test_skill.py`: static skill lint against capability registry - `tests/test_coverage_completeness.py`: meta-test that fails if any required (capability × mode) pair lacks a test; also validates CLI commands and MCP tools are registered in the capability registry **ADR** (`architecture/adr-001-cross-mode-capability-registry.md`): - Documents the registry pattern and FastMCP 3.x testing approach Key implementation note: FastMCP 3.x in-process results are in result.content[0].text (JSON string), not result.data directly. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
"""Tests for catalog validator."""
|
|
from bridge.catalog.models import (
|
|
ActorClass,
|
|
Catalog,
|
|
CatalogBridge,
|
|
CatalogDomain,
|
|
CatalogTarget,
|
|
)
|
|
from bridge.catalog.validator import validate_catalog
|
|
|
|
|
|
def _make_full_catalog() -> Catalog:
|
|
cat = Catalog()
|
|
cat.domains["coulombcore"] = CatalogDomain(id="coulombcore", name="CoulombCore")
|
|
cat.targets["state-hub"] = CatalogTarget(
|
|
id="state-hub",
|
|
domain="coulombcore",
|
|
kind="service",
|
|
reachable_via=["state-hub-coulombcore"],
|
|
)
|
|
cat.bridges["state-hub-coulombcore"] = CatalogBridge(
|
|
id="state-hub-coulombcore",
|
|
domain="coulombcore",
|
|
target="state-hub",
|
|
host="host.local",
|
|
remote_port=18000,
|
|
local_port=8000,
|
|
ssh_user="ubuntu",
|
|
ssh_key="~/.ssh/id_ops",
|
|
actor="agent.claude-coulombcore",
|
|
)
|
|
cat.actors["agent.claude-coulombcore"] = ActorClass(
|
|
id="agent.claude-coulombcore",
|
|
actor_class="automation",
|
|
)
|
|
return cat
|
|
|
|
|
|
class TestValidateCatalog:
|
|
def test_valid_catalog_no_errors(self):
|
|
cat = _make_full_catalog()
|
|
errors = validate_catalog(cat)
|
|
assert errors == []
|
|
|
|
def test_target_domain_must_exist(self):
|
|
cat = _make_full_catalog()
|
|
cat.targets["orphan"] = CatalogTarget(
|
|
id="orphan", domain="nonexistent-domain", kind="service"
|
|
)
|
|
errors = validate_catalog(cat)
|
|
assert any("orphan" in e and "nonexistent-domain" in e for e in errors)
|
|
|
|
def test_target_reachable_via_must_exist(self):
|
|
cat = _make_full_catalog()
|
|
cat.targets["state-hub"].reachable_via.append("nonexistent-bridge")
|
|
errors = validate_catalog(cat)
|
|
assert any("nonexistent-bridge" in e for e in errors)
|
|
|
|
def test_bridge_domain_must_exist(self):
|
|
cat = _make_full_catalog()
|
|
cat.bridges["state-hub-coulombcore"].domain = "missing-domain"
|
|
errors = validate_catalog(cat)
|
|
assert any("missing-domain" in e for e in errors)
|
|
|
|
def test_bridge_target_must_exist(self):
|
|
cat = _make_full_catalog()
|
|
cat.bridges["state-hub-coulombcore"].target = "missing-target"
|
|
errors = validate_catalog(cat)
|
|
assert any("missing-target" in e for e in errors)
|
|
|
|
def test_bridge_actor_must_exist(self):
|
|
cat = _make_full_catalog()
|
|
cat.bridges["state-hub-coulombcore"].actor = "nonexistent-actor"
|
|
errors = validate_catalog(cat)
|
|
assert any("nonexistent-actor" in e for e in errors)
|
|
|
|
def test_multiple_errors_all_reported(self):
|
|
cat = Catalog()
|
|
# Target with dangling domain and reachable_via
|
|
cat.targets["t1"] = CatalogTarget(
|
|
id="t1", domain="missing", kind="service", reachable_via=["missing-bridge"]
|
|
)
|
|
# Bridge with dangling domain + target + actor
|
|
cat.bridges["b1"] = CatalogBridge(
|
|
id="b1", domain="missing", target="missing", host="h",
|
|
remote_port=1, local_port=2, ssh_user="u", ssh_key="k", actor="missing-actor",
|
|
)
|
|
errors = validate_catalog(cat)
|
|
assert len(errors) >= 4
|
|
|
|
def test_empty_catalog_is_valid(self):
|
|
cat = Catalog()
|
|
assert validate_catalog(cat) == []
|