"""Collectors against real rows, plus the cron that drives them.

These are the tests that would have caught the three invented-name bugs
found while writing this domain (a status constant that does not exist,
a context key that is never written, a bucket vocabulary spelled from
memory). Each collector is exercised against actual source rows so the
column and constant names have to be real.
"""
from __future__ import annotations

from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select

from app.domains.automation_exception.collectors import (
    identity_sync_source,
    integration_source,
    notification_source,
    provisioning_source,
    run_collector,
    scheduled_job_source,
)
from app.domains.automation_exception.models import AutomationException
from app.domains.automation_exception.scheduler_handlers import (
    run_automation_exception_collect,
)
from app.domains.automation_exception.taxonomy import (
    CAUSE_CREDENTIAL,
    SEVERITY_CRITICAL,
    SOURCE_INTEGRATION,
    SOURCE_PROVISIONING,
)
from app.domains.provisioning.models import (
    STATUS_FAILED,
    STATUS_PENDING,
    ProvisioningTask,
)
from app.integrations.models import Integration
from app.notifications.models import (
    ACK_NOTE_PREFIX,
    STATUS_DEAD,
    NotificationOutbox,
)
from app.scheduled_jobs.models import (
    RUN_STATUS_FAILED,
    RUN_STATUS_INTERRUPTED,
    RUN_STATUS_OK,
    ScheduledJob,
    ScheduledJobRun,
)

NOW = datetime(2026, 9, 6, 12, 0, tzinfo=UTC)


async def _integration(db, *, slug="acme", enabled=True, status=None, msg=None):
    row = Integration(
        kind="software_provisioning",
        provider="acme",
        name="Acme",
        slug=slug,
        enabled=enabled,
        is_primary=False,
        last_test_status=status,
        last_test_message=msg,
        last_tested_at=NOW if status else None,
    )
    db.add(row)
    await db.flush()
    return row


@pytest.mark.asyncio
class TestIntegrationCollector:
    async def test_a_failed_connection_test_becomes_a_critical_credential_row(
        self, db_session,
    ):
        await _integration(
            db_session, status="failed", msg="401 unauthorized: token expired",
        )
        result = await integration_source.collect(db_session, now=NOW)
        assert len(result.candidates) == 1
        cand = result.candidates[0]
        assert cand.source_ref_id == "acme"
        assert cand.replayable is False
        assert cand.severity_facts["integration_enabled"] is True

    async def test_a_passing_test_is_reported_as_explicit_recovery(self, db_session):
        await _integration(db_session, status="ok")
        result = await integration_source.collect(db_session, now=NOW)
        assert result.candidates == []
        assert "acme" in result.recovered_refs

    async def test_a_never_tested_integration_is_not_an_exception(self, db_session):
        # Otherwise the workbench fills on day one with rows whose only
        # remedy is "click a button", before the nightly probe has run.
        await _integration(db_session, status=None)
        result = await integration_source.collect(db_session, now=NOW)
        assert result.candidates == []
        assert result.recovered_refs == set()


@pytest.mark.asyncio
class TestProvisioningCollector:
    async def _task(self, db, employee, **kw):
        row = ProvisioningTask(
            user_id=employee.id,
            action=kw.pop("action", "grant"),
            target_type="seat",
            source="test",
            execution_mode="auto",
            status=kw.pop("status", STATUS_FAILED),
            error_message=kw.pop("error_message", "429 rate limited"),
            attempt_count=kw.pop("attempt_count", 1),
            **kw,
        )
        db.add(row)
        await db.flush()
        return row

    async def test_a_failed_task_is_collected(self, db_session, employee_user):
        await self._task(db_session, employee_user[0])
        result = await provisioning_source.collect(db_session, now=NOW)
        assert len(result.candidates) == 1
        assert result.candidates[0].replayable is True
        assert result.absence_means_recovery is True

    async def test_a_task_in_transient_backoff_is_collected_too(
        self, db_session, employee_user,
    ):
        # The blind spot this closes: a vendor 429 that succeeds on the
        # third attempt leaves no trace anywhere today, so the platform's
        # own failure counts are a survivorship sample.
        await self._task(
            db_session, employee_user[0],
            status=STATUS_PENDING,
            not_before=NOW + timedelta(minutes=15),
        )
        result = await provisioning_source.collect(db_session, now=NOW)
        assert len(result.candidates) == 1
        assert result.candidates[0].detail["retry_pending"] is True

    async def test_a_fresh_pending_task_is_not_a_failure(
        self, db_session, employee_user,
    ):
        await self._task(
            db_session, employee_user[0],
            status=STATUS_PENDING, attempt_count=0, error_message=None,
        )
        result = await provisioning_source.collect(db_session, now=NOW)
        assert result.candidates == []

    async def test_a_backoff_that_has_elapsed_is_waiting_not_failing(
        self, db_session, employee_user,
    ):
        await self._task(
            db_session, employee_user[0],
            status=STATUS_PENDING,
            not_before=NOW - timedelta(minutes=5),
        )
        result = await provisioning_source.collect(db_session, now=NOW)
        assert result.candidates == []


@pytest.mark.asyncio
class TestNotificationCollector:
    async def _outbox(self, db, person, **kw):
        row = NotificationOutbox(
            recipient_person_id=person.id,
            template_key=kw.pop("template_key", "provisioning.revoke_failure"),
            title="t",
            body_text="b",
            channels=["feishu"],
            failed_channels=kw.pop("failed_channels", ["feishu"]),
            status=kw.pop("status", STATUS_DEAD),
            attempts=5,
            max_attempts=5,
            last_error=kw.pop("last_error", "feishu 401 unauthorized"),
            **kw,
        )
        db.add(row)
        await db.flush()
        return row

    async def test_a_dead_row_is_collected_with_its_real_bucket(
        self, db_session, employee_user,
    ):
        await self._outbox(db_session, employee_user[0])
        result = await notification_source.collect(db_session, now=NOW)
        assert len(result.candidates) == 1
        cand = result.candidates[0]
        assert cand.scope_key == "channel:feishu"
        # Resolved through the notifications domain's own resolver, not
        # an invented ``context['bucket']`` key that is never written.
        assert cand.severity_facts["notification_bucket"] is not None

    async def test_an_acknowledged_dead_row_is_already_closed(
        self, db_session, employee_user,
    ):
        await self._outbox(
            db_session, employee_user[0],
            last_error=f"{ACK_NOTE_PREFIX}: handled out of band",
        )
        result = await notification_source.collect(db_session, now=NOW)
        assert result.candidates == []


@pytest.mark.asyncio
class TestScheduledJobCollector:
    async def _job_with_run(self, db, *, status, started_at, handler="x.y"):
        job = (
            await db.execute(
                select(ScheduledJob).where(ScheduledJob.handler_key == handler),
            )
        ).scalar_one_or_none()
        if job is None:
            job = ScheduledJob(
                name=handler, display_name=handler, handler_key=handler,
                trigger_type="cron", cron_expr="0 * * * *", timezone="UTC",
                enabled=True,
            )
            db.add(job)
            await db.flush()
        run = ScheduledJobRun(
            job_id=job.id, started_at=started_at, finished_at=started_at,
            status=status, triggered_by="scheduler",
            error_message="boom" if status == RUN_STATUS_FAILED else None,
        )
        db.add(run)
        await db.flush()
        return job, run

    async def test_a_failed_run_is_collected_and_is_not_replayable(self, db_session):
        await self._job_with_run(
            db_session, status=RUN_STATUS_FAILED, started_at=NOW,
        )
        result = await scheduled_job_source.collect(db_session, now=NOW)
        assert len(result.candidates) == 1
        assert result.candidates[0].replayable is False
        # Absence can never mean recovery for a historical event.
        assert result.absence_means_recovery is False

    async def test_an_interrupted_run_is_not_a_failure(self, db_session):
        # 386 of 452 "failures" in a 30-day production window were this
        # artefact. Importing them would rebuild the illusion one layer up.
        await self._job_with_run(
            db_session, status=RUN_STATUS_INTERRUPTED, started_at=NOW,
        )
        result = await scheduled_job_source.collect(db_session, now=NOW)
        assert result.candidates == []

    async def test_a_later_success_is_reported_as_explicit_recovery(self, db_session):
        _job, run = await self._job_with_run(
            db_session, status=RUN_STATUS_FAILED, started_at=NOW - timedelta(hours=2),
        )
        await self._job_with_run(
            db_session, status=RUN_STATUS_OK, started_at=NOW,
        )
        result = await scheduled_job_source.collect(db_session, now=NOW)
        assert result.candidates == []
        assert str(run.id) in result.recovered_refs

    async def test_an_acknowledged_failure_is_already_closed(self, db_session):
        _job, run = await self._job_with_run(
            db_session, status=RUN_STATUS_FAILED, started_at=NOW,
        )
        run.acknowledged_at = NOW
        await db_session.flush()
        result = await scheduled_job_source.collect(db_session, now=NOW)
        assert result.candidates == []


@pytest.mark.asyncio
class TestIdentitySyncCollector:
    async def test_no_targets_means_no_candidates(self, db_session):
        result = await identity_sync_source.collect(db_session, now=NOW)
        assert result.candidates == []
        assert result.absence_means_recovery is False


@pytest.mark.asyncio
class TestRunCollectorIsolation:
    async def test_a_collector_that_raises_becomes_a_partial_result(
        self, db_session, monkeypatch,
    ):
        # The load-bearing property: a broken collector must not return
        # an empty list that reads as "all clear" and closes everything.
        async def boom(db, *, now=None):
            raise RuntimeError("source table unreadable")

        monkeypatch.setattr(
            "app.domains.automation_exception.collectors.integration_source.collect",
            boom,
        )
        # The registry captured the original function at import time, so
        # patch the tuple the runner walks.
        import app.domains.automation_exception.collectors as pkg

        monkeypatch.setattr(
            pkg, "COLLECTORS",
            tuple(
                (s, boom if s == SOURCE_INTEGRATION else f)
                for s, f in pkg.COLLECTORS
            ),
        )
        result = await run_collector(db_session, source=SOURCE_INTEGRATION, now=NOW)
        assert result.partial is True
        assert result.candidates == []
        assert result.warnings


@pytest.mark.asyncio
class TestCollectCron:
    async def test_the_cron_walks_every_source_and_reports_per_source(
        self, db_session, employee_user,
    ):
        await _integration(
            db_session, status="failed", msg="401 unauthorized: token expired",
        )
        db_session.add(
            ProvisioningTask(
                user_id=employee_user[0].id,
                action="revoke", target_type="seat", source="test",
                execution_mode="auto", status=STATUS_FAILED,
                error_message="429 rate limited", attempt_count=1,
            ),
        )
        await db_session.flush()

        summary = await run_automation_exception_collect(db_session, {})
        assert summary["sources_ingested"] == 5
        assert summary["sources_failed"] == 0
        assert summary["exceptions_created"] >= 2

        rows = (
            await db_session.execute(select(AutomationException))
        ).scalars().all()
        by_source = {r.source: r for r in rows}
        # An expired credential on a live integration is the one cause
        # that is critical on its own — everything downstream is dead.
        integ = by_source[SOURCE_INTEGRATION]
        assert integ.root_cause == CAUSE_CREDENTIAL
        assert integ.severity == SEVERITY_CRITICAL
        assert SOURCE_PROVISIONING in by_source

    async def test_running_the_cron_twice_is_a_no_op(
        self, db_session, employee_user,
    ):
        await _integration(db_session, status="failed", msg="401 unauthorized")
        await db_session.flush()
        first = await run_automation_exception_collect(db_session, {})
        second = await run_automation_exception_collect(db_session, {})
        assert first["exceptions_created"] == 1
        assert second["exceptions_created"] == 0
        assert second["occurrences_created"] == 0

    async def test_only_sources_config_narrows_the_walk(self, db_session):
        await _integration(db_session, status="failed", msg="401 unauthorized")
        await db_session.flush()
        summary = await run_automation_exception_collect(
            db_session, {"only_sources": [SOURCE_INTEGRATION]},
        )
        assert summary["sources_ingested"] == 1