generated from coulomb/repo-seed
feat: implement OpsCatalog extension (BRIDGE-WP-0002)
Adds the OpsCatalog subsystem: a Git-backed YAML catalog of operations domains, targets, bridges, and actor classes. Includes catalog loader, cross-reference validator, bridge resolver (inline-first, catalog fallback), and new CLI commands: `bridge targets`, `bridge targets show`, `bridge catalog list/validate/show`. Updates `up/down/restart` to resolve bridge names from the catalog when not defined inline. 142 tests, all green. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
203
tests/test_catalog_cli.py
Normal file
203
tests/test_catalog_cli.py
Normal file
@@ -0,0 +1,203 @@
|
||||
"""Tests for catalog CLI commands (targets, catalog list/validate/show)."""
|
||||
import json
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from bridge.cli import app
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
# Config with catalog_path pointing to a fixture
|
||||
BASE_CONFIG = textwrap.dedent("""\
|
||||
tunnels: {{}}
|
||||
actors: {{}}
|
||||
catalog_path: {catalog_path}
|
||||
""")
|
||||
|
||||
CONFIG_NO_CATALOG = textwrap.dedent("""\
|
||||
tunnels: {}
|
||||
actors: {}
|
||||
""")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def catalog_dir(tmp_path):
|
||||
root = tmp_path / "opscatalog"
|
||||
domain_dir = root / "domains" / "coulombcore"
|
||||
(domain_dir / "targets").mkdir(parents=True)
|
||||
(domain_dir / "bridges").mkdir(parents=True)
|
||||
actors_dir = root / "actors"
|
||||
actors_dir.mkdir(parents=True)
|
||||
|
||||
(domain_dir / "domain.yaml").write_text(textwrap.dedent("""\
|
||||
type: domain
|
||||
id: coulombcore
|
||||
name: CoulombCore Infrastructure
|
||||
description: Core infra
|
||||
environment: production
|
||||
"""))
|
||||
|
||||
(domain_dir / "targets" / "state-hub.yaml").write_text(textwrap.dedent("""\
|
||||
type: target
|
||||
id: state-hub
|
||||
domain: coulombcore
|
||||
kind: service
|
||||
description: State coordination service
|
||||
reachable_via:
|
||||
- state-hub-coulombcore
|
||||
"""))
|
||||
|
||||
(domain_dir / "bridges" / "state-hub-coulombcore.yaml").write_text(textwrap.dedent("""\
|
||||
type: bridge
|
||||
id: state-hub-coulombcore
|
||||
domain: coulombcore
|
||||
target: state-hub
|
||||
description: Ops bridge for state hub
|
||||
access_method: ssh-reverse
|
||||
host: coulombcore.local
|
||||
remote_port: 18000
|
||||
local_port: 8000
|
||||
ssh_user: ubuntu
|
||||
ssh_key: ~/.ssh/id_ops
|
||||
actor: agent.claude-coulombcore
|
||||
"""))
|
||||
|
||||
(actors_dir / "agents.yaml").write_text(textwrap.dedent("""\
|
||||
type: actor
|
||||
id: agent.claude-coulombcore
|
||||
class: automation
|
||||
description: Claude Code agent
|
||||
"""))
|
||||
return root
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config_file(tmp_path, catalog_dir):
|
||||
f = tmp_path / "tunnels.yaml"
|
||||
f.write_text(BASE_CONFIG.format(catalog_path=str(catalog_dir)))
|
||||
return f
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def env(config_file, tmp_path):
|
||||
return {
|
||||
"BRIDGE_CONFIG": str(config_file),
|
||||
"BRIDGE_STATE_DIR": str(tmp_path / "state"),
|
||||
}
|
||||
|
||||
|
||||
class TestTargetsCommand:
|
||||
def test_targets_shows_table(self, env):
|
||||
result = runner.invoke(app, ["targets"], env=env)
|
||||
assert result.exit_code == 0
|
||||
assert "state-hub" in result.output
|
||||
|
||||
def test_targets_json(self, env):
|
||||
result = runner.invoke(app, ["targets", "--json"], env=env)
|
||||
assert result.exit_code == 0
|
||||
data = json.loads(result.output)
|
||||
assert isinstance(data, list)
|
||||
assert any(t["target"] == "state-hub" for t in data)
|
||||
assert any(t["domain"] == "coulombcore" for t in data)
|
||||
|
||||
def test_targets_domain_filter(self, env):
|
||||
result = runner.invoke(app, ["targets", "--domain", "coulombcore"], env=env)
|
||||
assert result.exit_code == 0
|
||||
assert "state-hub" in result.output
|
||||
|
||||
def test_targets_domain_filter_unknown(self, env):
|
||||
result = runner.invoke(app, ["targets", "--domain", "nonexistent"], env=env)
|
||||
assert result.exit_code == 0
|
||||
# No results but no crash
|
||||
|
||||
def test_targets_no_catalog_configured(self, tmp_path):
|
||||
f = tmp_path / "tunnels.yaml"
|
||||
f.write_text(CONFIG_NO_CATALOG)
|
||||
result = runner.invoke(app, ["targets"], env={"BRIDGE_CONFIG": str(f)})
|
||||
assert result.exit_code == 1
|
||||
assert "catalog" in result.output.lower()
|
||||
|
||||
def test_targets_show_subcommand(self, env):
|
||||
result = runner.invoke(app, ["targets", "show", "state-hub"], env=env)
|
||||
assert result.exit_code == 0
|
||||
assert "state-hub" in result.output
|
||||
assert "coulombcore" in result.output
|
||||
|
||||
def test_targets_show_unknown(self, env):
|
||||
result = runner.invoke(app, ["targets", "show", "nonexistent"], env=env)
|
||||
assert result.exit_code == 1
|
||||
|
||||
|
||||
class TestCatalogCommand:
|
||||
def test_catalog_list(self, env):
|
||||
result = runner.invoke(app, ["catalog", "list"], env=env)
|
||||
assert result.exit_code == 0
|
||||
assert "coulombcore" in result.output
|
||||
|
||||
def test_catalog_list_json(self, env):
|
||||
result = runner.invoke(app, ["catalog", "list", "--json"], env=env)
|
||||
assert result.exit_code == 0
|
||||
data = json.loads(result.output)
|
||||
assert isinstance(data, list)
|
||||
assert any(d["domain"] == "coulombcore" for d in data)
|
||||
|
||||
def test_catalog_validate_clean(self, env):
|
||||
result = runner.invoke(app, ["catalog", "validate"], env=env)
|
||||
assert result.exit_code == 0
|
||||
assert "valid" in result.output.lower() or "ok" in result.output.lower() or "0" in result.output
|
||||
|
||||
def test_catalog_validate_with_errors(self, tmp_path):
|
||||
# Catalog with dangling reference
|
||||
root = tmp_path / "bad-catalog"
|
||||
domain_dir = root / "domains" / "d"
|
||||
(domain_dir / "targets").mkdir(parents=True)
|
||||
(domain_dir / "domain.yaml").write_text(
|
||||
"type: domain\nid: d\nname: D\n"
|
||||
)
|
||||
(domain_dir / "targets" / "t.yaml").write_text(
|
||||
"type: target\nid: t\ndomain: d\nkind: service\nreachable_via:\n - missing-bridge\n"
|
||||
)
|
||||
f = tmp_path / "tunnels.yaml"
|
||||
f.write_text(BASE_CONFIG.format(catalog_path=str(root)))
|
||||
result = runner.invoke(app, ["catalog", "validate"], env={"BRIDGE_CONFIG": str(f)})
|
||||
assert result.exit_code == 1
|
||||
assert "missing-bridge" in result.output
|
||||
|
||||
def test_catalog_show(self, env):
|
||||
result = runner.invoke(app, ["catalog", "show", "state-hub-coulombcore"], env=env)
|
||||
assert result.exit_code == 0
|
||||
assert "state-hub-coulombcore" in result.output
|
||||
assert "coulombcore.local" in result.output
|
||||
|
||||
def test_catalog_show_unknown(self, env):
|
||||
result = runner.invoke(app, ["catalog", "show", "nonexistent"], env=env)
|
||||
assert result.exit_code == 1
|
||||
|
||||
def test_catalog_no_catalog_configured(self, tmp_path):
|
||||
f = tmp_path / "tunnels.yaml"
|
||||
f.write_text(CONFIG_NO_CATALOG)
|
||||
result = runner.invoke(app, ["catalog", "list"], env={"BRIDGE_CONFIG": str(f)})
|
||||
assert result.exit_code == 1
|
||||
|
||||
|
||||
class TestUpWithCatalogFallback:
|
||||
def test_up_resolves_catalog_bridge(self, env):
|
||||
"""bridge up <catalog-bridge-name> works when name not in inline tunnels.yaml."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
with patch("bridge.cli.TunnelManager") as mock_mgr_cls:
|
||||
mock_mgr = MagicMock()
|
||||
mock_mgr.is_running.return_value = False
|
||||
mock_mgr_cls.return_value = mock_mgr
|
||||
|
||||
result = runner.invoke(app, ["up", "state-hub-coulombcore"], env=env)
|
||||
|
||||
assert result.exit_code == 0
|
||||
mock_mgr.start.assert_called_once()
|
||||
|
||||
def test_up_unknown_bridge_exit_1(self, env):
|
||||
result = runner.invoke(app, ["up", "totally-nonexistent"], env=env)
|
||||
assert result.exit_code == 1
|
||||
Reference in New Issue
Block a user