heartbeat/tests/test_settings_sections.py

import pytest
from hbd.server import settings as settings_mod

CFG = {
    "hbd_port": 50004,
    "interval": 20,
    "grace": 2,
    "users": {
        "alice": {"full_name": "Alice Smith", "admin": True, "password": "pbkdf2:sha256:abc",
                  "notification_channels": ["pushover_ops"]},
    },
    "oauth": {
        "gitea": {"type": "gitea", "url": "https://git.example.com",
                  "client_id": "cid", "client_secret": "csec", "label": "Sign in with Gitea"},
    },
    "notification_channels": {
        "pushover_ops": {"type": "pushover", "token": "tok", "user": "usr"},
    },
    "hosts": {},
}


def test_sections_have_section_mode():
    sections = settings_mod.get_settings_sections(CFG)
    for s in sections:
        assert "section_mode" in s, f"Section {s['id']} missing section_mode"
        assert s["section_mode"] in ("form", "yaml", "channels", "hosts", "thresholds")


def test_sections_have_api_section():
    sections = settings_mod.get_settings_sections(CFG)
    for s in sections:
        assert "api_section" in s, f"Section {s['id']} missing api_section"


def test_network_section_has_editable_fields():
    sections = settings_mod.get_settings_sections(CFG)
    network = next(s for s in sections if s["id"] == "network")
    assert network["section_mode"] == "form"
    assert network["api_section"] == "server"
    editable = [f for f in network["fields"] if f["editable"]]
    assert len(editable) >= 2  # hbd_port, ws_port at minimum


def test_thresholds_and_dns_section_modes():
    sections = settings_mod.get_settings_sections(CFG)
    by_id = {s["id"]: s for s in sections}
    assert by_id["thresholds"]["section_mode"] == "thresholds"
    assert by_id["thresholds"]["api_section"] == "thresholds"
    assert by_id["dns"]["section_mode"] == "form"
    assert by_id["dns"]["api_section"] == "dns"


def test_hosts_section_uses_hosts_mode():
    sections = settings_mod.get_settings_sections(CFG)
    hosts_sec = next(s for s in sections if s["id"] == "hosts")
    assert hosts_sec["section_mode"] == "hosts"
    assert hosts_sec["api_section"] == "hosts"


def test_channels_section_uses_channels_mode():
    sections = settings_mod.get_settings_sections(CFG)
    ch_sec = next(s for s in sections if s["id"] == "channels")
    assert ch_sec["section_mode"] == "channels"
    assert ch_sec["api_section"] == "notification_channels"
    assert len(ch_sec["channels"]) == 1
    ch = ch_sec["channels"][0]
    assert ch["name"] == "pushover_ops"
    assert ch["type"] == "pushover"
    assert "owner" in ch
    assert "editable" in ch


def test_channel_type_schemas_exported():
    assert hasattr(settings_mod, "CHANNEL_TYPE_SCHEMAS")
    for required_type in ("pushover", "email", "signal", "matrix", "sms_voipms"):
        assert required_type in settings_mod.CHANNEL_TYPE_SCHEMAS
        schema = settings_mod.CHANNEL_TYPE_SCHEMAS[required_type]
        assert "label" in schema
        assert "fields" in schema
        for f in schema["fields"]:
            assert "key" in f
            assert "type" in f
            assert "required" in f


def test_oauth_section_exists():
    sections = settings_mod.get_settings_sections(CFG)
    oauth = next((s for s in sections if s["id"] == "oauth"), None)
    assert oauth is not None
    assert oauth["section_mode"] == "form"
    assert oauth["api_section"] == "oauth"
    assert len(oauth["providers"]) == 1
    assert oauth["providers"][0]["name"] == "gitea"
    assert oauth["providers"][0]["client_secret"] == "•••"


def test_all_channel_names_returned():
    result = settings_mod.get_settings_data(CFG)
    assert "all_channel_names" in result
    assert "pushover_ops" in result["all_channel_names"]


def test_users_section_has_user_list():
    sections = settings_mod.get_settings_sections(CFG)
    users_sec = next(s for s in sections if s["id"] == "users")
    assert users_sec["section_mode"] == "form"
    assert users_sec["api_section"] == "users"
    assert len(users_sec["users"]) == 1
    assert users_sec["users"][0]["username"] == "alice"
    # Password hash never exposed
    assert "password" not in users_sec["users"][0]


# ---------------------------------------------------------------------------
# Per-user filtering (owners/managers on the settings page)
# ---------------------------------------------------------------------------

from types import SimpleNamespace  # noqa: E402


MULTI_CFG = {
    **CFG,
    "users": {
        "alice": {"full_name": "Alice", "admin": True, "password": "x"},
        "bob": {"full_name": "Bob", "admin": False, "password": "x"},
    },
    "notification_channels": {
        "global_ch": {"type": "pushover", "token": "t", "user": "u"},
        "bob_ch": {"type": "pushover", "token": "t", "user": "u", "owner": "bob"},
        "carol_ch": {"type": "pushover", "token": "t", "user": "u", "owner": "carol"},
    },
    "threshold_configs": {
        "bob_tc": {"owner": "bob", "thresholds": {}},
        "carol_tc": {"owner": "carol", "thresholds": {}},
    },
    "hosts": {
        "bobhost": {"owner": "bob"},
        "managedhost": {"owner": "carol", "managers": ["bob"]},
        "carolhost": {"owner": "carol"},
    },
}

BOB = SimpleNamespace(username="bob", admin=False)
ADMIN = SimpleNamespace(username="alice", admin=True)


def test_nonadmin_sees_only_three_sections():
    sections = settings_mod.get_settings_sections(MULTI_CFG, user=BOB)
    assert [s["id"] for s in sections] == ["channels", "hosts", "thresholds"]


def test_nonadmin_sections_hide_admin_fields():
    sections = settings_mod.get_settings_sections(MULTI_CFG, user=BOB)
    for s in sections:
        assert s["fields"] == []


def test_nonadmin_hosts_filtered_with_is_owner():
    sections = settings_mod.get_settings_sections(MULTI_CFG, user=BOB)
    hosts = next(s for s in sections if s["id"] == "hosts")["hosts"]
    by_name = {h["name"]: h for h in hosts}
    assert set(by_name) == {"bobhost", "managedhost"}
    assert by_name["bobhost"]["is_owner"] is True
    assert by_name["managedhost"]["is_owner"] is False


def test_nonadmin_channels_filtered_with_editable():
    sections = settings_mod.get_settings_sections(MULTI_CFG, user=BOB)
    chans = {c["name"]: c for c in next(s for s in sections if s["id"] == "channels")["channels"]}
    assert set(chans) == {"global_ch", "bob_ch"}
    assert chans["bob_ch"]["editable"] is True
    assert chans["global_ch"]["editable"] is False


def test_admin_sees_everything_with_editable():
    sections = settings_mod.get_settings_sections(MULTI_CFG, user=ADMIN)
    ids = [s["id"] for s in sections]
    assert "network" in ids and "users" in ids
    chans = {c["name"]: c for c in next(s for s in sections if s["id"] == "channels")["channels"]}
    assert set(chans) == {"global_ch", "bob_ch", "carol_ch"}
    assert all(c["editable"] for c in chans.values())


def test_settings_data_pickers_filtered_for_nonadmin():
    data = settings_mod.get_settings_data(MULTI_CFG, user=BOB)
    assert data["all_channel_names"] == ["bob_ch", "global_ch"]
    assert data["all_threshold_configs"] == ["bob_tc"]


def test_no_user_means_admin_view():
    sections = settings_mod.get_settings_sections(MULTI_CFG)  # auth disabled
    assert len(sections) > 3


def test_filtering_unaffected_by_other_users_in_config():
    """The users-section loop must not clobber the requesting username (regression)."""
    cfg = dict(MULTI_CFG)
    cfg["users"] = {
        "alice": {"full_name": "Alice", "admin": True, "password": "x"},
        "bob": {"full_name": "Bob", "admin": False, "password": "x"},
        "zed": {"full_name": "Zed", "admin": False, "password": "x"},  # bob is not last
    }
    sections = settings_mod.get_settings_sections(cfg, user=BOB)
    hosts = {h["name"] for h in next(s for s in sections if s["id"] == "hosts")["hosts"]}
    assert hosts == {"bobhost", "managedhost"}