FabledSteward/tests/integration/test_alert_covering_rules.py

"""Integration: a rule covering hosts (#5464) keeps one ordinary rule per
covered host, adds one for a host that joins its group, removes one for a host
that leaves, passes edits and on/off down to them, and refuses a change made
to a host's rule alone. Each host's rule fires on its own readings."""
from __future__ import annotations

import asyncio
import os
import uuid

import pytest

pytestmark = pytest.mark.integration


@pytest.fixture
def app():
    if not os.environ.get("STEWARD_DATABASE_URL"):
        pytest.skip("needs Postgres")
    from steward.app import create_app
    return create_app(testing=False)


def test_a_group_rule_keeps_one_rule_per_host_in_the_group(app):
    from sqlalchemy import select
    from steward.alerts import service as alert_svc
    from steward.core.alerts import record_metric
    from steward.models.alerts import AlertRule, AlertState, AlertStateEnum
    from steward.models.ansible_inventory import AnsibleGroup, AnsibleTarget
    from steward.models.hosts import Host
    from steward.models.users import User, UserRole

    async def _go():
        tag = uuid.uuid4().hex[:8]
        names = [f"cv-{tag}-{i}" for i in range(3)]
        async with app.db_sessionmaker() as s:
            async with s.begin():
                user = User(username=f"cv-{tag}", email=f"cv-{tag}@test", password_hash="x",
                            role=UserRole.admin)
                hosts = [Host(name=n, address=f"10.8.{i}.1") for i, n in enumerate(names)]
                s.add_all([user, *hosts])
                await s.flush()
                group = AnsibleGroup(name=f"web {tag}")
                targets = [AnsibleTarget(name=f"t-{n}", address=h.address, host_id=h.id)
                           for n, h in zip(names, hosts)]
                for t in targets[:2]:
                    t.groups = [group]
                s.add_all([group, *targets])
                for n in names:
                    await record_metric(s, "host_agent", n, "cpu_pct", 10.0)
                fields = await alert_svc.rule_fields(
                    s, name=f"Hot {tag}", source_module="host_agent", resource_name="",
                    metric_name="cpu_pct", operator=">", threshold=90,
                    covers="group", cover_group=group.name)
                parent = await alert_svc.add_rule(s, fields, created_by=user.id)
            ids = {"parent": parent.id, "user": user.id, "third": targets[2].id,
                   "group": group.id}

        async def kids():
            async with app.db_sessionmaker() as s:
                rows = (await s.execute(select(AlertRule).where(
                    AlertRule.parent_id == ids["parent"]))).scalars().all()
            return {r.resource_name: r for r in rows}

        async def reconcile():
            async with app.db_sessionmaker() as s:
                async with s.begin():
                    return await alert_svc.reconcile_children(
                        s, await s.get(AlertRule, ids["parent"]))

        # The two hosts in the group, each its own rule; the parent has no state.
        got = await kids()
        assert sorted(got) == names[:2]
        assert got[names[0]].name == f"Hot {tag}: {names[0]}"
        assert got[names[0]].threshold == 90 and got[names[0]].covers == "resource"
        async with app.db_sessionmaker() as s:
            assert (await s.execute(select(AlertState).where(
                AlertState.rule_id == ids["parent"]))).scalar_one_or_none() is None

        # A host's own readings fire its own rule only.
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await record_metric(s, "host_agent", names[0], "cpu_pct", 97.0)
        async with app.db_sessionmaker() as s:
            states = {st.rule_id: st.state for st in (await s.execute(select(AlertState).where(
                AlertState.rule_id.in_([r.id for r in got.values()])))).scalars()}
        assert states[got[names[0]].id] == AlertStateEnum.firing
        assert states[got[names[1]].id] == AlertStateEnum.inactive

        # The third host joins the group: covered on the next reconcile.
        async with app.db_sessionmaker() as s:
            async with s.begin():
                from sqlalchemy.orm import selectinload
                t = (await s.execute(select(AnsibleTarget).where(AnsibleTarget.id == ids["third"])
                                     .options(selectinload(AnsibleTarget.groups)))).scalar_one()
                t.groups = [await s.get(AnsibleGroup, ids["group"])]
        assert await reconcile() == ([names[2]], [])
        assert sorted(await kids()) == names

        # Raising the threshold passes down; the firing host's alert resets.
        async with app.db_sessionmaker() as s:
            async with s.begin():
                p = await s.get(AlertRule, ids["parent"])
                fields = await alert_svc.rule_fields(
                    s, name=p.name, source_module="host_agent", resource_name="",
                    metric_name="cpu_pct", operator=">", threshold=99,
                    covers="group", cover_group=p.cover_group)
                await alert_svc.apply_rule_change(s, p, fields, actor_id=ids["user"])
                await alert_svc.reconcile_children(s, p, actor_id=ids["user"])
        got = await kids()
        assert {r.threshold for r in got.values()} == {99}
        async with app.db_sessionmaker() as s:
            st = (await s.execute(select(AlertState).where(
                AlertState.rule_id == got[names[0]].id))).scalar_one()
        assert st.state == AlertStateEnum.inactive

        # Disabling the parent disables every host's rule.
        async with app.db_sessionmaker() as s:
            async with s.begin():
                p = await s.get(AlertRule, ids["parent"])
                await alert_svc.apply_rule_change(s, p, {}, enabled=False)
                await alert_svc.reconcile_children(s, p)
        assert not any(r.enabled for r in (await kids()).values())

        # A host leaving the group loses its rule.
        async with app.db_sessionmaker() as s:
            async with s.begin():
                t = (await s.execute(select(AnsibleTarget).where(AnsibleTarget.id == ids["third"])
                                     .options(selectinload(AnsibleTarget.groups)))).scalar_one()
                t.groups = []
        assert await reconcile() == ([], [names[2]])
        assert sorted(await kids()) == names[:2]

    asyncio.run(_go())


def test_a_host_rule_is_changed_through_its_covering_rule(app):
    """Over MCP, a host's rule under a covering rule refuses an edit made to it
    alone, and the covering rule lists the hosts it covers."""
    from steward.alerts import service as alert_svc
    from steward.core.alerts import record_metric
    from steward.core.api_tokens import Caller
    from steward.mcp import _context
    from steward.mcp.tools import alerts as a
    from steward.models.hosts import Host
    from steward.models.users import User, UserRole

    async def _go():
        tag = uuid.uuid4().hex[:8]
        name = f"cvm-{tag}"
        async with app.db_sessionmaker() as s:
            async with s.begin():
                user = User(username=f"cvm-{tag}", email=f"cvm-{tag}@test", password_hash="x",
                            role=UserRole.admin)
                s.add_all([user, Host(name=name, address="10.8.9.1")])
                await s.flush()
                await record_metric(s, "host_agent", name, "seconds_since_seen", 5.0)
            uid = user.id
        _context._app_ctx.set(app)
        _context._caller_ctx.set(Caller(user_id=uid, username=f"cvm-{tag}",
                                        role=UserRole.admin, token_id=7))
        made = await a.create_alert_rule(f"Offline {tag}", "host_agent", "",
                                         "seconds_since_seen", ">", 180, covers="all_hosts")
        assert made["covers"] == "all_hosts" and name in made["hosts"]
        with pytest.raises(alert_svc.PartOfCoveringRule, match="change or disable it there"):
            await a.update_alert_rule(f"Offline {tag}: {name}", threshold=60)
        listed = {r["rule_name"]: r for r in (await a.list_alert_rules(limit=500))["rules"]}
        assert listed[f"Offline {tag}: {name}"]["part_of"] == f"Offline {tag}"
        await a.update_alert_rule(f"Offline {tag}", threshold=300)
        listed = {r["rule_name"]: r for r in (await a.list_alert_rules(limit=500))["rules"]}
        assert listed[f"Offline {tag}: {name}"]["condition"].startswith("seconds_since_seen > 300")

    asyncio.run(_go())