"""Ingest-engine invariants.

The engine's whole value is that a cron can re-scan the same window
every 30 minutes forever without the ledger drifting. These tests hold
that line, plus the three rules whose violation would make the ledger
actively harmful: a partial scan closing live problems, an automatic
pass overwriting a human's verdict, and a ``wont_fix`` decision being
resurrected nightly.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select

from app.domains.automation_exception.collectors.base import (
    Candidate,
    CollectorResult,
)
from app.domains.automation_exception.ingest import (
    CAUSE_SOURCE_MANUAL,
    CAUSE_SOURCE_RULE,
    ingest_source,
    recompute_exceptions,
)
from app.domains.automation_exception.models import (
    AutomationException,
    AutomationExceptionOccurrence,
)
from app.domains.automation_exception.taxonomy import (
    CAUSE_CREDENTIAL,
    CAUSE_VENDOR_POLICY,
    SOURCE_PROVISIONING,
    STATUS_OPEN,
    STATUS_RESOLVED,
    STATUS_WONT_FIX,
)

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


def _candidate(
    ref: str,
    *,
    error: str = "401 unauthorized",
    scope: str = "software:zoom",
    person_id: uuid.UUID | None = None,
    occurred_at: datetime | None = None,
) -> Candidate:
    return Candidate(
        source_ref_type="provisioning_task",
        source_ref_id=ref,
        occurred_at=occurred_at or NOW,
        error_text=error,
        scope_key=scope,
        person_id=person_id,
        replayable=True,
    )


async def _exceptions(db) -> list[AutomationException]:
    return list(
        (await db.execute(select(AutomationException))).scalars().all(),
    )


async def _occurrences(db) -> list[AutomationExceptionOccurrence]:
    return list(
        (await db.execute(select(AutomationExceptionOccurrence))).scalars().all(),
    )


@pytest.mark.asyncio
class TestIdempotency:
    async def test_the_same_scan_twice_changes_nothing(self, db_session):
        result = CollectorResult(
            candidates=[_candidate("t1"), _candidate("t2")],
            absence_means_recovery=True,
        )
        first = await ingest_source(
            db_session, source=SOURCE_PROVISIONING, result=result, now=NOW,
        )
        assert first.exceptions_created == 1
        assert first.occurrences_created == 2

        second = await ingest_source(
            db_session, source=SOURCE_PROVISIONING, result=result, now=NOW,
        )
        assert second.exceptions_created == 0
        assert second.occurrences_created == 0

        assert len(await _exceptions(db_session)) == 1
        assert len(await _occurrences(db_session)) == 2

    async def test_counters_are_recomputed_not_incremented(self, db_session):
        result = CollectorResult(
            candidates=[_candidate("t1"), _candidate("t2")],
            absence_means_recovery=True,
        )
        for _ in range(3):
            await ingest_source(
                db_session, source=SOURCE_PROVISIONING, result=result, now=NOW,
            )
        exc = (await _exceptions(db_session))[0]
        # Three passes over two failing rows is still two occurrences.
        assert exc.occurrence_count == 2
        assert exc.open_occurrence_count == 2

    async def test_two_vendors_with_one_sentence_stay_two_problems(self, db_session):
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[
                    _candidate("t1", scope="software:zoom"),
                    _candidate("t2", scope="software:atlassian"),
                ],
                absence_means_recovery=True,
            ),
            now=NOW,
        )
        assert len(await _exceptions(db_session)) == 2


@pytest.mark.asyncio
class TestRecovery:
    async def test_a_row_that_stops_failing_recovers_and_resolves_the_cluster(
        self, db_session,
    ):
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[_candidate("t1")], absence_means_recovery=True,
            ),
            now=NOW,
        )
        # Next pass: nothing is failing any more.
        stats = await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(candidates=[], absence_means_recovery=True),
            now=NOW + timedelta(minutes=30),
        )
        assert stats.occurrences_recovered == 1
        exc = (await _exceptions(db_session))[0]
        assert exc.status == STATUS_RESOLVED
        assert exc.open_occurrence_count == 0
        assert exc.resolution_kind == "auto"

    async def test_a_partial_scan_may_not_close_anything(self, db_session):
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[_candidate("t1")], absence_means_recovery=True,
            ),
            now=NOW,
        )
        # The collector broke. Its empty list is not evidence of absence,
        # and treating it as such would silently mark every open problem
        # in this source as fixed.
        stats = await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[],
                absence_means_recovery=True,
                partial=True,
                warnings=["collector blew up"],
            ),
            now=NOW + timedelta(minutes=30),
        )
        assert stats.occurrences_recovered == 0
        assert stats.absence_recovery_applied is False
        exc = (await _exceptions(db_session))[0]
        assert exc.status == STATUS_OPEN
        assert exc.open_occurrence_count == 1

    async def test_absence_does_not_close_a_source_that_forbids_it(self, db_session):
        # Cron runs and sync items are history: absence proves nothing.
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[_candidate("t1")], absence_means_recovery=False,
            ),
            now=NOW,
        )
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(candidates=[], absence_means_recovery=False),
            now=NOW + timedelta(minutes=30),
        )
        exc = (await _exceptions(db_session))[0]
        assert exc.open_occurrence_count == 1
        assert exc.status == STATUS_OPEN

    async def test_an_explicit_recovery_is_always_honoured(self, db_session):
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[_candidate("t1")], absence_means_recovery=False,
            ),
            now=NOW,
        )
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[],
                recovered_refs={"t1"},
                absence_means_recovery=False,
            ),
            now=NOW + timedelta(minutes=30),
        )
        exc = (await _exceptions(db_session))[0]
        assert exc.status == STATUS_RESOLVED

    async def test_older_than_the_window_is_left_alone(self, db_session):
        old = NOW - timedelta(days=60)
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[_candidate("old", occurred_at=old)],
                absence_means_recovery=True,
            ),
            now=old,
        )
        # A windowed scan that legitimately did not look that far back
        # must not read its own boundary as recovery.
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[],
                absence_means_recovery=True,
                absence_window_start=NOW - timedelta(days=30),
            ),
            now=NOW,
        )
        exc = (await _exceptions(db_session))[0]
        assert exc.open_occurrence_count == 1


@pytest.mark.asyncio
class TestReopenAndHumanVerdicts:
    async def test_a_new_failure_reopens_a_resolved_cluster_and_counts_it(
        self, db_session,
    ):
        good = CollectorResult(
            candidates=[_candidate("t1")], absence_means_recovery=True,
        )
        await ingest_source(
            db_session, source=SOURCE_PROVISIONING, result=good, now=NOW,
        )
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(candidates=[], absence_means_recovery=True),
            now=NOW + timedelta(minutes=30),
        )
        assert (await _exceptions(db_session))[0].status == STATUS_RESOLVED

        stats = await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[_candidate("t9")], absence_means_recovery=True,
            ),
            now=NOW + timedelta(hours=2),
        )
        assert stats.exceptions_reopened == 1
        exc = (await _exceptions(db_session))[0]
        assert exc.status == STATUS_OPEN
        assert exc.recurrence_count == 1
        assert exc.resolved_at is None

    async def test_wont_fix_stays_closed_even_as_failures_keep_arriving(
        self, db_session,
    ):
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[_candidate("t1")], absence_means_recovery=True,
            ),
            now=NOW,
        )
        exc = (await _exceptions(db_session))[0]
        exc.status = STATUS_WONT_FIX
        exc.resolution_note = "vendor behaves this way by design"
        await db_session.flush()

        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[_candidate("t1"), _candidate("t2")],
                absence_means_recovery=True,
            ),
            now=NOW + timedelta(hours=1),
        )
        await db_session.refresh(exc)
        # Still closed. A decision that gets undone every night is not a
        # decision, it is noise.
        assert exc.status == STATUS_WONT_FIX

    async def test_a_manual_root_cause_is_never_overwritten_by_a_rule(
        self, db_session,
    ):
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[_candidate("t1", error="401 unauthorized")],
                absence_means_recovery=True,
            ),
            now=NOW,
        )
        exc = (await _exceptions(db_session))[0]
        assert exc.root_cause == CAUSE_CREDENTIAL
        assert exc.cause_source == CAUSE_SOURCE_RULE

        exc.root_cause = CAUSE_VENDOR_POLICY
        exc.cause_source = CAUSE_SOURCE_MANUAL
        await db_session.flush()

        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[_candidate("t1", error="401 unauthorized")],
                absence_means_recovery=True,
            ),
            now=NOW + timedelta(hours=1),
        )
        await db_session.refresh(exc)
        assert exc.root_cause == CAUSE_VENDOR_POLICY
        assert exc.cause_source == CAUSE_SOURCE_MANUAL


@pytest.mark.asyncio
class TestAffectedPeople:
    async def test_only_unrecovered_occurrences_count_towards_affected_people(
        self, db_session, employee_user, admin_user,
    ):
        # Both fixtures yield ``(person, token)``.
        a, b = employee_user[0].id, admin_user[0].id
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[
                    _candidate("t1", person_id=a),
                    _candidate("t2", person_id=b),
                ],
                absence_means_recovery=True,
            ),
            now=NOW,
        )
        exc = (await _exceptions(db_session))[0]
        assert exc.affected_person_count == 2

        # One of them recovers.
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[_candidate("t1", person_id=a)],
                absence_means_recovery=True,
            ),
            now=NOW + timedelta(minutes=30),
        )
        await db_session.refresh(exc)
        assert exc.affected_person_count == 1
        assert exc.open_occurrence_count == 1
        assert exc.occurrence_count == 2

    async def test_reconcile_repairs_a_drifted_counter(self, db_session):
        await ingest_source(
            db_session,
            source=SOURCE_PROVISIONING,
            result=CollectorResult(
                candidates=[_candidate("t1")], absence_means_recovery=True,
            ),
            now=NOW,
        )
        exc = (await _exceptions(db_session))[0]
        exc.occurrence_count = 999
        exc.open_occurrence_count = 999
        await db_session.flush()

        await recompute_exceptions(db_session, exception_ids=None, now=NOW)
        await db_session.refresh(exc)
        assert exc.occurrence_count == 1
        assert exc.open_occurrence_count == 1