ops-bridge/tests/test_config.py

"""Tests for config loading."""
import textwrap
import warnings

import pytest

from bridge.config import ConfigError, load_config
from bridge.models import ActorType


VALID_YAML = textwrap.dedent("""\
    tunnels:
      state-hub-coulombcore:
        host: coulombcore.local
        remote_port: 18000
        local_port: 8000
        ssh_user: ubuntu
        ssh_key: ~/.ssh/id_ops
        actor: agt-claude-coulombcore
        health_check:
          url: http://127.0.0.1:18000/health
          interval_seconds: 30
          timeout_seconds: 5
        reconnect:
          max_attempts: 0
          backoff_initial: 5
          backoff_max: 60

    actors:
      agt-claude-coulombcore:
        class: agt
        description: Claude Code agent on CoulombCore
      adm-bernd:
        class: adm
        description: Bernd Worsch
""")


@pytest.fixture
def config_file(tmp_path):
    f = tmp_path / "tunnels.yaml"
    f.write_text(VALID_YAML)
    return f


def test_load_valid_config(config_file, monkeypatch):
    monkeypatch.setenv("BRIDGE_CONFIG", str(config_file))
    cfg = load_config()
    assert "state-hub-coulombcore" in cfg.tunnels
    t = cfg.tunnels["state-hub-coulombcore"]
    assert t.host == "coulombcore.local"
    assert t.remote_port == 18000
    assert t.local_port == 8000
    assert t.ssh_user == "ubuntu"
    assert t.actor == "agt-claude-coulombcore"


def test_health_check_loaded(config_file, monkeypatch):
    monkeypatch.setenv("BRIDGE_CONFIG", str(config_file))
    cfg = load_config()
    t = cfg.tunnels["state-hub-coulombcore"]
    assert t.health_check is not None
    assert t.health_check.url == "http://127.0.0.1:18000/health"
    assert t.health_check.interval_seconds == 30


def test_reconnect_policy_loaded(config_file, monkeypatch):
    monkeypatch.setenv("BRIDGE_CONFIG", str(config_file))
    cfg = load_config()
    t = cfg.tunnels["state-hub-coulombcore"]
    assert t.reconnect.max_attempts == 0
    assert t.reconnect.backoff_initial == 5
    assert t.reconnect.backoff_max == 60


def test_actors_loaded(config_file, monkeypatch):
    monkeypatch.setenv("BRIDGE_CONFIG", str(config_file))
    cfg = load_config()
    assert "agt-claude-coulombcore" in cfg.actors
    a = cfg.actors["agt-claude-coulombcore"]
    assert a.actor_type == ActorType.AGT
    assert "adm-bernd" in cfg.actors


def test_missing_required_field_raises(tmp_path, monkeypatch):
    f = tmp_path / "bad.yaml"
    f.write_text(textwrap.dedent("""\
        tunnels:
          broken:
            remote_port: 18000
            local_port: 8000
        actors: {}
    """))
    monkeypatch.setenv("BRIDGE_CONFIG", str(f))
    with pytest.raises(ConfigError, match="host"):
        load_config()


def test_invalid_yaml_raises(tmp_path, monkeypatch):
    f = tmp_path / "bad.yaml"
    f.write_text("tunnels: [\nnot: valid: yaml")
    monkeypatch.setenv("BRIDGE_CONFIG", str(f))
    with pytest.raises(ConfigError):
        load_config()


def test_missing_config_file_raises(tmp_path, monkeypatch):
    monkeypatch.setenv("BRIDGE_CONFIG", str(tmp_path / "nonexistent.yaml"))
    with pytest.raises(ConfigError, match="not found"):
        load_config()


def test_tunnel_without_health_check(tmp_path, monkeypatch):
    f = tmp_path / "tunnels.yaml"
    f.write_text(textwrap.dedent("""\
        tunnels:
          simple:
            host: host.local
            remote_port: 9000
            local_port: 8000
            ssh_user: ubuntu
            ssh_key: ~/.ssh/id_rsa
            actor: adm-bernd
        actors:
          adm-bernd:
            class: adm
            description: Bernd
    """))
    monkeypatch.setenv("BRIDGE_CONFIG", str(f))
    cfg = load_config()
    assert cfg.tunnels["simple"].health_check is None


class TestActorTypeValidation:
    def test_canonical_agt_accepted(self, tmp_path, monkeypatch):
        f = tmp_path / "t.yaml"
        f.write_text(textwrap.dedent("""\
            tunnels:
              t:
                host: h
                remote_port: 1
                local_port: 2
                ssh_user: u
                ssh_key: ~/.ssh/k
                actor: agt-claude
            actors:
              agt-claude:
                class: agt
        """))
        monkeypatch.setenv("BRIDGE_CONFIG", str(f))
        cfg = load_config()
        assert cfg.actors["agt-claude"].actor_type == ActorType.AGT

    def test_canonical_atm_accepted(self, tmp_path, monkeypatch):
        f = tmp_path / "t.yaml"
        f.write_text(textwrap.dedent("""\
            tunnels:
              t:
                host: h
                remote_port: 1
                local_port: 2
                ssh_user: u
                ssh_key: ~/.ssh/k
                actor: atm-backup
            actors:
              atm-backup:
                class: atm
        """))
        monkeypatch.setenv("BRIDGE_CONFIG", str(f))
        cfg = load_config()
        assert cfg.actors["atm-backup"].actor_type == ActorType.ATM

    def test_wrong_prefix_raises_config_error(self, tmp_path, monkeypatch):
        f = tmp_path / "t.yaml"
        f.write_text(textwrap.dedent("""\
            tunnels:
              t:
                host: h
                remote_port: 1
                local_port: 2
                ssh_user: u
                ssh_key: ~/.ssh/k
                actor: adm-bernd
            actors:
              adm-bernd:
                class: agt
        """))
        monkeypatch.setenv("BRIDGE_CONFIG", str(f))
        with pytest.raises(ConfigError, match="must start with 'agt-'"):
            load_config()

    def test_missing_prefix_raises_config_error(self, tmp_path, monkeypatch):
        f = tmp_path / "t.yaml"
        f.write_text(textwrap.dedent("""\
            tunnels:
              t:
                host: h
                remote_port: 1
                local_port: 2
                ssh_user: u
                ssh_key: ~/.ssh/k
                actor: operator.bernd
            actors:
              operator.bernd:
                class: adm
        """))
        monkeypatch.setenv("BRIDGE_CONFIG", str(f))
        with pytest.raises(ConfigError, match="must start with 'adm-'"):
            load_config()

    def test_unknown_class_raises_config_error(self, tmp_path, monkeypatch):
        f = tmp_path / "t.yaml"
        f.write_text(textwrap.dedent("""\
            tunnels:
              t:
                host: h
                remote_port: 1
                local_port: 2
                ssh_user: u
                ssh_key: ~/.ssh/k
                actor: adm-bernd
            actors:
              adm-bernd:
                class: wizard
        """))
        monkeypatch.setenv("BRIDGE_CONFIG", str(f))
        with pytest.raises(ConfigError, match="unknown class"):
            load_config()

    def test_legacy_human_maps_to_adm_with_warning(self, tmp_path, monkeypatch):
        f = tmp_path / "t.yaml"
        f.write_text(textwrap.dedent("""\
            tunnels:
              t:
                host: h
                remote_port: 1
                local_port: 2
                ssh_user: u
                ssh_key: ~/.ssh/k
                actor: adm-bernd
            actors:
              adm-bernd:
                class: human
        """))
        monkeypatch.setenv("BRIDGE_CONFIG", str(f))
        with warnings.catch_warnings(record=True) as w:
            warnings.simplefilter("always")
            cfg = load_config()
        assert cfg.actors["adm-bernd"].actor_type == ActorType.ADM
        assert any("deprecated" in str(x.message).lower() for x in w)

    def test_legacy_automation_maps_to_atm_with_warning(self, tmp_path, monkeypatch):
        f = tmp_path / "t.yaml"
        f.write_text(textwrap.dedent("""\
            tunnels:
              t:
                host: h
                remote_port: 1
                local_port: 2
                ssh_user: u
                ssh_key: ~/.ssh/k
                actor: atm-cron
            actors:
              atm-cron:
                class: automation
        """))
        monkeypatch.setenv("BRIDGE_CONFIG", str(f))
        with warnings.catch_warnings(record=True) as w:
            warnings.simplefilter("always")
            cfg = load_config()
        assert cfg.actors["atm-cron"].actor_type == ActorType.ATM
        assert any("deprecated" in str(x.message).lower() for x in w)


class TestCertCommandConfig:
    def test_cert_command_parsed(self, tmp_path, monkeypatch):
        f = tmp_path / "t.yaml"
        f.write_text(textwrap.dedent("""\
            tunnels:
              t:
                host: h
                remote_port: 1
                local_port: 2
                ssh_user: u
                ssh_key: ~/.ssh/k
                actor: agt-bridge
                cert_command: "warden sign agt-bridge --pubkey /tmp/k.pub"
            actors:
              agt-bridge:
                class: agt
        """))
        monkeypatch.setenv("BRIDGE_CONFIG", str(f))
        cfg = load_config()
        assert cfg.tunnels["t"].cert_command == "warden sign agt-bridge --pubkey /tmp/k.pub"

    def test_no_cert_command_is_none(self, config_file, monkeypatch):
        monkeypatch.setenv("BRIDGE_CONFIG", str(config_file))
        cfg = load_config()
        assert cfg.tunnels["state-hub-coulombcore"].cert_command is None


def test_duplicate_local_ports_are_rejected():
    """A local-forward port collision is silent at runtime; fail at load."""
    from bridge.config import ConfigError, _parse_tunnels

    base = {
        "host": "192.0.2.1",
        "remote_port": 18000,
        "ssh_user": "tegwick",
        "ssh_key": "~/.ssh/id_ops",
        "actor": "agt-claude-railiance01",
        "direction": "local",
    }
    with pytest.raises(ConfigError, match="Duplicate local_port"):
        _parse_tunnels(
            {
                "k3s-a": {**base, "local_port": 16443},
                "k3s-b": {**base, "local_port": 16443},
            }
        )


def test_reverse_tunnels_may_share_a_local_port():
    """local_port is the destination on this workstation for a reverse tunnel.

    Every state-hub-* reverse tunnel targets 8000 on purpose, so each remote box
    reaches this hub at its own 18000. Rejecting that would break the tool.
    """
    from bridge.config import _parse_tunnels

    base = {
        "host": "192.0.2.1",
        "remote_port": 18000,
        "ssh_user": "tegwick",
        "ssh_key": "~/.ssh/id_ops",
        "actor": "agt-claude-railiance01",
    }
    tunnels = _parse_tunnels(
        {
            "state-hub-a": {**base, "local_port": 8000},
            "state-hub-b": {**base, "local_port": 8000},
        }
    )
    assert {t.local_port for t in tunnels.values()} == {8000}