"""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())