"""Integration: alert state against real Postgres. An acknowledgement holds
while the breach lasts and is on the record with who made it (#5463), an alert
clears after its OK run, and changing what a rule watches (or disabling it)
resets its state with a note and the person on the event."""
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_acknowledge_holds_clear_after_and_reset(app):
    from sqlalchemy import select
    from steward.alerts import service as alert_svc
    from steward.core.alerts import record_metric, record_metrics
    from steward.models.alerts import AlertEvent, AlertRule, AlertState, AlertStateEnum
    from steward.models.metrics import PluginMetric
    from steward.models.users import User, UserRole

    async def _go():
        tag = uuid.uuid4().hex[:8]
        res = f"box-{tag}"
        async with app.db_sessionmaker() as s:
            async with s.begin():
                user = User(username=f"al-{tag}", email=f"al-{tag}@test", password_hash="x",
                            role=UserRole.admin)
                s.add(user)
                await record_metrics(s, [PluginMetric(source_module="host_agent", resource_name=res,
                                                      metric_name="cpu_pct", value=1.0)])
                await s.flush()
                fields = await alert_svc.rule_fields(
                    s, name=f"hot {tag}", source_module="host_agent", resource_name=res,
                    metric_name="cpu_pct", operator=">", threshold=90,
                    consecutive_failures_required=1, consecutive_ok_required=2)
                rule = await alert_svc.add_rule(s, fields, created_by=user.id)
            rule_id, user_id = rule.id, user.id

        async def sample(v):
            async with app.db_sessionmaker() as s:
                async with s.begin():
                    await record_metric(s, "host_agent", res, "cpu_pct", 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()

        async def events():
            async with app.db_sessionmaker() as s:
                rows = (await s.execute(select(AlertEvent).where(AlertEvent.rule_id == rule_id)
                                        .order_by(AlertEvent.transitioned_at))).scalars().all()
            return [(e.from_state, e.to_state, e.note) for e in rows]

        async def last_actor():
            async with app.db_sessionmaker() as s:
                return (await s.execute(select(AlertEvent.actor_id).where(
                    AlertEvent.rule_id == rule_id).order_by(
                    AlertEvent.transitioned_at.desc()).limit(1))).scalar_one()

        await sample(95)
        assert (await state()).state == AlertStateEnum.firing
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await alert_svc.acknowledge(s, rule_id, user_id)
        assert await last_actor() == user_id
        for _ in range(3):
            await sample(97)  # still breached: the acknowledgement holds
        assert (await state()).state == AlertStateEnum.acknowledged
        await sample(10)      # one OK check of the two it needs
        assert (await state()).state == AlertStateEnum.acknowledged
        await sample(10)
        st = await state()
        assert st.state == AlertStateEnum.inactive and st.acknowledged_at is None
        assert await events() == [("inactive", "firing", None), ("firing", "acknowledged", None),
                                  ("acknowledged", "resolved", None)]
        assert await last_actor() is None     # the readings resolved it, not a person

        # Changing the threshold of a firing rule resets it, saying why.
        await sample(95)
        async with app.db_sessionmaker() as s:
            async with s.begin():
                r = await s.get(AlertRule, rule_id)
                fields = await alert_svc.rule_fields(
                    s, name=r.name, source_module=r.source_module, resource_name=r.resource_name,
                    metric_name=r.metric_name, operator=">", threshold=99,
                    consecutive_failures_required=1, consecutive_ok_required=2)
                await alert_svc.apply_rule_change(s, r, fields)
        assert (await state()).state == AlertStateEnum.inactive
        assert (await events())[-1] == ("firing", "resolved", "rule changed: threshold")

        # Renaming alone does not reset; disabling does.
        await sample(100)
        async with app.db_sessionmaker() as s:
            async with s.begin():
                r = await s.get(AlertRule, rule_id)
                await alert_svc.apply_rule_change(s, r, {"name": f"renamed {tag}"})
        assert (await state()).state == AlertStateEnum.firing
        async with app.db_sessionmaker() as s:
            async with s.begin():
                r = await s.get(AlertRule, rule_id)
                await alert_svc.apply_rule_change(s, r, {}, enabled=False, actor_id=user_id)
        assert (await state()).state == AlertStateEnum.inactive
        assert (await events())[-1] == ("firing", "resolved", "rule disabled")
        assert await last_actor() == user_id

        # Re-enabling starts the watch over: silence is measured from now.
        async with app.db_sessionmaker() as s:
            async with s.begin():
                r = await s.get(AlertRule, rule_id)
                await alert_svc.apply_rule_change(s, r, {}, enabled=True)
        st = await state()
        assert (st.state, st.last_sample_at, st.no_data_since) == (AlertStateEnum.inactive, None, None)
        assert (await events())[-1] == ("firing", "resolved", "rule disabled")

    asyncio.run(_go())


def test_record_metrics_marks_each_step(app):
    """A caller timing an ingest can tell the row write from the rule reads
    from the evaluation (#5149)."""
    from steward.alerts import service as alert_svc
    from steward.core.alerts import record_metrics
    from steward.models.metrics import PluginMetric
    from steward.models.users import User, UserRole

    class _Marks:
        def __init__(self):
            self.names = []

        def mark(self, name):
            self.names.append(name)

    def rows(res):
        return [PluginMetric(source_module="host_agent", resource_name=res,
                             metric_name="cpu_pct", value=5.0)]

    async def _go():
        tag = uuid.uuid4().hex[:8]
        watched, unwatched = f"box-{tag}", f"other-{tag}"
        async with app.db_sessionmaker() as s:
            async with s.begin():
                user = User(username=f"mk-{tag}", email=f"mk-{tag}@test", password_hash="x",
                            role=UserRole.admin)
                s.add(user)
                await record_metrics(s, rows(watched))
                await s.flush()
                fields = await alert_svc.rule_fields(
                    s, name=f"hot {tag}", source_module="host_agent", resource_name=watched,
                    metric_name="cpu_pct", operator=">", threshold=90)
                await alert_svc.add_rule(s, fields, created_by=user.id)

        seen = {}
        for res in (watched, unwatched):
            marks = _Marks()
            async with app.db_sessionmaker() as s:
                async with s.begin():
                    await record_metrics(s, rows(res), marks)
            seen[res] = marks.names
        return seen[watched], seen[unwatched]

    watched, unwatched = asyncio.run(_go())
    assert watched == ["flush", "rules", "evaluate"]
    assert unwatched == ["flush", "rules"]   # no rule watches it: nothing to evaluate


def test_a_rule_without_state_gets_one_and_is_watched_again(app):
    """A backup leaves alert state out (System 180, 2026-10-15): a restored
    rule has none, and evaluation skips it until alert_reconcile adds it."""
    from sqlalchemy import delete, func, select
    from steward.alerts import service as alert_svc
    from steward.core.alerts import record_metric
    from steward.models.alerts import AlertOperator, AlertRule, AlertState, AlertStateEnum
    from steward.models.users import User, UserRole

    async def _go():
        tag = uuid.uuid4().hex[:8]
        res = f"restored-{tag}"
        async with app.db_sessionmaker() as s:
            async with s.begin():
                user = User(username=f"rs-{tag}", email=f"rs-{tag}@test", password_hash="x",
                            role=UserRole.admin)
                s.add(user)
                await s.flush()
                rule = AlertRule(name=f"restored {tag}", source_module="host_agent",
                                 resource_name=res, metric_name="cpu_pct",
                                 operator=AlertOperator.gt, threshold=90.0,
                                 consecutive_failures_required=1, enabled=True,
                                 created_by=user.id)
                cover = AlertRule(name=f"all hosts {tag}", source_module="host_agent",
                                  resource_name="", metric_name="cpu_pct", covers="all_hosts",
                                  operator=AlertOperator.gt, threshold=90.0,
                                  consecutive_failures_required=1, enabled=True,
                                  created_by=user.id)
                s.add_all([rule, cover])
                await s.flush()
                await s.execute(delete(AlertState).where(AlertState.rule_id == rule.id))
            ids = (rule.id, cover.id)

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

        added = await alert_svc.ensure_rule_states(app)
        first = (await states(ids[0]), await states(ids[1]))
        await alert_svc.ensure_rule_states(app)
        second = await states(ids[0])
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await record_metric(s, "host_agent", res, "cpu_pct", 99.0)
        async with app.db_sessionmaker() as s:
            state = (await s.execute(select(AlertState.state).where(
                AlertState.rule_id == ids[0]))).scalar_one()
        return added, first, second, state

    added, first, second, state = asyncio.run(_go())
    assert added >= 1
    assert first == (1, 0)              # the covering rule has no state of its own
    assert second == 1                  # a second pass adds nothing
    assert state == AlertStateEnum.firing