FabledSteward/tests/integration/test_renames.py

"""Integration: a renamed monitor or host takes its alert rules, maintenance
windows and metric history with it (#5065), through the web edit routes,
against real Postgres."""
from __future__ import annotations

import asyncio
import os
import time
import uuid
from datetime import datetime, timedelta, timezone

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)


async def _operator(app, tag):
    from steward.models.users import User, UserRole
    uid = str(uuid.uuid4())
    async with app.db_sessionmaker() as s:
        async with s.begin():
            s.add(User(id=uid, username=f"ren-{tag}", email=f"ren-{tag}@test",
                       password_hash="x", role=UserRole.operator, is_active=True))
    client = app.test_client()
    async with client.session_transaction() as sess:
        sess.update(user_id=uid, user_role="operator", username=f"ren-{tag}",
                    _csrf="test-token", login_at=time.time())
    return client, uid


CSRF = {"X-CSRF-Token": "test-token"}


def test_a_renamed_monitor_keeps_its_firing_rule_window_and_history(app):
    from sqlalchemy import func, select
    from steward.alerts import service as alert_svc
    from steward.core.alerts import record_metric
    from steward.models.alerts import AlertRule, AlertState, AlertStateEnum, MaintenanceWindow
    from steward.models.metrics import PluginMetric, PluginMetricHourly
    from steward.models.monitors import Monitor

    async def _go():
        tag = uuid.uuid4().hex[:8]
        old, new = f"svc-{tag}", f"svc-{tag}-renamed"
        client, uid = await _operator(app, tag)
        mid = str(uuid.uuid4())
        now = datetime.now(timezone.utc)
        async with app.db_sessionmaker() as s:
            async with s.begin():
                s.add(Monitor(id=mid, name=old, type="tcp", target="10.9.9.9", config_json='{"port": 22}'))
                s.add(Monitor(id=str(uuid.uuid4()), name=f"other-{tag}", type="tcp", target="10.9.9.8",
                              config_json='{"port": 22}'))
                # History under the old name, and an orphan's under the new one.
                s.add(PluginMetric(source_module="tcp", resource_name=old, metric_name="is_up", value=1))
                s.add(PluginMetricHourly(source_module="tcp", resource_name=old, metric_name="is_up",
                                         bucket=now.replace(minute=0, second=0, microsecond=0) - timedelta(days=3),
                                         value_avg=1, value_max=1, sample_count=60))
                s.add(PluginMetric(source_module="tcp", resource_name=new, metric_name="is_up", value=0,
                                   recorded_at=now - timedelta(days=30)))
                await s.flush()
                fields = await alert_svc.rule_fields(
                    s, name=f"down {tag}", source_module="tcp", resource_name=old,
                    metric_name="is_up", operator="<", threshold=1,
                    consecutive_failures_required=1, consecutive_ok_required=1)
                rule = await alert_svc.add_rule(s, fields, created_by=uid)
                win = MaintenanceWindow(name=f"later {tag}", scope_source="tcp", scope_resource=old,
                                        created_by=uid, start_at=now + timedelta(days=1), end_at=now + timedelta(days=1, hours=1))
                s.add(win)
            rule_id, win_id = rule.id, win.id

        async def sample(name, v):
            async with app.db_sessionmaker() as s:
                async with s.begin():
                    await record_metric(s, "tcp", name, "is_up", v)

        async def state():
            async with app.db_sessionmaker() as s:
                return (await s.execute(select(AlertState).where(AlertState.rule_id == rule_id))).scalar_one()

        await sample(old, 0)
        assert (await state()).state == AlertStateEnum.firing

        # Renaming to a name another tcp monitor has is refused, and moves nothing.
        resp = await client.post(f"/monitors/{mid}/edit", headers=CSRF, form={
            "name": f"other-{tag}", "target": "10.9.9.9", "port": "22"})
        assert resp.status_code == 400 and "already named" in await resp.get_data(as_text=True)

        resp = await client.post(f"/monitors/{mid}/edit", headers=CSRF, form={
            "name": new, "target": "10.9.9.9", "port": "22"})
        assert resp.status_code == 302

        async with app.db_sessionmaker() as s:
            assert (await s.get(AlertRule, rule_id)).resource_name == new
            assert (await s.get(MaintenanceWindow, win_id)).scope_resource == new
            raw = (await s.execute(select(PluginMetric.value, PluginMetric.resource_name).where(
                PluginMetric.source_module == "tcp",
                PluginMetric.resource_name.in_([old, new])))).all()
            hourly = (await s.execute(select(func.count()).select_from(PluginMetricHourly).where(
                PluginMetricHourly.source_module == "tcp",
                PluginMetricHourly.resource_name == new))).scalar_one()
        # Old samples (1 and the 0 that fired it) moved; the orphan's 30-day-old 0 was dropped.
        assert {r for _, r in raw} == {new} and sorted(v for v, _ in raw) == [0, 1]
        assert hourly == 1

        # Still firing, and now watching the new name: one OK sample clears it.
        assert (await state()).state == AlertStateEnum.firing
        await sample(old, 1)  # the old name is no one's now
        assert (await state()).state == AlertStateEnum.firing
        await sample(new, 1)
        assert (await state()).state == AlertStateEnum.inactive

    asyncio.run(_go())


def test_a_renamed_host_carries_agent_and_container_metrics_and_their_rules(app):
    from sqlalchemy import select
    from steward.alerts import service as alert_svc
    from steward.models.alerts import AlertRule
    from steward.models.hosts import Host
    from steward.models.metrics import PluginMetric

    async def _go():
        tag = uuid.uuid4().hex[:8]
        old, new = f"h{tag}", f"h{tag}-new"
        client, uid = await _operator(app, tag)
        hid = str(uuid.uuid4())
        rows = [("host_agent", old), ("host_agent", f"{old}:/"), ("host_agent", f"{old}:core0"),
                ("docker", f"{old}/nginx"),
                # Not this host's: a longer name sharing the prefix, and a different source.
                ("host_agent", f"{old}x:/"), ("snmp", old)]
        async with app.db_sessionmaker() as s:
            async with s.begin():
                s.add(Host(id=hid, name=old, address="10.8.8.8"))
                s.add(Host(id=str(uuid.uuid4()), name=f"{old}x", address="10.8.8.9"))
                for src, res in rows:
                    s.add(PluginMetric(source_module=src, resource_name=res, metric_name="disk_used_pct", value=1))
                await s.flush()
                fields = await alert_svc.rule_fields(
                    s, name=f"disk {tag}", source_module="host_agent", resource_name=f"{old}:/",
                    metric_name="disk_used_pct", operator=">", threshold=90,
                    consecutive_failures_required=1, consecutive_ok_required=1)
                rule = await alert_svc.add_rule(s, fields, created_by=uid)
            rule_id = rule.id

        resp = await client.post(f"/hosts/{hid}", headers=CSRF, form={"name": f"{old}x", "address": "10.8.8.8"})
        assert resp.status_code == 400

        resp = await client.post(f"/hosts/{hid}", headers=CSRF, form={"name": new, "address": "10.8.8.8"})
        assert resp.status_code == 302
        async with app.db_sessionmaker() as s:
            got = set((await s.execute(select(PluginMetric.source_module, PluginMetric.resource_name).where(
                PluginMetric.resource_name.like(f"h{tag}%")))).all())
            assert (await s.get(AlertRule, rule_id)).resource_name == f"{new}:/"
        assert got == {("host_agent", new), ("host_agent", f"{new}:/"), ("host_agent", f"{new}:core0"),
                       ("docker", f"{new}/nginx"), ("host_agent", f"{old}x:/"), ("snmp", old)}

    asyncio.run(_go())