FabledSteward/tests/integration/test_ansible_overlap.py

"""Integration: overlapping Ansible runs (#5158).

Two runs on one host take turns, a schedule whose run is still going skips its
firing, and an alert rule whose action run is still going doesn't start
another. Needs Postgres for the run, schedule and alert rows.
"""
from __future__ import annotations

import asyncio
import os
import sys
import uuid
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace

import pytest

pytestmark = pytest.mark.integration

RECAP = ("PLAY RECAP ****\\n"
         "localhost : ok=1 changed=0 unreachable=0 failed=0 skipped=0 rescued=0 ignored=0")

# Notes when it starts and ends in a shared file, so the order shows overlap.
TURN = (
    "import sys, time\n"
    "open(sys.argv[1], 'a').write('start\\n')\n"
    "time.sleep(1)\n"
    "open(sys.argv[1], 'a').write('end\\n')\n"
    f"print('{RECAP}')\n"
)

INVENTORY = "all:\n  hosts:\n    box:\n      ansible_host: 10.99.0.7\n"


@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 _new_run(app, source: str, status=None) -> str:
    from steward.models.ansible import AnsibleRun, AnsibleRunStatus
    run_id = str(uuid.uuid4())
    async with app.db_sessionmaker() as s:
        async with s.begin():
            s.add(AnsibleRun(id=run_id, playbook_path="t.yml", inventory_path=None,
                             source_name=source, triggered_by=None,
                             status=status or AnsibleRunStatus.running))
    return run_id


async def _set_run(app, run_id, status) -> None:
    from sqlalchemy import update
    from steward.models.ansible import AnsibleRun
    async with app.db_sessionmaker() as s:
        async with s.begin():
            await s.execute(update(AnsibleRun).where(AnsibleRun.id == run_id)
                            .values(status=status))


def test_two_runs_on_one_host_take_turns(app, monkeypatch, tmp_path):
    from sqlalchemy import select
    from steward.ansible import executor
    from steward.models.ansible import AnsibleRun, AnsibleRunStatus
    turns = tmp_path / "turns"
    monkeypatch.setattr(executor, "build_ansible_command",
                        lambda *_a, **_k: [sys.executable, "-c", TURN, str(turns)])
    source = f"overlap-{uuid.uuid4().hex[:8]}"

    async def _go():
        first, second = await _new_run(app, source), await _new_run(app, source)
        await asyncio.wait_for(asyncio.gather(
            executor.start_run(app, first, "t.yml", "", str(tmp_path),
                               inventory_content=INVENTORY),
            executor.start_run(app, second, "t.yml", "", str(tmp_path),
                               inventory_content=INVENTORY),
        ), 60)
        async with app.db_sessionmaker() as s:
            return (await s.execute(select(AnsibleRun).where(
                AnsibleRun.source_name == source))).scalars().all()

    runs = asyncio.run(_go())
    assert turns.read_text().split() == ["start", "end", "start", "end"]
    assert {r.status for r in runs} == {AnsibleRunStatus.success}
    assert any("waiting for run" in (r.output or "") for r in runs)


def test_a_schedule_skips_while_its_last_run_is_going(app, monkeypatch):
    from steward.ansible import scheduler
    from steward.models.ansible import AnsibleRunStatus
    from steward.models.ansible_schedule import AnsibleSchedule
    source = f"sched-{uuid.uuid4().hex[:8]}"
    fired: list[str] = []

    async def fake_trigger(_app, **kw):
        fired.append(kw["playbook_path"])
        return SimpleNamespace(id=str(uuid.uuid4())), None, None

    monkeypatch.setattr(scheduler, "trigger_run", fake_trigger)
    is_due = scheduler.is_due                    # leave other tests' schedules alone
    monkeypatch.setattr(scheduler, "is_due",
                        lambda s, now: s.source_name == source and is_due(s, now))
    long_ago = datetime.now(timezone.utc) - timedelta(hours=2)

    async def _go():
        running = await _new_run(app, source)
        sid = str(uuid.uuid4())
        async with app.db_sessionmaker() as s:
            async with s.begin():
                s.add(AnsibleSchedule(id=sid, name=source, source_name=source,
                                      playbook_path="t.yml", interval_seconds=60,
                                      enabled=True, last_run_at=long_ago,
                                      last_run_id=running))
        try:
            await scheduler.run_due_schedules(app)
            async with app.db_sessionmaker() as s:
                skipped = await s.get(AnsibleSchedule, sid)
            assert fired == []
            assert skipped.last_error.startswith("Skipped")
            assert skipped.last_run_id == running        # still names the run going
            assert skipped.last_run_at > long_ago        # the cadence holds

            await _set_run(app, running, AnsibleRunStatus.success)
            async with app.db_sessionmaker() as s:
                async with s.begin():
                    (await s.get(AnsibleSchedule, sid)).last_run_at = long_ago
            await scheduler.run_due_schedules(app)
            async with app.db_sessionmaker() as s:
                fresh = await s.get(AnsibleSchedule, sid)
            assert fired == ["t.yml"]
            assert fresh.last_error is None and fresh.last_run_id != running
        finally:
            async with app.db_sessionmaker() as s:
                async with s.begin():
                    await s.delete(await s.get(AnsibleSchedule, sid))

    asyncio.run(_go())


def test_an_alert_action_skips_while_its_last_run_is_going(app, monkeypatch):
    from steward.ansible import runner
    from steward.core import alerts
    from steward.models.alerts import AlertEvent, AlertOperator, AlertRule
    from steward.models.ansible import AnsibleRunStatus
    from steward.models.users import User, UserRole
    tag = uuid.uuid4().hex[:8]
    uid, rule_id = str(uuid.uuid4()), str(uuid.uuid4())
    started: list[str] = []
    action = {"source": "t", "playbook": "fix.yml", "inventory": "steward:all"}

    def rule():
        return AlertRule(id=rule_id, name=f"hot {tag}", source_module="host_agent",
                         resource_name=f"box-{tag}", metric_name="cpu_pct",
                         operator=AlertOperator.gt, threshold=90.0, enabled=True,
                         ansible_action=action, created_by=uid)

    async def _event(run_id=None) -> str:
        ev_id = str(uuid.uuid4())
        async with app.db_sessionmaker() as s:
            async with s.begin():
                s.add(AlertEvent(id=ev_id, rule_id=rule_id, from_state="inactive",
                                 to_state="firing", metric_value=95.0,
                                 ansible_run_id=run_id))
        return ev_id

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                s.add(User(id=uid, username=f"u{tag}", email=f"{tag}@e.test",
                           password_hash="x", role=UserRole.admin, is_active=True))
            async with s.begin():
                s.add(rule())
        going = await _new_run(app, f"alert-{tag}")
        await _event(going)                      # the earlier firing's run
        nxt = await _new_run(app, f"alert-{tag}", AnsibleRunStatus.success)

        async def fake_trigger(_app, **kw):
            started.append(kw["playbook_path"])
            return SimpleNamespace(id=nxt), None, None

        monkeypatch.setattr(runner, "trigger_run", fake_trigger)

        refire = await _event()
        await alerts._run_ansible_action(app, rule(), 95.0, refire)
        async with app.db_sessionmaker() as s:
            skipped = await s.get(AlertEvent, refire)
        assert started == []
        assert f"skipped: run {going}" in skipped.note
        assert skipped.ansible_run_id is None

        await _set_run(app, going, AnsibleRunStatus.success)
        later = await _event()
        await alerts._run_ansible_action(app, rule(), 95.0, later)
        async with app.db_sessionmaker() as s:
            linked = await s.get(AlertEvent, later)
        assert started == ["fix.yml"]
        assert linked.ansible_run_id == nxt

    asyncio.run(_go())