"""Cross-domain regressions for #2504 terminal privacy and #2507 fairness."""

from __future__ import annotations

import ast
import importlib.util
import uuid
from datetime import UTC, datetime, timedelta
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.approval.models import ApprovalSubmissionIntent
from app.approval.submission_intents import (
    compact_terminal_intent_payload,
    defer_submission_intent_retry,
    reconciliation_due_predicate,
    reconciliation_order,
    submission_fingerprint,
    terminal_intent_resource_id,
)
from app.identity.person_models import IdentityPerson

SERVICE_FILES = (
    "approval/service.py",
    "domains/announcement_request/service.py",
    "domains/vpn_application/service.py",
    "domains/ai_tool_application/service.py",
    "domains/domain_registration/service.py",
    "domains/sso/service.py",
    "identity/consumers/request_service.py",
    "domains/data_access/service.py",
    "domains/apikey/service.py",
    "domains/cmdb/scope_request_service.py",
    "domains/cmdb/dns_request_service.py",
)


def test_terminal_receipt_removes_all_recovery_only_sensitive_fields() -> None:
    semantic = {"reason": "secret reason", "pii_justification": "sensitive"}
    intent = ApprovalSubmissionIntent(
        scope="test",
        applicant_id=uuid.uuid4(),
        idempotency_key="stable-key",
        payload_fingerprint=submission_fingerprint(semantic),
        provider_request_uuid=str(uuid.uuid4()),
        state="completed",
        payload={
            "semantic": semantic,
            "provider": {"identity_value": "ou_secret", "form": "secret form"},
            "finalization": {
                "ip_address": "192.0.2.1",
                "user_agent": "secret agent",
            },
            "provider_response": {"raw": "secret response"},
            "trace_id": "trace-secret",
        },
    )

    compact_terminal_intent_payload(
        intent, resource_type="test_resource", resource_id="resource-1",
    )

    assert intent.payload == {
        "schema": "terminal-v1",
        "resource_type": "test_resource",
        "resource_id": "resource-1",
    }
    assert terminal_intent_resource_id(
        intent, resource_type="test_resource",
    ) == "resource-1"
    assert intent.payload_fingerprint == submission_fingerprint(semantic)


def test_all_eleven_domain_reconcilers_share_terminal_and_fairness_policy() -> None:
    app_root = Path(__file__).parents[2] / "app"
    for relative in SERVICE_FILES:
        source = (app_root / relative).read_text(encoding="utf-8")
        tree = ast.parse(source)
        called = {
            node.func.id
            for node in ast.walk(tree)
            if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
        }
        called_methods = {
            node.func.attr
            for node in ast.walk(tree)
            if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
        }
        assert "compact_terminal_intent_payload" in called, relative
        assert "reconciliation_due_predicate" in called, relative
        assert "reconciliation_order" in called, relative
        if relative == "approval/service.py":
            # Asset recovery owns an external-I/O lease and therefore needs
            # the stricter attempt-token fenced defer variant.
            assert "_defer_asset_reconcile_failure" in called_methods, relative
        else:
            assert "defer_submission_intent_retry" in called, relative


@pytest.mark.asyncio
async def test_batch_plus_one_progresses_after_poison_prefix_is_deferred(
    db_session: AsyncSession,
) -> None:
    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name="Fair intent requester",
        work_email=f"intent-fair-{uuid.uuid4().hex[:8]}@test.com",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()
    now = datetime.now(UTC)
    intent_ids: list[uuid.UUID] = []
    for offset in range(4):
        intent = ApprovalSubmissionIntent(
            id=uuid.uuid4(),
            scope="fairness-test",
            applicant_id=person.id,
            idempotency_key=f"key-{offset}",
            payload_fingerprint="f" * 64,
            provider_request_uuid=str(uuid.uuid4()),
            state="provider_created",
            payload={"provider": {}, "finalization": {}},
            created_at=now + timedelta(seconds=offset),
            updated_at=now + timedelta(seconds=offset),
        )
        db_session.add(intent)
        intent_ids.append(intent.id)
    await db_session.commit()

    async def due(limit: int) -> list[uuid.UUID]:
        return list((await db_session.scalars(
            select(ApprovalSubmissionIntent.id)
            .where(
                ApprovalSubmissionIntent.scope == "fairness-test",
                reconciliation_due_predicate(datetime.now(UTC)),
            )
            .order_by(*reconciliation_order())
            .limit(limit)
        )).all())

    first_tick = await due(3)
    assert first_tick == intent_ids[:3]
    for intent_id in first_tick:
        await defer_submission_intent_retry(
            db_session, intent_id, RuntimeError("poison finalizer"),
        )

    second_tick = await due(3)
    assert intent_ids[3] in second_tick
    assert all(intent_id not in second_tick for intent_id in first_tick)


@pytest.mark.asyncio
async def test_due_retry_progresses_amid_continuous_new_unscheduled_rows(
    db_session: AsyncSession,
) -> None:
    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name="Continuous traffic requester",
        work_email=f"intent-stream-{uuid.uuid4().hex[:8]}@test.com",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()
    now = datetime.now(UTC)
    due_id = uuid.uuid4()
    db_session.add(ApprovalSubmissionIntent(
        id=due_id,
        scope="continuous-fairness-test",
        applicant_id=person.id,
        idempotency_key="due-retry",
        payload_fingerprint="d" * 64,
        provider_request_uuid=str(uuid.uuid4()),
        state="provider_created",
        payload={},
        created_at=now - timedelta(hours=2),
        updated_at=now - timedelta(minutes=5),
        next_attempt_at=now - timedelta(minutes=1),
    ))
    for offset in range(5):
        db_session.add(ApprovalSubmissionIntent(
            id=uuid.uuid4(),
            scope="continuous-fairness-test",
            applicant_id=person.id,
            idempotency_key=f"new-{offset}",
            payload_fingerprint="n" * 64,
            provider_request_uuid=str(uuid.uuid4()),
            state="provider_created",
            payload={},
            created_at=now + timedelta(seconds=offset),
            updated_at=now + timedelta(seconds=offset),
            next_attempt_at=None,
        ))
    await db_session.commit()

    selected = list((await db_session.scalars(
        select(ApprovalSubmissionIntent.id)
        .where(
            ApprovalSubmissionIntent.scope == "continuous-fairness-test",
            reconciliation_due_predicate(datetime.now(UTC)),
        )
        .order_by(*reconciliation_order())
        .limit(1)
    )).all())
    assert selected == [due_id]


def test_migration_449_resumes_after_each_nontransactional_ddl_step() -> None:
    path = (
        Path(__file__).parents[2]
        / "alembic/versions/449_approval_intent_retry_schedule.py"
    )
    spec = importlib.util.spec_from_file_location("migration_449", path)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    mocked_op = MagicMock()
    with (
        patch.object(module, "op", mocked_op),
        patch.object(module, "_column_names", return_value={"next_attempt_at"}),
        patch.object(module, "_index_names", return_value=set()),
    ):
        module.upgrade()
    mocked_op.add_column.assert_not_called()
    mocked_op.create_index.assert_called_once()
    mocked_op.execute.assert_called_once()

    mocked_op.reset_mock()
    with (
        patch.object(module, "op", mocked_op),
        patch.object(module, "_column_names", return_value={"next_attempt_at"}),
        patch.object(
            module,
            "_index_names",
            return_value={"ix_approval_submission_intents_next_attempt_at"},
        ),
    ):
        module.upgrade()
    mocked_op.add_column.assert_not_called()
    mocked_op.create_index.assert_not_called()
    mocked_op.execute.assert_called_once()
    backfill_sql = str(mocked_op.execute.call_args.args[0])
    assert "ELSE payload" in backfill_sql
    assert "last_error = NULL" in backfill_sql
    assert "WHERE state IN ('completed', 'compensated', 'failed')" in backfill_sql
    expected_scope_contracts = {
        "scope = 'asset_request'": ("employee_request", "employee_request_id"),
        "scope = 'api_key_request'": ("api_key_assignment", "idempotency_key"),
        "scope = 'announcement_request'": (
            "announcement_request", "$.finalization.request_id",
        ),
        "scope = 'data_access_group_grant'": (
            "data_access_request", "$.finalization.request_id",
        ),
        "scope LIKE 'cmdb_scope:%'": (
            "cmdb_cloud_scope_request", "$.finalization.request_id",
        ),
        "scope LIKE 'cmdb_dns%'": (
            "cmdb_dns_record_request", "$.finalization.request_id",
        ),
        "scope = 'vpn_application_request'": (
            "vpn_application", "$.finalization.application_id",
        ),
        "scope = 'ai_tool_application_request'": (
            "ai_tool_application", "$.finalization.application_id",
        ),
        "scope = 'domain_registration_request'": (
            "domain_registration", "$.finalization.domain_registration_id",
        ),
        "scope = 'sso_grant_request'": (
            "sso_grant", "$.finalization.grant_id",
        ),
        "scope = 'identity_consumer_request'": (
            "identity_consumer_request", "$.finalization.consumer_request_id",
        ),
    }
    for predicate, (resource_type, resource_expression) in expected_scope_contracts.items():
        assert f"WHEN {predicate} THEN '{resource_type}'" in backfill_sql
        assert resource_expression in backfill_sql