"""Integration: syslog lines stored against Postgres (#5999). A new sender is
offered and its lines counted, not kept; an added source's lines are kept;
the ring keeps each source within its age window and byte cap."""
from __future__ import annotations

import asyncio
import os
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)


def _mod(sub):
    """The loaded plugin's module: importing plugins.syslog.models by path
    would define its tables a second time (#5775)."""
    from steward.core.plugin_manager import loaded_plugin_submodule
    module = loaded_plugin_submodule("syslog", sub)
    assert module is not None, "the syslog plugin isn't loaded"
    return module


def _received(address, text, at):
    receiver = _mod("receiver")
    return receiver.Received(at, address, _mod("parse").parse(f"<11>Oct 19 01:12:03 synnas kernel: {text}"))


def test_an_unknown_sender_is_offered_and_an_added_one_kept(app):
    from sqlalchemy import delete, func, select

    SyslogLine, SyslogSource = _mod("models").SyslogLine, _mod("models").SyslogSource
    store_lines = _mod("receiver").store_lines

    now = datetime(2026, 10, 19, 3, 0, tzinfo=timezone.utc)

    async def go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await s.execute(delete(SyslogSource))
                await store_lines(s, [_received("10.9.0.1", "first", now),
                                      _received("10.9.0.1", "second", now + timedelta(seconds=1))])
        async with app.db_sessionmaker() as s:
            src = (await s.execute(select(SyslogSource))).scalar_one()
            assert (src.state, src.lines_total, src.hostname_seen) == ("offered", 2, "synnas")
            assert src.last_line == "second" and src.last_severity == 3
            assert (await s.execute(select(func.count()).select_from(SyslogLine))).scalar_one() == 0
            src_id = src.id
        async with app.db_sessionmaker() as s:
            async with s.begin():
                (await s.get(SyslogSource, src_id)).state = "accepted"
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await store_lines(s, [_received("10.9.0.1", "kept", now + timedelta(seconds=2))])
        async with app.db_sessionmaker() as s:
            lines = (await s.execute(select(SyslogLine))).scalars().all()
            assert [(ln.app, ln.hostname, ln.message, ln.severity) for ln in lines] == [("kernel", "synnas", "kept", 3)]
            assert (await s.get(SyslogSource, src_id)).lines_total == 3
            async with s.begin():
                await s.execute(delete(SyslogSource))

    asyncio.run(go())


def test_the_ring_keeps_each_source_within_its_window_and_cap(app):
    from sqlalchemy import delete, select

    SyslogLine, SyslogSource = _mod("models").SyslogLine, _mod("models").SyslogSource
    prune_syslog_lines = _mod("retention").prune_syslog_lines

    now = datetime(2026, 10, 19, 3, 0, tzinfo=timezone.utc)
    line40 = "x" * 40                    # cap 100 keeps 3 (prefixes 0, 40, 80)

    async def go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await s.execute(delete(SyslogSource))
                a = SyslogSource(address="10.9.0.2", state="accepted", first_seen_at=now, last_seen_at=now)
                b = SyslogSource(address="10.9.0.3", state="accepted", first_seen_at=now, last_seen_at=now)
                s.add_all([a, b])
                await s.flush()

                def line(src, minutes_ago, text):
                    return SyslogLine(source_id=src.id, received_at=now - timedelta(minutes=minutes_ago),
                                      facility=1, severity=6, message=text)

                s.add(line(a, 60 * 24 * 30, "old"))
                s.add_all([line(a, 6 - i, line40) for i in range(1, 6)])
                s.add_all([line(b, 2, line40), line(b, 1, line40)])
                a_id, b_id = a.id, b.id
        async with app.db_sessionmaker() as s:
            async with s.begin():
                counts = await prune_syslog_lines(s, retention_days=14, max_bytes_per_source=100, now=now)
            kept = (await s.execute(select(SyslogLine.source_id))).scalars().all()
            assert counts == {"age_pruned": 1, "size_pruned": 2}
            assert kept.count(a_id) == 3 and kept.count(b_id) == 2
            async with s.begin():
                await s.execute(delete(SyslogSource))

    asyncio.run(go())