generated from coulomb/repo-seed
Bernd Worsch
91d031ae20
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>
204 lines
6.9 KiB
Python
204 lines
6.9 KiB
Python
"""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
|