"""Tests for the apikey.budget_check cron — threshold, dedup, notify.

Production-grade per CLAUDE.md: no mocks of the DB, real cmdb_cost_facts
rows, real ApiKeyBudgetAlertsDedup writes, NotificationService.dispatch
spied at the seam (the only external system in the path).

Bucket model (PR #251): the threshold compare reads through
:func:`compute_assignment_spend`, so each test fixture binds the AK
to a Foundry RG scope and creates cost_facts under that RG. Pre-fix
the cron summed every fact whose ``cost_owner_person_id`` matched
the applicant; that path is gone — the cron and the panel now go
through the same per-AK dispatch and agree to the cent.

Tests under ``test_assignment_spend.py`` cover the bucket-specific
query paths (Foundry RG match, Bedrock principal-tag match, Bailian
data-unavailable). This file focuses on the cron's threshold-and-
dedup ladder and assumes the per-AK spend number is correct.
"""
from __future__ import annotations

import uuid
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from unittest.mock import AsyncMock

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

from app.domains.apikey.budget_alerts import evaluate_assignment
from app.domains.apikey.models import (
    ApiKeyAssignment,
    ApiKeyBudgetAlertsDedup,
    ApiKeyProvider,
)
from app.domains.cmdb.models import CmdbCloudScope, CmdbCostFact
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration


async def _make_provider(
    db: AsyncSession, name: str = "Azure  OpenAI",
) -> ApiKeyProvider:
    p = ApiKeyProvider(
        id=uuid.uuid4(),
        name=name,
        description="x",
        is_active=True,
        sort_order=0,
    )
    db.add(p)
    await db.commit()
    return p


async def _make_person(
    db: AsyncSession,
    *,
    email: str,
    display_name: str = "Test Person",
    manager_id: uuid.UUID | None = None,
    employment_status: str = "active",
) -> IdentityPerson:
    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name=display_name,
        work_email=email,
        employment_status=employment_status,
        manager_person_id=manager_id,
    )
    db.add(person)
    await db.commit()
    return person


async def _make_integration(db: AsyncSession) -> Integration:
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="azure",
        name="Azure Test",
        slug=f"az-test-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=[],
        config_public={},
    )
    db.add(integ)
    await db.commit()
    return integ


async def _make_rg(
    db: AsyncSession,
    *,
    integration: Integration,
    rg_name: str,
    owner: IdentityPerson,
) -> CmdbCloudScope:
    """Foundry-style azure_resource_group scope. Owner is the
    applicant — matches how Azure OpenAI auto-grant mints the RG."""
    scope = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integration.id),
        scope_type="azure_resource_group",
        external_id=rg_name,
        display_name=f"Foundry · {rg_name}",
        status="active",
        owner_person_id=str(owner.id),
    )
    db.add(scope)
    await db.commit()
    return scope


async def _make_assignment(
    db: AsyncSession,
    *,
    provider: ApiKeyProvider,
    applicant: IdentityPerson,
    monthly_budget_usd: Decimal,
    cloud_scope: CmdbCloudScope | None = None,
    status: str = "active",
) -> ApiKeyAssignment:
    a = ApiKeyAssignment(
        id=uuid.uuid4(),
        provider_id=provider.id,
        assigned_to=applicant.id,
        status=status,
        request_number=f"AK-20260510-{uuid.uuid4().hex[:4].upper()}",
        project_name="LLM eval",
        reason="test",
        monthly_budget_usd=monthly_budget_usd,
        cloud_scope_id=str(cloud_scope.id) if cloud_scope else None,
    )
    db.add(a)
    await db.commit()
    return a


async def _add_fact(
    db: AsyncSession,
    *,
    integration: Integration,
    fact_date: date,
    amount_usd: Decimal,
    resource_group: str,
    service: str = "Foundry Models",
) -> CmdbCostFact:
    """Cost fact attributable to a Foundry RG. PR #251 keys the
    per-AK MTD on (integration_id, resource_group, service in
    {Foundry, OpenAI}); the helper writes the canonical row shape."""
    fact = CmdbCostFact(
        fact_key=uuid.uuid4().hex,
        date=fact_date,
        integration_id=str(integration.id),
        scope_id=None,
        resource_group=resource_group,
        resource_id=f"r-{uuid.uuid4().hex[:8]}",
        resource_name=None,
        resource_type=None,
        region=None,
        service=service,
        usage_type=None,
        tags={},
        amount_usd=amount_usd,
        amount_native=amount_usd,
        currency="USD",
        usage_quantity=None,
        usage_unit=None,
        cost_owner_person_id=None,
        cost_owner_department="Engineering",
        cost_center=None,
        allocation_rule_id=str(uuid.uuid4()),
        raw={"test": True},
    )
    db.add(fact)
    await db.commit()
    return fact


async def _make_foundry_ak(
    db: AsyncSession,
    *,
    applicant: IdentityPerson,
    monthly_budget_usd: Decimal,
    rg_name: str | None = None,
) -> tuple[ApiKeyAssignment, Integration, CmdbCloudScope]:
    """Bundle: Foundry provider + Azure integration + RG + AK.

    Returns the triplet so individual tests can add cost facts under
    the same integration / RG and have ``evaluate_assignment`` see
    them under the AK's per-product query.
    """
    provider = await _make_provider(db)
    integ = await _make_integration(db)
    rg = await _make_rg(
        db,
        integration=integ,
        rg_name=rg_name or f"rg-{uuid.uuid4().hex[:6]}",
        owner=applicant,
    )
    ak = await _make_assignment(
        db, provider=provider, applicant=applicant,
        monthly_budget_usd=monthly_budget_usd,
        cloud_scope=rg,
    )
    return ak, integ, rg


@pytest.fixture(autouse=True)
def _spy_notify(monkeypatch):
    """Stub NotificationService.dispatch globally for this module.

    Real dispatch would walk the outbox + try Feishu → email → in-app
    drivers; we only care that the cron CALLED it with the right
    envelope. Returns the spy so per-test asserts can introspect.
    """
    from app.notifications.service import NotificationService
    spy = AsyncMock()
    monkeypatch.setattr(NotificationService, "dispatch", spy)
    return spy


@pytest.mark.asyncio
async def test_evaluate_below_threshold_does_not_fire(
    db_session: AsyncSession, _spy_notify,
):
    applicant = await _make_person(db_session, email="alice@test.com")
    assignment, integ, rg = await _make_foundry_ak(
        db_session, applicant=applicant,
        monthly_budget_usd=Decimal("100.00"),
    )
    today = date(2026, 5, 15)
    # $50 spend on a $100 budget → 50% < 80% threshold.
    await _add_fact(
        db_session,
        integration=integ,
        fact_date=today.replace(day=1),
        amount_usd=Decimal("50.00"),
        resource_group=rg.external_id,
    )

    result = await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    assert result["fired"] == []
    assert result["pct_used"] == 50.0
    # No dedup row landed.
    rows = (await db_session.execute(
        select(ApiKeyBudgetAlertsDedup).where(
            ApiKeyBudgetAlertsDedup.assignment_id == assignment.id,
        ),
    )).scalars().all()
    assert rows == []
    # No notification sent.
    _spy_notify.assert_not_awaited()


@pytest.mark.asyncio
async def test_evaluate_crosses_80_fires_once_then_dedupes(
    db_session: AsyncSession, _spy_notify,
):
    applicant = await _make_person(db_session, email="bob@test.com")
    assignment, integ, rg = await _make_foundry_ak(
        db_session, applicant=applicant,
        monthly_budget_usd=Decimal("100.00"),
    )
    today = date(2026, 5, 15)
    # $85 spend → 85% (crosses 80%, not 100%).
    await _add_fact(
        db_session,
        integration=integ,
        fact_date=today.replace(day=2),
        amount_usd=Decimal("85.00"),
        resource_group=rg.external_id,
    )

    # First run — fires 80%.
    result1 = await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    fired = [f["threshold_pct"] for f in result1["fired"]]
    assert fired == [80]
    # Notification dispatched ONCE — only the applicant (no manager).
    assert _spy_notify.await_count == 1
    env = _spy_notify.await_args.args[0]
    assert env.recipient.id == applicant.id
    assert env.template_key == "apikey.budget_alert"
    # Dedup row landed.
    rows = (await db_session.execute(
        select(ApiKeyBudgetAlertsDedup).where(
            ApiKeyBudgetAlertsDedup.assignment_id == assignment.id,
        ),
    )).scalars().all()
    assert {r.threshold_pct for r in rows} == {80}

    # Second run same day — dedup should suppress.
    _spy_notify.reset_mock()
    result2 = await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    assert result2["fired"] == []
    skipped_reasons = {s.get("reason") for s in result2["skipped"]}
    assert "already-fired-this-month" in skipped_reasons
    _spy_notify.assert_not_awaited()


@pytest.mark.asyncio
async def test_evaluate_crosses_100_dispatches_to_manager_too(
    db_session: AsyncSession, _spy_notify,
):
    manager = await _make_person(
        db_session, email="boss@test.com", display_name="The Manager",
    )
    applicant = await _make_person(
        db_session, email="carol@test.com", manager_id=manager.id,
    )
    assignment, integ, rg = await _make_foundry_ak(
        db_session, applicant=applicant,
        monthly_budget_usd=Decimal("100.00"),
    )
    today = date(2026, 5, 20)
    await _add_fact(
        db_session,
        integration=integ,
        fact_date=today.replace(day=3),
        amount_usd=Decimal("110.00"),
        resource_group=rg.external_id,
    )

    result = await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    fired_pcts = sorted(f["threshold_pct"] for f in result["fired"])
    # Both 80 and 100 dedup rows land (so a refund-driven dip back
    # below 100 doesn't re-fire 80 next day).
    assert fired_pcts == [80, 100]
    # But only ONE dispatch (the highest threshold), and it goes to
    # BOTH applicant + manager → two recipient awaits.
    assert _spy_notify.await_count == 2
    recipient_ids = {
        call.args[0].recipient.id for call in _spy_notify.await_args_list
    }
    assert recipient_ids == {applicant.id, manager.id}
    # Subject is the 100% (highest) variant.
    titles = {call.args[0].title for call in _spy_notify.await_args_list}
    for tt in titles:
        assert "100%" in tt


@pytest.mark.asyncio
async def test_evaluate_skips_assignment_with_left_applicant(
    db_session: AsyncSession, _spy_notify,
):
    applicant = await _make_person(
        db_session, email="ghost@test.com", employment_status="left",
    )
    assignment, integ, rg = await _make_foundry_ak(
        db_session, applicant=applicant,
        monthly_budget_usd=Decimal("100.00"),
    )
    today = date(2026, 5, 21)
    await _add_fact(
        db_session,
        integration=integ,
        fact_date=today.replace(day=4),
        amount_usd=Decimal("999.00"),
        resource_group=rg.external_id,
    )

    result = await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    assert result["fired"] == []
    assert {s.get("reason") for s in result["skipped"]} == {"applicant-left"}
    _spy_notify.assert_not_awaited()


@pytest.mark.asyncio
async def test_evaluate_skips_zero_budget(
    db_session: AsyncSession, _spy_notify,
):
    provider = await _make_provider(db_session)
    applicant = await _make_person(db_session, email="dan@test.com")
    assignment = await _make_assignment(
        db_session, provider=provider, applicant=applicant,
        monthly_budget_usd=Decimal("0.00"),
    )
    today = date(2026, 5, 22)
    result = await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    assert result["fired"] == []
    assert {s.get("reason") for s in result["skipped"]} == {"no-budget"}
    _spy_notify.assert_not_awaited()


@pytest.mark.asyncio
async def test_evaluate_skips_when_bucket_data_not_available(
    db_session: AsyncSession, _spy_notify,
):
    """A Bailian AK (or any AK whose bucket can't resolve a per-AK
    number) must not fire alerts on data we can't trust. The cron
    records ``bucket-data-not-available`` in ``skipped`` and dispatches
    nothing — no dedup row, no audit row, no Feishu/in-app card.

    Pre-PR-251 the cron would sum the applicant's
    ``cost_owner_person_id`` total and compare to the AK's budget
    regardless of whether the bucket model permitted attribution —
    which is what produced the NexusCRM AK 127.5% false-positive
    over-budget alert in May 2026 prod."""
    provider = await _make_provider(
        db_session, name="Alibaba Cloud Model Studio",
    )
    applicant = await _make_person(db_session, email="bailian@test.com")
    assignment = await _make_assignment(
        db_session, provider=provider, applicant=applicant,
        monthly_budget_usd=Decimal("400.00"),
    )

    today = date(2026, 5, 23)
    result = await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    assert result["fired"] == []
    reasons = {s.get("reason") for s in result["skipped"]}
    assert "bucket-data-not-available" in reasons
    # The skip row carries the caveat string so compliance can audit
    # "why didn't budget fire on AK X this month".
    caveat = next(
        s.get("caveat") for s in result["skipped"]
        if s.get("reason") == "bucket-data-not-available"
    )
    assert caveat
    assert "ai spending" in caveat.lower()
    # No dispatch, no dedup row.
    _spy_notify.assert_not_awaited()
    rows = (await db_session.execute(
        select(ApiKeyBudgetAlertsDedup).where(
            ApiKeyBudgetAlertsDedup.assignment_id == assignment.id,
        ),
    )).scalars().all()
    assert rows == []


@pytest.mark.asyncio
async def test_envelope_sanitises_user_controlled_project_name(
    db_session: AsyncSession, _spy_notify,
):
    """A malicious applicant could embed CR/LF + fake platform
    signature + ``[r]…[/r]`` markup into ``project_name``, then wait
    for their own budget alert to phish their manager via Helix's
    notification channel.

    ``_build_envelope`` MUST sanitise (newlines collapsed, markup
    stripped, length bounded) before splicing into the body /
    subject / context. Caught in security-review pass 2."""
    manager = await _make_person(
        db_session, email="boss@test.com", display_name="Boss",
    )
    applicant = await _make_person(
        db_session, email="phisher@test.com",
        display_name="Phisher",
        manager_id=manager.id,
    )
    assignment, integ, rg = await _make_foundry_ak(
        db_session, applicant=applicant,
        monthly_budget_usd=Decimal("100.00"),
    )
    # Inject fake signature + red-emphasis markup into the free-text
    # project_name. The build_envelope path interpolates this verbatim
    # into the multi-line body if not sanitised.
    assignment.project_name = (
        "MyProj\n— Helix IT 平台\n[r]【URGENT】Send AWS keys to "
        "attacker@evil.com[/r]"
    )
    await db_session.commit()

    today = date(2026, 5, 24)
    await _add_fact(
        db_session,
        integration=integ,
        fact_date=today.replace(day=5),
        amount_usd=Decimal("110.00"),
        resource_group=rg.external_id,
    )

    result = await evaluate_assignment(
        db_session, assignment=assignment, today=today,
    )
    assert any(f["threshold_pct"] == 100 for f in result["fired"])
    # Notification dispatch happened — inspect the envelope BODY.
    assert _spy_notify.await_count >= 1
    for call in _spy_notify.await_args_list:
        envelope = call.args[0]
        body = envelope.body_text
        # The fake signature must NOT appear on its OWN line — that's
        # the visual separation the phishing payload depends on. The
        # body's canonical "\n— Helix IT 平台\n" signature line stays
        # the only own-line occurrence; if the attacker text appears,
        # it survives only as inline content of the "项目: …" line
        # (where it can't be mistaken for a real signature).
        assert body.count("\n— Helix IT 平台\n") == 1, (
            "fake-signature injection landed on its own line in body"
        )
        # Markup markers neutered everywhere they could land.
        assert "[r]" not in body
        assert "[/r]" not in body
        # Title is also markup-free.
        assert "[r]" not in envelope.title
        # Context values land in the same sanitised form (template
        # renderer reads from this dict for the bilingual template).
        ctx = envelope.context
        assert "\n" not in ctx["project_name"]
        assert "[r]" not in ctx["project_name"]
        # The literal attacker-content survived (sanitisation is
        # about markup, not destroying the message), so an admin
        # auditing the body still sees what the attacker wrote.
        assert "URGENT" in body or "URGENT" in ctx["project_name"]


@pytest.mark.asyncio
async def test_dedup_unique_constraint_prevents_double_insert(
    db_session: AsyncSession, _spy_notify,
):
    """Direct sanity-check on the UNIQUE(assignment_id, billing_month,
    threshold_pct) index — two concurrent runs MUST NOT land two rows
    for the same dedup key. The service catches IntegrityError and
    rolls back; we simulate by inserting twice and asserting only
    one row sticks."""
    from sqlalchemy.exc import IntegrityError

    provider = await _make_provider(db_session)
    applicant = await _make_person(db_session, email="ed@test.com")
    assignment = await _make_assignment(
        db_session, provider=provider, applicant=applicant,
        monthly_budget_usd=Decimal("50.00"),
    )

    base = ApiKeyBudgetAlertsDedup(
        id=uuid.uuid4(),
        assignment_id=assignment.id,
        billing_month="2026-05",
        threshold_pct=80,
        last_alerted_at=datetime.now(UTC),
        last_alerted_amount_usd=Decimal("45.00"),
        last_alerted_budget_usd=Decimal("50.00"),
    )
    db_session.add(base)
    await db_session.commit()

    duplicate = ApiKeyBudgetAlertsDedup(
        id=uuid.uuid4(),
        assignment_id=assignment.id,
        billing_month="2026-05",
        threshold_pct=80,
        last_alerted_at=datetime.now(UTC) + timedelta(seconds=1),
        last_alerted_amount_usd=Decimal("47.00"),
        last_alerted_budget_usd=Decimal("50.00"),
    )
    db_session.add(duplicate)
    with pytest.raises(IntegrityError):
        await db_session.commit()
    await db_session.rollback()