haproxy-openmanager/backend/tests/test_keepalived_parser.py

"""Issue #27 follow-up (v1.10.4) — unit tests for parsing an EXISTING keepalived.conf so a
hand-maintained VIP can be adopted.

Pure-function tests; no DB, no network. The parser exists because the heartbeat carries only
the VIP address and a best-effort MASTER/BACKUP, while rendering a node's config needs eleven
fields — and because adoption REPLACES the operator's file, so anything our renderer cannot
reproduce has to be reported as a blocker rather than silently dropped.
"""
from __future__ import annotations

import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

os.environ.setdefault("SECRET_KEY", "test-secret-key-for-keepalived-parser-tests")

from services import keepalived_config as kc  # noqa: E402
from services.keepalived_parser import (  # noqa: E402
    KeepalivedParseError, analyse_keepalived_conf, build_adoption_candidate,
    parse_keepalived_conf,
)


# A realistic hand-maintained config: two nodes, unicast VRRP, password auth, HAProxy check.
HANDWRITTEN = """\
! Configuration File for keepalived
global_defs {
    enable_script_security
    script_user root
}

vrrp_script chk_haproxy {
    script "/etc/keepalived/check_haproxy.sh"
    interval 2
    weight -21
}

vrrp_instance VI_1 {
    state MASTER
    interface eth0        # public leg
    virtual_router_id 51
    priority 150
    advert_int 1
    authentication {
        auth_type PASS
        auth_pass s3cr3t
    }
    unicast_src_ip 10.0.0.11
    unicast_peer {
        10.0.0.12
    }
    virtual_ipaddress {
        10.0.0.100/24 dev eth0
    }
    track_script {
        chk_haproxy
    }
}
"""


def _only_candidate(text):
    parsed = parse_keepalived_conf(text)
    assert len(parsed["instances"]) == 1
    return build_adoption_candidate(parsed, parsed["instances"][0])


def test_parses_a_handwritten_config_into_model_fields():
    cand = _only_candidate(HANDWRITTEN)
    assert cand["adoptable"] is True, cand["blockers"]
    assert cand["blockers"] == []
    assert cand["vip"] == {
        "virtual_ip": "10.0.0.100",
        "prefix_length": 24,
        "virtual_router_id": 51,
        "advert_int": 1,
        "use_unicast": True,
        "track_haproxy": True,
        "auth_pass": "s3cr3t",
    }
    assert cand["member"] == {"network_interface": "eth0", "role": "MASTER", "priority": 150}
    assert cand["peers"] == ["10.0.0.12"] and cand["unicast_src_ip"] == "10.0.0.11"
    assert cand["defaulted"] == []          # every value came from the file, nothing assumed


def test_comment_and_layout_variants():
    # `!` and `#` both start comments; a block may open and close on one line; a quoted
    # script path keeps its spaces. None of this may change the parse.
    text = """\
#!/not/a/shebang — this whole line is a comment
vrrp_script chk { script "/opt/my scripts/chk.sh" }
vrrp_instance VI_1 { state BACKUP
    interface eth1   ! trailing bang comment
    virtual_router_id 7
    priority 90
    virtual_ipaddress { 192.168.5.9/32 dev eth1 }
}
"""
    parsed = parse_keepalived_conf(text)
    assert parsed["scripts"]["chk"]["script"] == "/opt/my scripts/chk.sh"
    inst = parsed["instances"][0]
    assert inst["state"] == "BACKUP" and inst["interface"] == "eth1"
    assert inst["virtual_router_id"] == 7 and inst["priority"] == 90
    assert inst["virtual_ips"] == [
        {"address": "192.168.5.9", "prefix_length": 32, "dev": "eth1", "extra": []}
    ]


def test_documented_defaults_are_applied_and_flagged():
    # keepalived's own defaults for absent directives. Applying them re-renders the same
    # behaviour, so they are allowed — but the UI must be able to say they were assumed.
    text = """\
vrrp_instance VI_1 {
    interface eth0
    virtual_router_id 12
    virtual_ipaddress { 10.1.1.5/24 dev eth0 }
}
"""
    cand = _only_candidate(text)
    assert cand["adoptable"] is True, cand["blockers"]
    assert cand["member"]["role"] == "BACKUP" and cand["member"]["priority"] == 100
    assert cand["vip"]["advert_int"] == 1
    assert sorted(cand["defaulted"]) == ["advert_int", "priority", "state"]
    # No authentication block and no track_script — both legal, both faithfully represented.
    assert cand["vip"]["auth_pass"] is None and cand["vip"]["track_haproxy"] is False
    assert cand["vip"]["use_unicast"] is False


def _blockers_for(text):
    return " | ".join(_only_candidate(text)["blockers"])


def test_directives_we_cannot_render_block_adoption():
    # THE central safety property: adoption overwrites the file, so a failover hook we do not
    # render would be destroyed. It must stop the flow, not warn.
    text = HANDWRITTEN.replace(
        "    track_script {", '    notify_master "/usr/local/bin/promote.sh"\n    track_script {')
    blockers = _blockers_for(text)
    assert "notify_master" in blockers and "would delete it" in blockers
    assert _only_candidate(text)["adoptable"] is False


def test_multiple_addresses_in_one_instance_block_adoption():
    text = HANDWRITTEN.replace("        10.0.0.100/24 dev eth0",
                               "        10.0.0.100/24 dev eth0\n        10.0.0.101/24 dev eth0")
    blockers = _blockers_for(text)
    assert "2 addresses" in blockers and "10.0.0.101" in blockers


def test_missing_vrid_blocks_adoption_with_the_split_brain_reason():
    text = HANDWRITTEN.replace("    virtual_router_id 51\n", "")
    blockers = _blockers_for(text)
    assert "no virtual_router_id" in blockers and "separate VRRP domains" in blockers


def test_missing_prefix_blocks_adoption():
    # Our renderer always writes an explicit prefix; guessing one would change the netmask of a
    # live VIP, so the operator has to state it.
    text = HANDWRITTEN.replace("10.0.0.100/24 dev eth0", "10.0.0.100 dev eth0")
    blockers = _blockers_for(text)
    assert "no explicit prefix length" in blockers


def test_address_on_a_different_dev_blocks_adoption():
    text = HANDWRITTEN.replace("10.0.0.100/24 dev eth0", "10.0.0.100/24 dev eth1")
    blockers = _blockers_for(text)
    assert "dev eth1" in blockers and "interface eth0" in blockers


def test_foreign_track_script_blocks_adoption():
    text = HANDWRITTEN.replace("        chk_haproxy", "        chk_custom")
    blockers = _blockers_for(text)
    assert "chk_custom" in blockers and "replaced by OpenManager" in blockers


def test_unsupported_auth_type_blocks_adoption():
    text = HANDWRITTEN.replace("auth_type PASS", "auth_type AH")
    assert "auth_type AH" in _blockers_for(text)


def test_fractional_advert_int_blocks_adoption():
    # Rounding 0.5s to 1s changes VRRP timing, so adopt-and-alter is not acceptable.
    text = HANDWRITTEN.replace("advert_int 1", "advert_int 0.5")
    blockers = _blockers_for(text)
    assert "advert_int 0.5" in blockers and "fractional" in blockers


def test_half_configured_unicast_blocks_adoption():
    text = HANDWRITTEN.replace("    unicast_peer {\n        10.0.0.12\n    }\n", "")
    assert "unicast_src_ip without unicast_peer" in _blockers_for(text)


def test_sync_group_and_lvs_sections_block_adoption():
    text = HANDWRITTEN + """
vrrp_sync_group VG1 {
    group {
        VI_1
    }
}
virtual_server 10.0.0.100 80 {
    lb_algo rr
}
"""
    parsed = parse_keepalived_conf(text)
    assert [g["name"] for g in parsed["sync_groups"]] == ["VG1"]
    directives = " ".join(d["directive"] for d in parsed["unsupported"])
    assert "vrrp_sync_group VG1" in directives and "virtual_server" in directives
    # Both are top-level, so EVERY candidate in the file is blocked — a sync group changes
    # failover semantics for the instances it groups.
    cand = build_adoption_candidate(parsed, parsed["instances"][0])
    assert cand["adoptable"] is False


def test_extra_global_defs_are_reported_as_losses():
    text = HANDWRITTEN.replace("    script_user root",
                               "    script_user root\n    router_id LVS_DEVEL")
    parsed = parse_keepalived_conf(text)
    directives = " ".join(d["directive"] for d in parsed["unsupported"])
    assert "global_defs/router_id LVS_DEVEL" in directives
    assert parsed["global_defs"]["router_id"] == "LVS_DEVEL"


def test_multiple_instances_yield_one_candidate_each():
    text = HANDWRITTEN + """
vrrp_instance VI_2 {
    state BACKUP
    interface eth0
    virtual_router_id 52
    priority 100
    advert_int 1
    virtual_ipaddress { 10.0.0.200/24 dev eth0 }
}
"""
    analysed = analyse_keepalived_conf(text)
    assert analysed["instance_count"] == 2
    names = [c["instance_name"] for c in analysed["candidates"]]
    assert names == ["VI_1", "VI_2"]
    assert [c["vip"]["virtual_ip"] for c in analysed["candidates"]] == ["10.0.0.100", "10.0.0.200"]
    assert all(c["adoptable"] for c in analysed["candidates"])


def test_unbalanced_braces_raise():
    for bad in ("vrrp_instance VI_1 {\n    state MASTER\n", "}\n"):
        raised = False
        try:
            parse_keepalived_conf(bad)
        except KeepalivedParseError:
            raised = True
        assert raised, f"should have raised for {bad!r}"


def test_our_own_render_round_trips_with_zero_blockers():
    """The invariant that keeps the parser honest: a config WE generated must parse back into
    the same model with nothing unsupported. If a future change to render_keepalived_conf emits
    a directive the parser does not know, this fails — instead of adoption silently reporting
    that OpenManager's own output is unadoptable."""
    vip = {"id": 3, "name": "web-vip", "virtual_ip": "10.0.0.100", "prefix_length": 24,
           "virtual_router_id": 51, "advert_int": 1, "use_unicast": True, "track_haproxy": True}
    members = [{"role": "MASTER", "priority": 150, "network_interface": "eth0",
                "agent_id": 1, "ip_address": "10.0.0.11"},
               {"role": "BACKUP", "priority": 100, "network_interface": "eth0",
                "agent_id": 2, "ip_address": "10.0.0.12"}]
    rendered = kc.render_keepalived_conf(
        vip=vip, members=members, this_agent=members[0],
        peer_ips=["10.0.0.12"], auth_pass_plain="s3cr3t")

    cand = _only_candidate(rendered)
    assert cand["adoptable"] is True, cand["blockers"]
    assert cand["vip"]["virtual_ip"] == vip["virtual_ip"]
    assert cand["vip"]["prefix_length"] == vip["prefix_length"]
    assert cand["vip"]["virtual_router_id"] == vip["virtual_router_id"]
    assert cand["vip"]["track_haproxy"] is True and cand["vip"]["use_unicast"] is True
    assert cand["member"] == {"network_interface": "eth0", "role": "MASTER", "priority": 150}
    assert cand["vip"]["auth_pass"] == "s3cr3t"

    # And the same for the no-auth / multicast / untracked shape, which renders fewer blocks.
    plain = kc.render_keepalived_conf(
        vip={**vip, "use_unicast": False, "track_haproxy": False},
        members=members, this_agent=members[1], peer_ips=[], auth_pass_plain=None)
    cand2 = _only_candidate(plain)
    assert cand2["adoptable"] is True, cand2["blockers"]
    assert cand2["vip"]["use_unicast"] is False and cand2["vip"]["track_haproxy"] is False
    assert cand2["vip"]["auth_pass"] is None