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>
75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
"""Tests for domain models."""
|
|
from bridge.models import (
|
|
ActorInfo,
|
|
BridgeState,
|
|
HealthCheckConfig,
|
|
ReconnectPolicy,
|
|
TunnelConfig,
|
|
)
|
|
|
|
|
|
class TestBridgeState:
|
|
def test_all_states_defined(self):
|
|
states = {s.value for s in BridgeState}
|
|
assert states == {"stopped", "starting", "connected", "degraded", "reconnecting", "failed"}
|
|
|
|
def test_state_is_string(self):
|
|
assert BridgeState.STOPPED == "stopped"
|
|
|
|
|
|
class TestReconnectPolicy:
|
|
def test_defaults(self):
|
|
p = ReconnectPolicy()
|
|
assert p.max_attempts == 0
|
|
assert p.backoff_initial == 5
|
|
assert p.backoff_max == 60
|
|
|
|
def test_custom(self):
|
|
p = ReconnectPolicy(max_attempts=3, backoff_initial=2, backoff_max=30)
|
|
assert p.max_attempts == 3
|
|
|
|
|
|
class TestHealthCheckConfig:
|
|
def test_required_url(self):
|
|
h = HealthCheckConfig(url="http://127.0.0.1:18000/health")
|
|
assert h.url == "http://127.0.0.1:18000/health"
|
|
assert h.interval_seconds == 30
|
|
assert h.timeout_seconds == 5
|
|
|
|
|
|
class TestTunnelConfig:
|
|
def test_minimal(self):
|
|
t = TunnelConfig(
|
|
name="test-tunnel",
|
|
host="host.local",
|
|
remote_port=18000,
|
|
local_port=8000,
|
|
ssh_user="ubuntu",
|
|
ssh_key="~/.ssh/id_ops",
|
|
actor="operator.bernd",
|
|
)
|
|
assert t.name == "test-tunnel"
|
|
assert t.health_check is None
|
|
assert isinstance(t.reconnect, ReconnectPolicy)
|
|
|
|
def test_with_health_check(self):
|
|
hc = HealthCheckConfig(url="http://127.0.0.1:18000/health")
|
|
t = TunnelConfig(
|
|
name="test",
|
|
host="h",
|
|
remote_port=1,
|
|
local_port=2,
|
|
ssh_user="u",
|
|
ssh_key="k",
|
|
actor="a",
|
|
health_check=hc,
|
|
)
|
|
assert t.health_check is hc
|
|
|
|
|
|
class TestActorInfo:
|
|
def test_fields(self):
|
|
a = ActorInfo(name="operator.bernd", actor_class="human", description="Bernd")
|
|
assert a.name == "operator.bernd"
|
|
assert a.actor_class == "human"
|