heartbeat/tests/test_config_access.py

"""Tests for ownership rules and scoped config merges (hbd.server.config_access)."""
import pytest
from hbd.server import config_access as ca


# ---------------------------------------------------------------------------
# Visibility helpers
# ---------------------------------------------------------------------------

def test_is_global_when_no_owner():
    assert ca.is_global({"type": "pushover"})
    assert ca.is_global({"type": "email", "owner": ""})
    assert ca.is_global(None)          # non-dict is treated as global
    assert not ca.is_global({"type": "email", "owner": "alice"})


def test_user_can_use_global_or_own():
    assert ca.user_can_use({"type": "pushover"}, "alice")
    assert ca.user_can_use({"owner": "alice"}, "alice")
    assert not ca.user_can_use({"owner": "bob"}, "alice")


HOSTS = {
    "web1": {"owner": "alice", "watch": True},
    "web2": {"owner": "bob", "managers": ["alice"]},
    "web3": {"owner": "bob", "managers": "carol"},      # string manager form
    "web4": {"watch": True},                            # unowned
}


def test_user_hosts_owner_and_manager():
    assert set(ca.user_hosts(HOSTS, "alice")) == {"web1", "web2"}
    assert set(ca.user_hosts(HOSTS, "carol")) == {"web3"}
    assert ca.user_hosts(HOSTS, "dave") == {}
    assert ca.user_hosts(None, "alice") == {}


CHANNELS = {
    "global_ch": {"type": "pushover"},
    "alice_ch": {"type": "email", "owner": "alice"},
    "bob_ch": {"type": "signal", "owner": "bob"},
}


def test_user_channels_global_plus_own():
    assert set(ca.user_channels(CHANNELS, "alice")) == {"global_ch", "alice_ch"}
    assert set(ca.user_channels(CHANNELS, "dave")) == {"global_ch"}


TCS = {
    "default": {"thresholds": {}},
    "alice_tc": {"owner": "alice", "thresholds": {}},
}


def test_user_threshold_configs_global_plus_own():
    assert set(ca.user_threshold_configs(TCS, "alice")) == {"default", "alice_tc"}
    assert set(ca.user_threshold_configs(TCS, "bob")) == {"default"}


# ---------------------------------------------------------------------------
# merge_hosts_scoped
# ---------------------------------------------------------------------------

MERGE_HOSTS = {
    "mine": {"owner": "alice", "watch": True, "notification_channels": ["global_ch"]},
    "managed": {"owner": "bob", "managers": ["alice"], "watch": True,
                "notification_channels": ["bob_ch"]},
    "foreign": {"owner": "bob", "watch": False},
}
MERGE_CHANNELS = {
    "global_ch": {"type": "pushover"},
    "alice_ch": {"type": "email", "owner": "alice"},
    "bob_ch": {"type": "signal", "owner": "bob"},
}
MERGE_TCS = {"alice_tc": {"owner": "alice"}, "bob_tc": {"owner": "bob"}}


def _merge_hosts(payload, existing=None):
    return ca.merge_hosts_scoped(
        existing if existing is not None else dict(MERGE_HOSTS),
        payload, "alice", MERGE_CHANNELS, MERGE_TCS,
    )


def _full_payload(**overrides):
    """Payload covering alice's full visible subset, with per-host overrides."""
    p = {
        "mine": {"owner": "alice", "watch": True, "notification_channels": ["global_ch"]},
        "managed": {"watch": True, "notification_channels": ["bob_ch"]},
    }
    p.update(overrides)
    return p


def test_merge_hosts_preserves_foreign_hosts():
    result = _merge_hosts(_full_payload())
    assert result["foreign"] == {"owner": "bob", "watch": False}


def test_merge_hosts_manager_edits_settings():
    result = _merge_hosts(_full_payload(managed={"watch": False, "dyndns": True,
                                                 "notification_channels": ["bob_ch"]}))
    assert result["managed"]["watch"] is False
    assert result["managed"]["dyndns"] is True
    # access fields carried over untouched
    assert result["managed"]["owner"] == "bob"
    assert result["managed"]["managers"] == ["alice"]


def test_merge_hosts_manager_cannot_change_owner():
    with pytest.raises(ca.ScopedMergeError, match="managed"):
        _merge_hosts(_full_payload(managed={"watch": True, "owner": "alice"}))


def test_merge_hosts_manager_cannot_change_managers():
    with pytest.raises(ca.ScopedMergeError, match="managers"):
        _merge_hosts(_full_payload(managed={"watch": True, "managers": ["alice", "carol"]}))


def test_merge_hosts_manager_same_access_values_ok():
    # Submitting unchanged access fields is not a violation.
    result = _merge_hosts(_full_payload(managed={"watch": True, "owner": "bob",
                                                 "managers": ["alice"],
                                                 "notification_channels": ["bob_ch"]}))
    assert result["managed"]["owner"] == "bob"


def test_merge_hosts_manager_cannot_delete():
    payload = {"mine": {"owner": "alice", "watch": True}}  # 'managed' missing
    with pytest.raises(ca.ScopedMergeError, match="managed"):
        _merge_hosts(payload)


def test_merge_hosts_owner_can_delete():
    payload = _full_payload()
    del payload["mine"]
    result = _merge_hosts(payload)
    assert "mine" not in result
    assert "managed" in result and "foreign" in result


def test_merge_hosts_owner_can_transfer_ownership():
    result = _merge_hosts(_full_payload(mine={"owner": "bob", "watch": True}))
    assert result["mine"]["owner"] == "bob"


def test_merge_hosts_new_host_owner_forced():
    result = _merge_hosts(_full_payload(newhost={"watch": True, "owner": "bob"}))
    assert result["newhost"]["owner"] == "alice"


def test_merge_hosts_cannot_touch_foreign_host():
    with pytest.raises(ca.ScopedMergeError, match="foreign"):
        _merge_hosts(_full_payload(foreign={"watch": True}))


def test_merge_hosts_cannot_add_foreign_private_channel():
    with pytest.raises(ca.ScopedMergeError, match="bob_ch"):
        _merge_hosts(_full_payload(mine={"owner": "alice", "watch": True,
                                         "notification_channels": ["bob_ch"]}))


def test_merge_hosts_keeps_preexisting_foreign_assignment():
    # 'managed' already has bob_ch; keeping it is fine.
    result = _merge_hosts(_full_payload())
    assert result["managed"]["notification_channels"] == ["bob_ch"]


def test_merge_hosts_can_add_global_and_own_channel():
    result = _merge_hosts(_full_payload(mine={"owner": "alice", "watch": True,
                                              "notification_channels": ["global_ch", "alice_ch"]}))
    assert result["mine"]["notification_channels"] == ["global_ch", "alice_ch"]


def test_merge_hosts_threshold_assignment_validated():
    with pytest.raises(ca.ScopedMergeError, match="bob_tc"):
        _merge_hosts(_full_payload(mine={"owner": "alice", "watch": True,
                                         "threshold_config": ["bob_tc"]}))
    result = _merge_hosts(_full_payload(mine={"owner": "alice", "watch": True,
                                              "threshold_config": ["default", "alice_tc"]}))
    assert result["mine"]["threshold_config"] == ["default", "alice_tc"]


def test_merge_hosts_rejects_unknown_fields():
    with pytest.raises(ca.ScopedMergeError, match="sneaky"):
        _merge_hosts(_full_payload(mine={"owner": "alice", "watch": True, "sneaky": 1}))


def test_merge_hosts_owner_clearing_list_removes_key():
    result = _merge_hosts(_full_payload(mine={"owner": "alice", "watch": True}))
    assert "notification_channels" not in result["mine"]