"""Integration (#5756): what a host's agent says happened while it was away
becomes the host's downtime, once, whatever the agent retries.

The registration is seeded and the rows read with plain SQL: importing the
plugin's models again after the app loaded the plugin raises "Table already
defined" (see test_host_agent_alerts)."""
from __future__ import annotations

import asyncio
import hashlib
import os
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 _agent(app, last_seen: datetime):
    from sqlalchemy import text
    from steward.models.hosts import Host
    tag = uuid.uuid4().hex[:8]
    token = f"tok-{tag}-{uuid.uuid4().hex}"
    async with app.db_sessionmaker() as s:
        async with s.begin():
            host = Host(name=f"down-{tag}", address="10.9.9.8")
            s.add(host)
            await s.flush()
            await s.execute(text(
                "INSERT INTO host_agent_registrations "
                "(id, host_id, token_hash, token_created_at, last_seen_at, created_at, updated_at) "
                "VALUES (:id, :host, :hash, now(), :seen, now(), now())"),
                {"id": str(uuid.uuid4()), "host": host.id, "seen": last_seen,
                 "hash": hashlib.sha256(token.encode()).hexdigest()})
            host_id = host.id
    return host_id, token


async def _post(app, token, samples, events):
    resp = await app.test_client().post(
        "/plugins/host_agent/ingest", json={"samples": samples, "events": events},
        headers={"Authorization": f"Bearer {token}"})
    assert resp.status_code == 200, await resp.get_data(as_text=True)


async def _rows(app, host_id):
    from sqlalchemy import text
    async with app.db_sessionmaker() as s:
        async with s.begin():
            return (await s.execute(text(
                "SELECT kind, started_at, ended_at FROM host_downtime "
                "WHERE host_id = :h ORDER BY started_at"), {"h": host_id})).all()


def test_a_reboot_the_agent_announced_is_one_reboot_with_its_length(app):
    async def _go():
        now = datetime.now(timezone.utc)
        stop, boot, back = now - timedelta(minutes=5), now - timedelta(minutes=3), now - timedelta(minutes=2)
        host_id, token = await _agent(app, stop)
        stopping = {"kind": "stopping", "at": stop.isoformat(), "host_shutdown": True}
        # The stopping agent's last post carries no sample, and is retried.
        await _post(app, token, [], [stopping])
        await _post(app, token, [], [stopping])
        assert [(k, e) for k, _, e in await _rows(app, host_id)] == [("reboot", None)]

        started = {"kind": "started", "at": back.isoformat(), "boot_at": boot.isoformat()}
        await _post(app, token, [{"ts": back.isoformat(), "cpu_pct": 3.0}], [started])
        assert await _rows(app, host_id) == [("reboot", stop, back)]

    asyncio.run(_go())


def test_an_agent_restart_on_a_host_that_stayed_up_is_the_agent(app):
    async def _go():
        now = datetime.now(timezone.utc)
        stop, back = now - timedelta(seconds=40), now - timedelta(seconds=30)
        host_id, token = await _agent(app, stop)
        # systemd couldn't say; the boot time, days before the stop, settles it.
        await _post(app, token, [], [{"kind": "stopping", "at": stop.isoformat(), "host_shutdown": None}])
        await _post(app, token, [{"ts": back.isoformat(), "cpu_pct": 3.0}],
                    [{"kind": "started", "at": back.isoformat(),
                      "boot_at": (now - timedelta(days=3)).isoformat()}])
        assert await _rows(app, host_id) == [("agent_stop", stop, back)]

    asyncio.run(_go())


def test_a_boot_nobody_announced_is_a_restart_without_a_shutdown(app):
    async def _go():
        now = datetime.now(timezone.utc)
        last, boot, back = now - timedelta(minutes=20), now - timedelta(minutes=4), now - timedelta(minutes=3)
        host_id, token = await _agent(app, last)
        started = {"kind": "started", "at": back.isoformat(), "boot_at": boot.isoformat()}
        await _post(app, token, [{"ts": back.isoformat(), "cpu_pct": 3.0}], [started])
        await _post(app, token, [{"ts": now.isoformat(), "cpu_pct": 3.0}], [started])
        assert await _rows(app, host_id) == [("unclean", last, back)]

    asyncio.run(_go())


def test_lost_contact_is_the_span_the_agent_reports(app):
    async def _go():
        now = datetime.now(timezone.utc)
        since, until = now - timedelta(minutes=45), now
        host_id, token = await _agent(app, since)
        lost = {"kind": "lost_contact", "from": since.isoformat(), "to": until.isoformat()}
        await _post(app, token, [{"ts": until.isoformat(), "cpu_pct": 3.0}], [lost])
        await _post(app, token, [{"ts": until.isoformat(), "cpu_pct": 3.0}], [lost])
        assert await _rows(app, host_id) == [("lost_contact", since, until)]

    asyncio.run(_go())


def test_a_host_rebooting_inside_its_grace_is_silent_with_reason(app):
    """Its other rules' silence is its agent's, as when it is offline (#5721),
    until the grace runs out; then agent offline speaks for it."""
    from steward.core.alerts import offline_agents
    from steward.models.hosts import Host
    from sqlalchemy import select

    async def _go():
        now = datetime.now(timezone.utc)
        host_id, token = await _agent(app, now - timedelta(minutes=1))
        await _post(app, token, [], [{"kind": "stopping", "at": (now - timedelta(minutes=1)).isoformat(),
                                      "host_shutdown": True}])
        old_id, old_token = await _agent(app, now - timedelta(hours=2))
        await _post(app, old_token, [], [{"kind": "stopping", "at": (now - timedelta(hours=1)).isoformat(),
                                          "host_shutdown": True}])
        async with app.db_sessionmaker() as s:
            async with s.begin():
                away = await offline_agents(s)
                names = dict((await s.execute(select(Host.id, Host.name).where(
                    Host.id.in_([host_id, old_id])))).all())
        assert names[host_id] in away
        assert names[old_id] not in away       # an hour on: past its 10-minute grace

    asyncio.run(_go())