heartbeat/tests/test_notification_channels_api.py

"""Tests for notification channel CRUD via configio helpers and visibility logic."""
import pytest
from hbd.server import configio, settings as settings_mod


SAMPLE_YAML = """\
hbd_port: 50004
notification_channels:
  pushover_ops:
    type: pushover
    token: abc123
    user: usr456
"""


# ---------------------------------------------------------------------------
# configio helpers
# ---------------------------------------------------------------------------

def test_apply_channel_adds_new_entry(tmp_path):
    f = tmp_path / ".hb.yaml"
    f.write_text(SAMPLE_YAML)
    data = configio.read_roundtrip(str(f))
    configio.apply_channel(data, "email_ops", {"type": "email", "recipients": ["ops@example.com"]})
    assert "email_ops" in data["notification_channels"]
    assert data["notification_channels"]["email_ops"]["type"] == "email"
    # Existing channel preserved
    assert "pushover_ops" in data["notification_channels"]


def test_apply_channel_updates_existing(tmp_path):
    f = tmp_path / ".hb.yaml"
    f.write_text(SAMPLE_YAML)
    data = configio.read_roundtrip(str(f))
    configio.apply_channel(data, "pushover_ops", {"type": "pushover", "token": "new_tok", "user": "new_usr"})
    assert data["notification_channels"]["pushover_ops"]["token"] == "new_tok"


def test_apply_channel_creates_section_if_absent():
    data = {"hbd_port": 50004}
    configio.apply_channel(data, "test_ch", {"type": "pushover", "token": "t", "user": "u"})
    assert "notification_channels" in data
    assert "test_ch" in data["notification_channels"]


def test_delete_channel_removes_entry(tmp_path):
    f = tmp_path / ".hb.yaml"
    f.write_text(SAMPLE_YAML)
    data = configio.read_roundtrip(str(f))
    configio.delete_channel(data, "pushover_ops")
    assert "pushover_ops" not in data["notification_channels"]


def test_delete_channel_noop_for_missing():
    data = {"notification_channels": {"ch1": {"type": "pushover"}}}
    configio.delete_channel(data, "nonexistent")  # must not raise
    assert "ch1" in data["notification_channels"]


def test_delete_channel_noop_when_no_section():
    data = {}
    configio.delete_channel(data, "anything")  # must not raise


def test_apply_channel_persisted_after_write(tmp_path):
    f = tmp_path / ".hb.yaml"
    f.write_text(SAMPLE_YAML)
    data = configio.read_roundtrip(str(f))
    configio.apply_channel(data, "signal_ops", {"type": "signal", "user": "+1", "recipient": "+2"})
    configio.write_config(str(f), data)
    result = configio.read_roundtrip(str(f))
    assert "signal_ops" in result["notification_channels"]
    assert result["notification_channels"]["signal_ops"]["user"] == "+1"
    # Original channel preserved
    assert "pushover_ops" in result["notification_channels"]


def test_delete_channel_persisted_after_write(tmp_path):
    f = tmp_path / ".hb.yaml"
    f.write_text(SAMPLE_YAML)
    data = configio.read_roundtrip(str(f))
    configio.delete_channel(data, "pushover_ops")
    configio.write_config(str(f), data)
    result = configio.read_roundtrip(str(f))
    assert "pushover_ops" not in (result.get("notification_channels") or {})


# ---------------------------------------------------------------------------
# Visibility logic (mirrors http.py _visible_channels_for_user)
# ---------------------------------------------------------------------------

def _visible(config, user):
    """Local copy of the visibility helper for unit testing without the HTTP layer."""
    all_channels = config.get("notification_channels") or {}
    if user.get("admin"):
        return set(all_channels.keys())
    username = user["username"]
    return {
        name for name, cfg in all_channels.items()
        if isinstance(cfg, dict) and (not cfg.get("private") or cfg.get("owner") == username)
    }


CONFIG_VISIBILITY = {
    "notification_channels": {
        "pub_ch":      {"type": "pushover", "token": "t", "user": "u"},
        "alice_priv":  {"type": "email", "owner": "alice", "private": True,
                        "recipients": ["a@a.com"], "sender": "s@a.com", "smtp_server": "s"},
        "bob_priv":    {"type": "signal", "owner": "bob", "private": True,
                        "user": "+1", "recipient": "+2"},
        "admin_owned": {"type": "pushover", "token": "t2", "user": "u2", "owner": "adminuser"},
    }
}


def test_public_channel_visible_to_all():
    for uname in ("alice", "bob", "carol"):
        user = {"username": uname, "admin": False}
        assert "pub_ch" in _visible(CONFIG_VISIBILITY, user)


def test_private_channel_visible_only_to_owner():
    alice = {"username": "alice", "admin": False}
    bob   = {"username": "bob",   "admin": False}
    carol = {"username": "carol", "admin": False}

    assert "alice_priv" in _visible(CONFIG_VISIBILITY, alice)
    assert "alice_priv" not in _visible(CONFIG_VISIBILITY, bob)
    assert "alice_priv" not in _visible(CONFIG_VISIBILITY, carol)

    assert "bob_priv" in _visible(CONFIG_VISIBILITY, bob)
    assert "bob_priv" not in _visible(CONFIG_VISIBILITY, alice)


def test_admin_sees_all_channels():
    admin = {"username": "adminuser", "admin": True}
    visible = _visible(CONFIG_VISIBILITY, admin)
    assert visible == {"pub_ch", "alice_priv", "bob_priv", "admin_owned"}


def test_admin_owned_channel_is_public_by_default():
    alice = {"username": "alice", "admin": False}
    assert "admin_owned" in _visible(CONFIG_VISIBILITY, alice)


# ---------------------------------------------------------------------------
# Channel type schemas
# ---------------------------------------------------------------------------

def test_all_required_types_in_schema():
    for t in ("pushover", "email", "signal", "matrix", "sms_voipms"):
        assert t in settings_mod.CHANNEL_TYPE_SCHEMAS


def test_schema_fields_have_required_keys():
    for type_id, schema in settings_mod.CHANNEL_TYPE_SCHEMAS.items():
        assert "label" in schema, f"{type_id} missing label"
        assert "fields" in schema, f"{type_id} missing fields"
        for f in schema["fields"]:
            for k in ("key", "label", "type", "required"):
                assert k in f, f"{type_id} field missing {k!r}"


def test_secret_fields_use_secret_type():
    """Known secret fields must be typed 'secret' so the UI masks them."""
    secret_keys = {"token", "user_key", "api_key", "api_password",
                   "smtp_password", "access_token"}
    for type_id, schema in settings_mod.CHANNEL_TYPE_SCHEMAS.items():
        for f in schema["fields"]:
            if f["key"] in secret_keys:
                assert f["type"] == "secret", (
                    f"{type_id}.{f['key']} should be type 'secret'"
                )


def test_channel_labels_not_empty():
    for type_id, schema in settings_mod.CHANNEL_TYPE_SCHEMAS.items():
        assert schema["label"].strip(), f"{type_id} has empty label"