"""Tests for ``cmdb_resource_creator_service`` + the driver-side
``get_resource_creator`` parsing helpers.

The service-layer tests use a fake driver to avoid hitting real
CloudTrail / Audit Log APIs (which would require live cloud
credentials and would burn rate budget). The driver-side parsing
helpers are tested separately with raw event fixtures so we cover
the wire-shape edge cases.

Service-layer mocking strategy: we monkeypatch
``build_driver_by_id`` (re-exported through the service module's
namespace) to return a ``_FakeCloudDriver`` instance whose
``get_resource_creator`` is configurable per-test. This avoids
touching the real driver-registry / config-decrypt path entirely
— test integration rows can stay empty.
"""
from __future__ import annotations

import json
import uuid
from datetime import UTC, datetime, timedelta

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

from app.domains.cmdb import resource_creator_service
from app.domains.cmdb.models import (
    RESOURCE_CREATOR_STATUS_ERROR,
    RESOURCE_CREATOR_STATUS_NOT_FOUND,
    RESOURCE_CREATOR_STATUS_RESOLVED,
    RESOURCE_CREATOR_STATUS_UNSUPPORTED,
    RESOURCE_CREATOR_STATUS_WINDOW_EXCEEDED,
    CmdbResourceCreator,
)
from app.domains.cmdb.resource_creator_service import (
    get_cached_creator,
    list_unresolved_resources,
    reset_and_resolve_resource_creator,
    resolve_resource_creator,
)
from app.identity.person_models import IdentityPerson
from app.integrations.driver_base import (
    ResourceCreatorInfo,
    ResourceCreatorWindowExceeded,
)
from app.integrations.models import Integration


class _FakeCloudDriver:
    """Stand-in for the real cloud driver in service-layer tests.

    Holds a ``get_resource_creator`` callable that tests swap in to
    cover each branch (resolved / None / NotImplementedError /
    ResourceCreatorWindowExceeded / transient error).

    NOTE: not inheriting from :class:`CloudDriver` because the
    base class has 5+ abstractmethod stubs (``list_servers``,
    ``list_domains``, etc) that the service path doesn't touch
    but Python's abstract-class machinery would refuse to
    instantiate. ``_install_fake_driver`` below patches the
    service's ``CloudDriver`` reference so the
    ``isinstance(driver, CloudDriver)`` gate accepts the fake.
    """

    def __init__(self, get_creator):
        self._get_creator = get_creator
        self.calls = 0

    async def get_resource_creator(self, resource_id, **kwargs):
        self.calls += 1
        return await self._get_creator(resource_id, **kwargs)

    async def close(self):
        pass


def _install_fake_driver(monkeypatch, fake: _FakeCloudDriver) -> None:
    """Wire ``_FakeCloudDriver`` into the service's
    ``build_driver_by_id`` slot for one test."""
    # The service's ``isinstance(driver, CloudDriver)`` gate is the
    # one production check that depends on type identity; patch it
    # to accept our duck-typed fake. If the service later renames
    # the symbol or changes the gate, this fixture breaks loudly
    # (a feature, not a bug — the test enforces the contract).
    monkeypatch.setattr(
        resource_creator_service, "CloudDriver", _FakeCloudDriver,
    )

    async def _build(integration_id, *, session=None):
        return fake

    monkeypatch.setattr(
        resource_creator_service, "build_driver_by_id", _build,
    )


# ── Test fixtures ──────────────────────────────────────────────


@pytest_asyncio.fixture
async def aws_integration(db_session: AsyncSession) -> Integration:
    """Minimal cloud integration row (provider=aws_cloud) so the
    service can resolve to a driver class even though we mock the
    driver in tests below."""
    integ = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="aws_cloud",
        name="Test AWS",
        slug="test-aws-creator",
        enabled=True,
        config_public={},
    )
    db_session.add(integ)
    await db_session.commit()
    await db_session.refresh(integ)
    return integ


@pytest_asyncio.fixture
async def alice_person(db_session: AsyncSession) -> IdentityPerson:
    """A Helix Person row so the service's email→person matching
    has something to match against."""
    person = IdentityPerson(
        id=uuid.uuid4(),
        hris_person_id=f"ext-{uuid.uuid4().hex[:8]}",
        display_name="Alice Cooper",
        work_email="alice@example.com",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.commit()
    await db_session.refresh(person)
    return person


# ── Service: cache + driver-call orchestration ────────────────


@pytest.mark.asyncio
async def test_resolve_persists_resolved_row_and_matches_person(
    db_session: AsyncSession,
    aws_integration: Integration,
    alice_person: IdentityPerson,
    monkeypatch,
):
    """Happy path — driver returns a creator with a known email,
    service persists the row with status='resolved' and stamps
    creator_person_id from the email match."""
    info = ResourceCreatorInfo(
        principal_id="arn:aws:iam::123:user/alice",
        email="alice@example.com",
        display_name="Alice Cooper",
        created_at_cloud=datetime(2026, 5, 1, 10, 0, tzinfo=UTC),
        source="cloudtrail",
        raw={"event_name": "RunInstances"},
    )

    async def _fake_get(resource_id, **kwargs):
        return info

    _install_fake_driver(monkeypatch, _FakeCloudDriver(_fake_get))

    integ_id = str(aws_integration.id)
    row = await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-abc",
    )
    assert row.status == RESOURCE_CREATOR_STATUS_RESOLVED, (
        f"status={row.status} last_error={row.last_error}"
    )
    assert row.source == "cloudtrail"
    assert row.creator_email == "alice@example.com"
    assert str(row.creator_person_id) == str(alice_person.id)
    assert row.creator_principal_id == "arn:aws:iam::123:user/alice"
    assert row.created_at_cloud is not None
    assert row.resolved_at is not None


@pytest.mark.asyncio
async def test_resolve_caches_terminal_status_and_skips_driver(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """Second resolve on a terminal-status row must NOT call the
    driver — that's the cache hit path. The fake driver counts
    calls; second resolve_resource_creator on the same key
    keeps the counter unchanged."""

    async def _fake_get(resource_id, **kwargs):
        return ResourceCreatorInfo(
            principal_id="arn:test", email=None,
            source="cloudtrail",
        )

    fake = _FakeCloudDriver(_fake_get)
    _install_fake_driver(monkeypatch, fake)

    integ_id = str(aws_integration.id)
    await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-cache",
    )
    assert fake.calls == 1

    row2 = await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-cache",
    )
    assert fake.calls == 1  # cache hit — no second driver call
    assert row2.status == RESOURCE_CREATOR_STATUS_RESOLVED


@pytest.mark.asyncio
async def test_resolve_force_refresh_bypasses_cache(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """``force=True`` re-calls the driver even for a cached
    resolved row — used by the dashboard's 'Re-resolve' button.

    Drops ``FORCE_REFRESH_MIN_GAP_SECONDS`` to 0 so the
    rate-limit cooldown doesn't suppress the back-to-back call
    in this test. (Production default 60s is exercised in
    ``test_resolve_force_refresh_respects_cooldown``.)
    """

    async def _fake_get(resource_id, **kwargs):
        return ResourceCreatorInfo(
            principal_id="arn:test", email=None, source="cloudtrail",
        )

    fake = _FakeCloudDriver(_fake_get)
    _install_fake_driver(monkeypatch, fake)
    monkeypatch.setattr(
        resource_creator_service, "FORCE_REFRESH_MIN_GAP_SECONDS", 0,
    )

    integ_id = str(aws_integration.id)
    await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-fr",
    )
    assert fake.calls == 1

    await resolve_resource_creator(
        db_session, integration_id=integ_id,
        resource_id="i-fr", force=True,
    )
    assert fake.calls == 2


@pytest.mark.asyncio
async def test_resolve_force_refresh_respects_cooldown(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """A second ``force=True`` within the
    ``FORCE_REFRESH_MIN_GAP_SECONDS`` window returns the cache
    instead of burning another audit-API call — protects against
    a mashed 'Re-resolve' button draining the rate budget."""

    async def _fake_get(resource_id, **kwargs):
        return ResourceCreatorInfo(
            principal_id="arn:test", email=None, source="cloudtrail",
        )

    fake = _FakeCloudDriver(_fake_get)
    _install_fake_driver(monkeypatch, fake)
    # Keep production default — 60s — but a fresh row's
    # last_attempt_at is "now" so the second force call within
    # the test's <1ms window is well inside the cooldown.

    integ_id = str(aws_integration.id)
    await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-cooldown",
    )
    assert fake.calls == 1

    await resolve_resource_creator(
        db_session, integration_id=integ_id,
        resource_id="i-cooldown", force=True,
    )
    # No second driver call — cooldown suppressed it.
    assert fake.calls == 1


@pytest.mark.asyncio
async def test_resolve_none_result_persists_not_found(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """Driver returning ``None`` (audit log returned no event for
    this resource) → terminal status='not_found'."""

    async def _fake_get(resource_id, **kwargs):
        return None

    _install_fake_driver(monkeypatch, _FakeCloudDriver(_fake_get))

    integ_id = str(aws_integration.id)
    row = await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-old",
    )
    assert row.status == RESOURCE_CREATOR_STATUS_NOT_FOUND
    assert row.creator_email is None
    assert row.resolved_at is None


@pytest.mark.asyncio
async def test_resolve_window_exceeded_persists_terminal_status(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """``ResourceCreatorWindowExceeded`` from the driver →
    terminal ``window_exceeded`` so the cron sweep stops retrying."""

    async def _fake_get(resource_id, **kwargs):
        raise ResourceCreatorWindowExceeded("90-day window exceeded")

    _install_fake_driver(monkeypatch, _FakeCloudDriver(_fake_get))

    integ_id = str(aws_integration.id)
    row = await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-windowed",
    )
    assert row.status == RESOURCE_CREATOR_STATUS_WINDOW_EXCEEDED
    assert "window" in (row.last_error or "").lower()


@pytest.mark.asyncio
async def test_resolve_not_implemented_persists_unsupported(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """Driver raising NotImplementedError → terminal
    ``unsupported`` (cron stops retrying)."""

    async def _fake_get(resource_id, **kwargs):
        raise NotImplementedError("not yet implemented")

    _install_fake_driver(monkeypatch, _FakeCloudDriver(_fake_get))

    integ_id = str(aws_integration.id)
    row = await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-unsup",
    )
    assert row.status == RESOURCE_CREATOR_STATUS_UNSUPPORTED


@pytest.mark.asyncio
async def test_resolve_transient_error_increments_attempt_count(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """Transient driver exception → ``status='error'`` +
    ``attempt_count`` bumped so cron retries are bounded."""

    async def _fake_get(resource_id, **kwargs):
        raise RuntimeError("ThrottlingException")

    _install_fake_driver(monkeypatch, _FakeCloudDriver(_fake_get))

    integ_id = str(aws_integration.id)
    row1 = await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-flaky",
    )
    assert row1.status == RESOURCE_CREATOR_STATUS_ERROR
    assert row1.attempt_count == 1
    assert "ThrottlingException" in (row1.last_error or "")

    row2 = await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-flaky",
    )
    assert row2.status == RESOURCE_CREATOR_STATUS_ERROR
    assert row2.attempt_count == 2


@pytest.mark.asyncio
async def test_resolve_missing_email_leaves_person_id_null(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """Driver returns a principal_id but no email (federated
    identity / IAM user without a recognisable email pattern) →
    ``creator_person_id`` is NULL even though status='resolved'."""

    async def _fake_get(resource_id, **kwargs):
        return ResourceCreatorInfo(
            principal_id="arn:aws:iam::123:user/svc-account",
            email=None, source="cloudtrail",
        )

    _install_fake_driver(monkeypatch, _FakeCloudDriver(_fake_get))

    integ_id = str(aws_integration.id)
    row = await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-fed",
    )
    assert row.status == RESOURCE_CREATOR_STATUS_RESOLVED
    assert row.creator_person_id is None
    assert row.creator_email is None


@pytest.mark.asyncio
async def test_resolve_email_match_is_case_insensitive(
    db_session: AsyncSession,
    aws_integration: Integration,
    alice_person: IdentityPerson,
    monkeypatch,
):
    """Driver may return ``ALICE@EXAMPLE.COM`` (CloudTrail
    upper-cases some federated email claims). Match must be
    case-insensitive against ``identity_persons.work_email``."""

    async def _fake_get(resource_id, **kwargs):
        return ResourceCreatorInfo(
            principal_id="arn:test",
            email="ALICE@example.COM",
            source="cloudtrail",
        )

    _install_fake_driver(monkeypatch, _FakeCloudDriver(_fake_get))

    integ_id = str(aws_integration.id)
    row = await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-case",
    )
    assert row.status == RESOURCE_CREATOR_STATUS_RESOLVED
    assert str(row.creator_person_id) == str(alice_person.id)


@pytest.mark.asyncio
async def test_resolve_unknown_integration_raises_lookup_error(
    db_session: AsyncSession,
):
    """Caller-error: integration doesn't exist → LookupError. The
    HTTP layer maps this to 404 NotFoundError."""
    with pytest.raises(LookupError):
        await resolve_resource_creator(
            db_session,
            integration_id=str(uuid.uuid4()),
            resource_id="i-foo",
        )


@pytest.mark.asyncio
async def test_get_cached_creator_returns_none_for_unseen(
    db_session: AsyncSession,
    aws_integration: Integration,
):
    """Cache miss returns None — no implicit create on read."""
    row = await get_cached_creator(
        db_session,
        integration_id=str(aws_integration.id),
        resource_id="i-never-seen",
    )
    assert row is None


# ── Operator escape hatch: reset + force-resolve ───────────────


@pytest.mark.asyncio
async def test_reset_and_resolve_clears_attempt_cap_and_fires_driver(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """A row stuck at ``attempt_count >= MAX_ATTEMPTS`` is excluded
    from the cron sweep. ``reset_and_resolve_resource_creator`` is
    the operator escape hatch — it must (a) zero ``attempt_count``,
    (b) clear the force-refresh cooldown, and (c) immediately
    re-attempt against the live driver. The post-reset state must
    make the row eligible again for cron pickup.
    """
    from app.domains.cmdb.models import (
        RESOURCE_CREATOR_MAX_ATTEMPTS,
        RESOURCE_CREATOR_SOURCE_CLOUDTRAIL,
    )

    integ_id = str(aws_integration.id)
    # Seed a row at the attempt cap with a recent ``last_attempt_at``
    # so the force-refresh cooldown would normally suppress retry.
    stuck = CmdbResourceCreator(
        integration_id=integ_id,
        resource_id="i-stuck",
        source=RESOURCE_CREATOR_SOURCE_CLOUDTRAIL,
        status=RESOURCE_CREATOR_STATUS_ERROR,
        attempt_count=RESOURCE_CREATOR_MAX_ATTEMPTS,
        last_attempt_at=datetime.now(UTC),
        last_error="ThrottlingException",
    )
    db_session.add(stuck)
    await db_session.commit()

    # Driver now returns a clean event so the retry should land
    # status='resolved'.
    async def _fake_get(resource_id, **kwargs):
        return ResourceCreatorInfo(
            principal_id="arn:aws:iam::123:user/alice",
            email="alice@example.com",
            source="cloudtrail",
        )

    fake = _FakeCloudDriver(_fake_get)
    _install_fake_driver(monkeypatch, fake)

    row = await reset_and_resolve_resource_creator(
        db_session,
        integration_id=integ_id,
        resource_id="i-stuck",
    )

    # Driver was actually called — the cooldown didn't short-circuit
    # this path (the bare ``GET /creator?refresh=true`` path WOULD
    # have suppressed it via :data:`FORCE_REFRESH_MIN_GAP_SECONDS`).
    assert fake.calls == 1
    # Post-resolve state reflects the new outcome.
    assert row.status == RESOURCE_CREATOR_STATUS_RESOLVED
    assert row.creator_email == "alice@example.com"
    # ``attempt_count`` went 5 → 0 → 1 (the resolve increments once).
    # The key invariant the cron eligibility check depends on is
    # ``attempt_count < RESOURCE_CREATOR_MAX_ATTEMPTS`` AFTER the
    # call — anything else means the cron is still locked out.
    assert row.attempt_count < RESOURCE_CREATOR_MAX_ATTEMPTS
    # The pre-reset error message is cleared on the resolved row.
    assert row.last_error is None


@pytest.mark.asyncio
async def test_reset_and_resolve_on_missing_row_creates_one(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """The reset endpoint is idempotent on a missing row — the
    UPDATE no-ops, the subsequent ``resolve_resource_creator`` call
    inserts the row from scratch via its own upsert path.
    """

    async def _fake_get(resource_id, **kwargs):
        return None  # not_found

    _install_fake_driver(monkeypatch, _FakeCloudDriver(_fake_get))

    integ_id = str(aws_integration.id)
    row = await reset_and_resolve_resource_creator(
        db_session,
        integration_id=integ_id,
        resource_id="i-fresh",
    )
    assert row.status == RESOURCE_CREATOR_STATUS_NOT_FOUND
    assert row.attempt_count == 1  # the fresh insert's first attempt


@pytest.mark.asyncio
async def test_reset_and_resolve_unknown_integration_raises_lookup_error(
    db_session: AsyncSession,
):
    """Same contract as ``resolve_resource_creator`` — missing
    integration is a caller error → ``LookupError``. The HTTP layer
    maps to 404 NotFoundError.
    """
    with pytest.raises(LookupError):
        await reset_and_resolve_resource_creator(
            db_session,
            integration_id=str(uuid.uuid4()),
            resource_id="i-foo",
        )


@pytest.mark.asyncio
async def test_reset_and_resolve_when_driver_still_errors_leaves_row_cron_eligible(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """The whole POINT of `reset_and_resolve` is to restore cron
    eligibility for a row stuck at the attempt cap. If the post-
    reset driver call ALSO errors, we must end with
    ``attempt_count < MAX_ATTEMPTS`` AND ``status='error'`` so the
    next cron tick picks up the row again. A regression that
    forgets to increment_attempt on the reset+error path would
    silently leave the row at attempt_count=0 forever
    (cron-eligible but burning rate budget for no reason).
    """
    from app.domains.cmdb.models import (
        RESOURCE_CREATOR_MAX_ATTEMPTS,
        RESOURCE_CREATOR_SOURCE_CLOUDTRAIL,
    )

    integ_id = str(aws_integration.id)
    stuck = CmdbResourceCreator(
        integration_id=integ_id,
        resource_id="i-stuck-2",
        source=RESOURCE_CREATOR_SOURCE_CLOUDTRAIL,
        status=RESOURCE_CREATOR_STATUS_ERROR,
        attempt_count=RESOURCE_CREATOR_MAX_ATTEMPTS,
        last_attempt_at=datetime.now(UTC),
        last_error="OriginalThrottlingException",
    )
    db_session.add(stuck)
    await db_session.commit()

    async def _fake_get(resource_id, **kwargs):
        raise RuntimeError("FreshAuthFailure")

    _install_fake_driver(monkeypatch, _FakeCloudDriver(_fake_get))

    row = await reset_and_resolve_resource_creator(
        db_session,
        integration_id=integ_id,
        resource_id="i-stuck-2",
    )

    assert row.status == RESOURCE_CREATOR_STATUS_ERROR
    # Post-reset+fail: attempt_count must have moved off zero so
    # the row records "this WAS a retry attempt" — AND must be
    # below MAX so the cron picks it back up. The integer between
    # 0 and MAX is the invariant the operator relies on.
    assert 0 < row.attempt_count < RESOURCE_CREATOR_MAX_ATTEMPTS
    # Fresh error message replaces the stale one — the reset
    # branch cleared last_error to NULL first, then the upsert
    # writes the new exception's redacted form.
    assert "FreshAuthFailure" in (row.last_error or "")
    assert "OriginalThrottlingException" not in (row.last_error or "")

    # Cron-eligibility check: list_unresolved_resources must return
    # this row again. (The list reads from ``cmdb_cost_facts`` for
    # the candidate pool, so we have to seed a fresh fact first.)
    from datetime import date as _date

    from app.domains.cmdb.cost_service import ingest_cost_facts
    from app.integrations.driver_base import CostFact

    await ingest_cost_facts(
        db_session,
        integration_id=integ_id,
        facts=[CostFact(
            date=_date.today() - timedelta(days=1),
            integration_id=integ_id, service="EC2",
            amount_usd=1.0, amount_native=1.0, currency="USD",
            scope=None, resource_group="rg",
            resource_id="i-stuck-2",
            resource_name="i-stuck-2", resource_type="ec2",
            region="us-east-1", usage_type="vcpu", tags={},
        )],
    )

    candidates = await list_unresolved_resources(
        db_session, integration_id=integ_id, limit=50,
    )
    assert ("i-stuck-2") in {rid for _, rid in candidates}


@pytest.mark.asyncio
async def test_reset_clears_creator_pii_so_failed_retry_doesnt_leave_stale_email(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """GDPR-erasure invariant: when an operator clicks "Force retry"
    on a previously-resolved row AND the post-reset driver call
    fails, the DB row MUST NOT retain the prior resolve's creator
    PII (email / principal_id / display_name / person_id).

    Why this matters: ``identity_persons.work_email`` can be purged
    via the leaver / GDPR-erasure flow, but
    ``cmdb_resource_creators.creator_email`` is a plain ``String(255)``
    with no FK cascade. Pre-fix, the state transition
    {resolved → reset → fail → error} left ``creator_email`` at the
    old value indefinitely (the API masked it on non-resolved
    rows, but at-rest residue survived).
    """
    from app.domains.cmdb.models import (
        RESOURCE_CREATOR_SOURCE_CLOUDTRAIL,
    )

    integ_id = str(aws_integration.id)
    # Seed a fully-resolved row with creator PII populated.
    resolved = CmdbResourceCreator(
        integration_id=integ_id,
        resource_id="i-resolved-then-reset",
        source=RESOURCE_CREATOR_SOURCE_CLOUDTRAIL,
        status=RESOURCE_CREATOR_STATUS_RESOLVED,
        creator_principal_id="arn:aws:iam::111:user/alice",
        creator_email="alice@example.com",
        creator_display_name="Alice",
        attempt_count=1,
        resolved_at=datetime.now(UTC),
        created_at_cloud=datetime.now(UTC),
        last_attempt_at=datetime.now(UTC),
    )
    db_session.add(resolved)
    await db_session.commit()

    # Post-reset driver fails transiently.
    async def _fake_get(rid, **kw):
        raise RuntimeError("PostResetThrottling")

    _install_fake_driver(monkeypatch, _FakeCloudDriver(_fake_get))

    row = await reset_and_resolve_resource_creator(
        db_session,
        integration_id=integ_id,
        resource_id="i-resolved-then-reset",
    )

    assert row.status == RESOURCE_CREATOR_STATUS_ERROR
    # GDPR-residue assertions: NO creator-bearing column should
    # carry forward from the prior resolved state.
    assert row.creator_email is None
    assert row.creator_principal_id is None
    assert row.creator_display_name is None
    assert row.creator_person_id is None
    assert row.resolved_at is None
    assert row.created_at_cloud is None


@pytest.mark.asyncio
async def test_last_error_sanitization_strips_secrets(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """``last_error`` is returned to every reader passing
    ``can_view_finance OR can_read_cloud`` (finance / cloud-admin /
    management / auditor / global-admin). Driver exceptions from
    boto3 / googleapiclient can embed pre-signed URL params,
    bearer tokens, and AWS access-key IDs. The sanitizer in
    ``resource_creator_service._sanitize_last_error`` MUST scrub
    these before persist.
    """

    async def _fake_get(resource_id, **kwargs):
        # Realistic-shaped failure: AWS botocore wraps the signed
        # URL plus a credential string into the exception repr.
        raise RuntimeError(
            "ClientError: An error occurred (SignatureDoesNotMatch) "
            "when calling LookupEvents operation: "
            "Authorization: Bearer "
            "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ4In0.abcdefghijklmnop "
            "AWSAccessKeyId=AKIAIOSFODNN7EXAMPLE&"
            "Signature=tHe%2BAct%2BSign%2BPayload%2BHere"
        )

    _install_fake_driver(monkeypatch, _FakeCloudDriver(_fake_get))

    integ_id = str(aws_integration.id)
    row = await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-leak",
    )
    assert row.status == RESOURCE_CREATOR_STATUS_ERROR
    # We MUST NOT leak: the bearer token, the access-key ID, or
    # the signature blob.
    err = row.last_error or ""
    assert "eyJhbGciOi" not in err  # JWT prefix
    assert "AKIAIOSFODNN7EXAMPLE" not in err  # AWS access-key shape
    assert "tHe%2BAct%2BSign" not in err  # signature payload
    # Should still be diagnostically useful — the exception class
    # name + error code survives.
    assert "RuntimeError" in err or "SignatureDoesNotMatch" in err
    assert "REDACTED" in err  # scrub marker present
    # The Bearer prefix must be space-separated, not collapsed
    # into ``Bearer=[REDACTED]`` — Round 2 fix for lambda
    # formatting bug. Operators reading audit rows need the
    # original separator preserved.
    assert "Bearer [REDACTED]" in err or "Bearer=" not in err


@pytest.mark.asyncio
async def test_last_error_sanitization_strips_aws_arn_account_id(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """boto3 ``ClientError: AccessDenied`` errors embed the calling
    IAM principal's full ARN including the 12-digit AWS account
    number. Account IDs aren't credentials but are recon-sensitive
    metadata — disclosed to finance / management / auditor readers
    via ``last_error``. Round 2 added ARN-shape redaction.
    """
    async def _fake_get(resource_id, **kwargs):
        raise RuntimeError(
            "ClientError: An error occurred (AccessDenied) when "
            "calling the LookupEvents operation: User: "
            "arn:aws:iam::123456789012:user/finance-svc is not "
            "authorized to perform: cloudtrail:LookupEvents"
        )

    _install_fake_driver(monkeypatch, _FakeCloudDriver(_fake_get))

    integ_id = str(aws_integration.id)
    row = await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-arn",
    )
    err = row.last_error or ""
    # Account id is scrubbed.
    assert "123456789012" not in err
    assert "REDACTED-ACCOUNT-ID" in err
    # ARN scaffolding stays so operators still see "this was an
    # IAM error" not just "[REDACTED] error".
    assert "arn:aws:iam::" in err
    assert "user/finance-svc" in err  # principal path preserved


@pytest.mark.asyncio
async def test_last_error_sanitization_strips_aliyun_and_google_tokens(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """Cross-provider coverage: Aliyun LTAI access-key shape +
    Google ya29 OAuth token shape must both scrub. Round 2 added
    these patterns after Round 2 security agent flagged the gap.
    """
    async def _fake_get(resource_id, **kwargs):
        raise RuntimeError(
            "AliyunError: AccessKeyId LTAI5abcdefghij123456 invalid; "
            "fallback OAuth ya29.a0AfH6SMBabcdefghijklmnopqrstuv"
        )

    _install_fake_driver(monkeypatch, _FakeCloudDriver(_fake_get))

    integ_id = str(aws_integration.id)
    row = await resolve_resource_creator(
        db_session, integration_id=integ_id, resource_id="i-ali-goog",
    )
    err = row.last_error or ""
    assert "LTAI5abcdefghij123456" not in err
    assert "ya29.a0AfH6SMB" not in err
    assert "REDACTED" in err


# ── Cron candidate selection ───────────────────────────────────


@pytest.mark.asyncio
async def test_list_unresolved_skips_synthetic_resource_ids(
    db_session: AsyncSession,
    aws_integration: Integration,
):
    """The cron must not waste API calls on synthetic ingest-side
    ids (``aws-account:...`` / ``aws-principal:...``) that have no
    cloud-side audit-log counterpart."""
    from datetime import date as _date

    from app.domains.cmdb.cost_service import ingest_cost_facts
    from app.integrations.driver_base import CostFact

    integ_id = str(aws_integration.id)
    today = _date.today() - timedelta(days=1)
    facts = [
        CostFact(
            date=today, integration_id=integ_id, service="EC2",
            amount_usd=10.0, amount_native=10.0, currency="USD",
            scope=None,
            resource_group="rg",
            resource_id="i-real-instance",
            resource_name="i-real-instance", resource_type="ec2",
            region="us-east-1", usage_type="vcpu",
            tags={},
        ),
        CostFact(
            date=today, integration_id=integ_id, service="EC2",
            amount_usd=5.0, amount_native=5.0, currency="USD",
            scope=None, resource_group="rg",
            resource_id="aws-account:178:AmazonEC2",
            resource_name="aws-account:178:AmazonEC2",
            resource_type="aws-aggregated", region=None,
            usage_type="aggregate", tags={},
        ),
    ]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )

    candidates = await list_unresolved_resources(
        db_session, integration_id=integ_id, limit=20,
    )
    rids = {r for _, r in candidates}
    assert "i-real-instance" in rids
    assert "aws-account:178:AmazonEC2" not in rids


@pytest.mark.asyncio
async def test_list_unresolved_skips_terminal_rows(
    db_session: AsyncSession,
    aws_integration: Integration,
):
    """Resources with a terminal-status cache row must NOT come
    back as candidates — they've already been resolved (or
    determined unresolvable) and cron mustn't retry them."""
    from datetime import date as _date

    from app.domains.cmdb.cost_service import ingest_cost_facts
    from app.integrations.driver_base import CostFact

    integ_id = str(aws_integration.id)
    today = _date.today() - timedelta(days=1)
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[
            CostFact(
                date=today, integration_id=integ_id, service="EC2",
                amount_usd=1.0, amount_native=1.0, currency="USD",
                scope=None, resource_group="rg",
                resource_id="i-already-done",
                resource_name="i-already-done", resource_type="ec2",
                region="us-east-1", usage_type="vcpu", tags={},
            ),
        ],
    )

    # Pre-seed a resolved row.
    db_session.add(CmdbResourceCreator(
        integration_id=integ_id,
        resource_id="i-already-done",
        source="cloudtrail",
        status=RESOURCE_CREATOR_STATUS_RESOLVED,
        creator_email="x@y.z",
        attempt_count=1,
    ))
    await db_session.commit()

    candidates = await list_unresolved_resources(
        db_session, integration_id=integ_id, limit=20,
    )
    rids = {r for _, r in candidates}
    assert "i-already-done" not in rids


@pytest.mark.asyncio
async def test_list_unresolved_skips_resources_outside_90d_window(
    db_session: AsyncSession,
    aws_integration: Integration,
):
    """Resources whose latest cost_fact is older than the 90-day
    audit-log lookback window must NOT come back as candidates —
    every provider's audit retention defaults to 90d, so attempting
    those just burns rate budget for a guaranteed
    ``window_exceeded`` terminal. Anchors the 2026-05-25 freshness
    filter that bounded the 277k-rid initial sweep cost in prod.
    """
    from datetime import date as _date

    from app.domains.cmdb.cost_service import ingest_cost_facts
    from app.integrations.driver_base import CostFact

    integ_id = str(aws_integration.id)
    fresh_date = _date.today() - timedelta(days=5)
    stale_date = _date.today() - timedelta(days=120)

    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=[
            CostFact(
                date=fresh_date, integration_id=integ_id,
                service="EC2",
                amount_usd=1.0, amount_native=1.0, currency="USD",
                scope=None, resource_group="rg",
                resource_id="i-fresh",
                resource_name="i-fresh", resource_type="ec2",
                region="us-east-1", usage_type="vcpu", tags={},
            ),
            CostFact(
                date=stale_date, integration_id=integ_id,
                service="EC2",
                amount_usd=1.0, amount_native=1.0, currency="USD",
                scope=None, resource_group="rg",
                resource_id="i-stale",
                resource_name="i-stale", resource_type="ec2",
                region="us-east-1", usage_type="vcpu", tags={},
            ),
        ],
    )

    candidates = await list_unresolved_resources(
        db_session, integration_id=integ_id, limit=20,
    )
    rids = {r for _, r in candidates}
    assert "i-fresh" in rids
    assert "i-stale" not in rids

    # The freshness window is configurable per-call — explicit
    # ``candidate_window_days=365`` should re-admit the stale rid.
    candidates_wide = await list_unresolved_resources(
        db_session, integration_id=integ_id, limit=20,
        candidate_window_days=365,
    )
    rids_wide = {r for _, r in candidates_wide}
    assert "i-fresh" in rids_wide
    assert "i-stale" in rids_wide


# ── Handler: per-resolve commit semantics ──────────────────────


@pytest.mark.asyncio
async def test_sweep_handler_commits_each_resolve_before_systemic_raise(
    db_session: AsyncSession,
    aws_integration: Integration,
    monkeypatch,
):
    """The cron handler used to share one ``work_session`` across
    all 200 resolves; when the systemic-failure ``RuntimeError``
    fired at the >50% threshold check the ``async with session``
    exit rolled back every ``_upsert_creator_row`` call, leaving
    ``cmdb_resource_creators`` at zero rows even on a batch where
    most resolves succeeded.

    Regression test: ingest 6 cost_facts (1 will resolve, 5 will
    error → >50% threshold + ≥5 absolute → handler raises). Assert
    that all 6 cmdb_resource_creators rows are durably persisted
    AFTER the handler's RuntimeError — proves the per-iteration
    commit lands rows the instant they're written. The operator-
    facing value of this fix is: when the handler raises, the
    ``last_error`` rows are visible to the admin UI so they can
    actually diagnose WHICH cloud / scope is broken instead of
    seeing an empty table + a generic "≥50% threshold" message.
    """
    from datetime import date as _date

    from sqlalchemy import select as _select

    from app.domains.cmdb.cost_service import ingest_cost_facts
    from app.integrations.driver_base import CostFact
    from app.scheduled_jobs.handlers import (
        run_resource_creator_sweep,
    )

    integ_id = str(aws_integration.id)
    today = _date.today() - timedelta(days=1)
    facts = [
        CostFact(
            date=today, integration_id=integ_id, service="EC2",
            amount_usd=1.0, amount_native=1.0, currency="USD",
            scope=None, resource_group="rg",
            resource_id=f"i-sweep-{i}",
            resource_name=f"i-sweep-{i}", resource_type="ec2",
            region="us-east-1", usage_type="vcpu", tags={},
        )
        for i in range(6)
    ]
    await ingest_cost_facts(
        db_session, integration_id=integ_id, facts=facts,
    )
    await db_session.commit()

    # 5 resources fail, 1 succeeds (>50% + ≥5 → handler raises).
    info_ok = ResourceCreatorInfo(
        principal_id="arn:aws:iam::123:user/x",
        email="x@example.com",
        display_name="X User",
        created_at_cloud=datetime(2026, 5, 1, tzinfo=UTC),
        raw={"ok": True},
    )

    async def get_creator(rid, **kwargs):
        if rid == "i-sweep-0":
            return info_ok
        raise RuntimeError(
            f"simulated driver auth failure for {rid}",
        )

    fake = _FakeCloudDriver(get_creator)
    _install_fake_driver(monkeypatch, fake)

    with pytest.raises(RuntimeError, match=r"50% threshold"):
        await run_resource_creator_sweep(db_session, {"limit": 10})

    # The handler's per-iteration commit must have persisted every
    # row before the threshold check raised. If the fix regresses
    # to a single end-of-handler commit, this assertion drops to
    # zero rows.
    rows = (await db_session.execute(
        _select(CmdbResourceCreator).where(
            CmdbResourceCreator.integration_id == integ_id,
        ),
    )).scalars().all()
    by_status = {r.resource_id: r.status for r in rows}
    assert by_status["i-sweep-0"] == RESOURCE_CREATOR_STATUS_RESOLVED
    for i in range(1, 6):
        assert by_status[f"i-sweep-{i}"] == (
            RESOURCE_CREATOR_STATUS_ERROR
        ), f"i-sweep-{i} status was {by_status.get(f'i-sweep-{i}')}"


# ── Driver-side parsing helpers ────────────────────────────────


def test_cloudtrail_event_extracts_sso_email_from_assumed_role_arn() -> None:
    """AWS SSO assumed-role ARNs embed the email after the last
    `/`. The parser must pull it out so the service can match to
    a Helix Person without an extra resolution step."""
    import json as _json

    from app.integrations.drivers.aws_cloud import AwsCloudDriver
    event = {
        "EventName": "RunInstances",
        "EventTime": datetime(2026, 5, 1, tzinfo=UTC),
        "EventId": "ev-1",
        "Username": "alice",
        "CloudTrailEvent": _json.dumps({
            "userIdentity": {
                "type": "AssumedRole",
                "arn": (
                    "arn:aws:sts::123:assumed-role/"
                    "AWSReservedSSO_DevOps_a1b2c3/alice@pingcap.com"
                ),
                "principalId": "AROA...:alice@pingcap.com",
                "accountId": "123",
            },
        }),
    }
    info = AwsCloudDriver._cloudtrail_event_to_creator_info(event)
    assert info.email == "alice@pingcap.com"
    assert info.display_name == "Alice"
    assert info.source == "cloudtrail"


def test_gcp_audit_entry_to_creator_info_uses_principal_email() -> None:
    """GCP Audit Log entries put the principal at
    protoPayload.authenticationInfo.principalEmail."""
    from app.integrations.drivers.gcp_cloud import GcpCloudDriver

    entry = {
        "logName": "projects/p/logs/cloudaudit.googleapis.com/activity",
        "timestamp": "2026-05-01T10:00:00.123Z",
        "protoPayload": {
            "methodName": "v1.compute.instances.insert",
            "resourceName": "projects/p/zones/us-c/instances/foo",
            "serviceName": "compute.googleapis.com",
            "authenticationInfo": {
                "principalEmail": "bob@example.com",
                "principalSubject": "user:bob@example.com",
            },
        },
    }
    info = GcpCloudDriver._gcp_audit_entry_to_creator_info(entry)
    assert info.email == "bob@example.com"
    assert info.principal_id == "user:bob@example.com"
    assert info.source == "gcp_audit"
    assert info.created_at_cloud is not None


@pytest.mark.asyncio
async def test_aliyun_get_resource_creator_walks_regions_until_hit() -> None:
    """ActionTrail is region-pinned: the bare host
    ``actiontrail.aliyuncs.com`` doesn't resolve in DNS (verified
    2026-05-25 against prod EC2). The driver must walk
    :attr:`_ACTIONTRAIL_REGIONS` so that resources in any region
    get their create event attributed. First region with a non-
    empty match wins; transport-layer failure on one region must
    fall through to the next instead of being surfaced as a fake
    ``not_found``.

    Anchors the 2026-05-25 fix: the intl driver's region tuple is
    ``(ap-southeast-1, us-east-1, eu-central-1)``. We mock
    ``_acs3_call`` to ConnectError on the first region, return an
    empty page on the second, and return the create event on the
    third — assert the result is the event from the third region
    AND that the driver called all three.
    """
    from app.integrations.drivers.alicloud_intl_cloud import (
        AlicloudIntlCloudDriver,
    )

    driver = AlicloudIntlCloudDriver.__new__(AlicloudIntlCloudDriver)
    # The base class' __init__ requires integration row + config /
    # http-client wiring; bypass it for a pure unit test by
    # ``__new__`` then attaching only what ``get_resource_creator``
    # touches.
    driver.integration_id = "intg-test"
    driver.config = {
        "access_key_id": "ak", "access_key_secret": "sk",
    }

    hosts_called: list[str] = []

    async def fake_acs3_call(*, host, **kwargs):
        hosts_called.append(host)
        if host == "actiontrail.ap-southeast-1.aliyuncs.com":
            import httpx
            raise httpx.ConnectError(
                "[Errno -2] Name or service not known",
            )
        if host == "actiontrail.us-east-1.aliyuncs.com":
            return {"Events": [], "NextToken": None}
        if host == "actiontrail.eu-central-1.aliyuncs.com":
            return {
                "Events": [
                    {
                        "EventName": "RunInstances",
                        "EventId": "ev-1",
                        "EventTime": "2026-05-01T10:00:00Z",
                        "EventSource": "ecs.aliyuncs.com",
                        "UserIdentity": {
                            "Type": "RAMUser",
                            "PrincipalId": "12345:bob",
                            "UserName": "bob@example.com",
                            "AccountId": "12345",
                        },
                    },
                ],
                "NextToken": None,
            }
        raise AssertionError(f"unexpected host: {host}")

    driver._acs3_call = fake_acs3_call  # type: ignore[assignment]

    info = await driver.get_resource_creator("i-eu-central-only-id")
    assert info is not None
    assert info.email == "bob@example.com"
    assert hosts_called == [
        "actiontrail.ap-southeast-1.aliyuncs.com",
        "actiontrail.us-east-1.aliyuncs.com",
        "actiontrail.eu-central-1.aliyuncs.com",
    ]


@pytest.mark.asyncio
async def test_aliyun_get_resource_creator_propagates_4xx_without_walking_regions() -> None:
    """A 4xx response from one region (e.g. RAM 403 NoPermission)
    must NOT silently fall through to the next region — that would
    let an auth gap masquerade as a transport blip and the
    operator would never see the permission error. ``_acs3_call``
    wraps 4xx / 5xx as ``RuntimeError`` (NOT one of the httpx
    transport classes the region-fallthrough catch lists), so the
    re-raise in the ``except RuntimeError`` arm propagates past
    the transport-only ``except`` and out of the driver → service
    layer records ``status='error'`` with the auth message intact.

    Regression guard: if someone "helpfully" widens the
    region-fallthrough ``except`` back to a bare ``Exception``,
    this test drops to "called all 3 regions + returned None",
    which is the false-not_found bug class.
    """
    from app.integrations.drivers.alicloud_intl_cloud import (
        AlicloudIntlCloudDriver,
    )

    driver = AlicloudIntlCloudDriver.__new__(AlicloudIntlCloudDriver)
    driver.integration_id = "intg-test"
    driver.config = {
        "access_key_id": "ak", "access_key_secret": "sk",
    }
    hosts_called: list[str] = []

    async def fake_acs3_call(*, host, **kwargs):
        hosts_called.append(host)
        # _acs3_call's real shape: 4xx wrapped as RuntimeError.
        raise RuntimeError(
            "Aliyun LookupEvents 403: {\"Code\": \"NoPermission\", "
            "\"Message\": \"You are not authorized to perform this "
            "operation. Action: actiontrail:LookupEvents\"}",
        )

    driver._acs3_call = fake_acs3_call  # type: ignore[assignment]

    with pytest.raises(RuntimeError, match="NoPermission"):
        await driver.get_resource_creator("i-needs-perms")
    # Walk stops at the first region's auth error — does NOT
    # silently retry on us-east-1 / eu-central-1 just because the
    # first region was "broken". 4xx is operator-actionable.
    assert hosts_called == ["actiontrail.ap-southeast-1.aliyuncs.com"]


@pytest.mark.asyncio
async def test_aliyun_get_resource_creator_returns_none_when_all_regions_clean_but_empty() -> None:
    """Every region reachable and responsive (200 OK) but no
    matching create events anywhere in the 90-day window → return
    ``None``, which the service layer maps to terminal
    ``status='not_found'``. Critical distinction: clean-empty must
    NOT raise (the cron would re-attempt forever) and must NOT
    look like a transport failure. This is the genuine "we looked
    and there's nothing to attribute" path.
    """
    from app.integrations.drivers.alicloud_intl_cloud import (
        AlicloudIntlCloudDriver,
    )

    driver = AlicloudIntlCloudDriver.__new__(AlicloudIntlCloudDriver)
    driver.integration_id = "intg-test"
    driver.config = {
        "access_key_id": "ak", "access_key_secret": "sk",
    }
    hosts_called: list[str] = []

    async def fake_acs3_call(*, host, **kwargs):
        hosts_called.append(host)
        return {"Events": [], "NextToken": None}

    driver._acs3_call = fake_acs3_call  # type: ignore[assignment]

    result = await driver.get_resource_creator("i-genuinely-not-found")
    assert result is None
    # All three regions visited because each came back empty.
    assert len(hosts_called) == 3


@pytest.mark.asyncio
async def test_aliyun_get_resource_creator_refuses_empty_region_tuple() -> None:
    """Defence: a subclass that overrides ``_ACTIONTRAIL_REGIONS``
    to an empty tuple would silently return ``None`` for every
    resource — the operator would see a stream of false
    ``not_found`` and never realise the driver is misconfigured.
    The method must refuse loudly instead.
    """
    from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver

    class _BrokenDriver(AliyunCloudDriver):
        _ACTIONTRAIL_REGIONS = ()  # type: ignore[assignment]

    driver = _BrokenDriver.__new__(_BrokenDriver)
    driver.integration_id = "intg-test"
    driver.config = {
        "access_key_id": "ak", "access_key_secret": "sk",
    }

    # ``NotImplementedError`` (not RuntimeError) so the service-layer
    # ``except NotImplementedError`` arm maps it to terminal
    # ``status='unsupported'`` — no 5-attempt retry waste, HTTP path
    # returns a clean ``unsupported`` payload instead of 500.
    with pytest.raises(
        NotImplementedError,
        match=r"no ActionTrail regions configured",
    ):
        await driver.get_resource_creator("i-anything")


@pytest.mark.asyncio
async def test_aliyun_get_resource_creator_raises_when_all_regions_dns_fail() -> None:
    """When EVERY region's ActionTrail endpoint hits a transport
    error (DNS NXDOMAIN, TCP refused, ReadTimeout), the driver
    must re-raise so the service layer marks the row as
    ``status='error'`` instead of silently returning ``None``
    (which the service maps to a TERMINAL ``not_found`` and the
    cron would never retry — a 2026-05-25 regression class:
    transport-broken resources got buried as "not found" and were
    never re-attempted after a network fix).
    """
    import httpx

    from app.integrations.drivers.alicloud_intl_cloud import (
        AlicloudIntlCloudDriver,
    )

    driver = AlicloudIntlCloudDriver.__new__(AlicloudIntlCloudDriver)
    driver.integration_id = "intg-test"
    driver.config = {
        "access_key_id": "ak", "access_key_secret": "sk",
    }
    err = httpx.ConnectError("[Errno -2] Name or service not known")

    async def fake_acs3_call(*, host, **kwargs):
        raise err

    driver._acs3_call = fake_acs3_call  # type: ignore[assignment]

    with pytest.raises(httpx.ConnectError):
        await driver.get_resource_creator("i-broken")


def test_aliyun_event_to_creator_info_keeps_principal_id() -> None:
    """Aliyun ActionTrail returns UserIdentity.{PrincipalId,
    UserName}. The parser keeps the principal id verbatim and
    extracts an email only when UserName looks like one."""
    from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver

    event = {
        "EventName": "RunInstances",
        "EventId": "ev-1",
        "EventTime": "2026-05-01T10:00:00Z",
        "EventSource": "ecs.aliyuncs.com",
        "UserIdentity": {
            "Type": "RAMUser",
            "PrincipalId": "12345:carol",
            "UserName": "carol@example.com",
            "AccountId": "12345",
        },
    }
    info = AliyunCloudDriver._aliyun_event_to_creator_info(event)
    assert info.email == "carol@example.com"
    assert info.principal_id == "12345:carol"
    assert info.source == "aliyun_actiontrail"


def test_azure_event_to_creator_info_uses_caller_upn() -> None:
    """Azure Activity Log puts the principal at ``caller`` (AAD
    UPN). The parser uses it as the email when it contains ``@``."""
    from app.integrations.drivers.azure_cloud import AzureCloudDriver

    event = {
        "operationName": {"value": "Microsoft.Compute/virtualMachines/write"},
        "eventTimestamp": "2026-05-01T10:00:00Z",
        "caller": "dave@example.com",
        "correlationId": "abc",
        "resourceProviderName": {"value": "Microsoft.Compute"},
        "status": {"value": "Succeeded"},
        "claims": {
            "name": "Dave Davidson",
            "upn": "dave@example.com",
            "oid": "obj-1234",
        },
    }
    info = AzureCloudDriver._azure_event_to_creator_info(
        event, datetime(2026, 5, 1, 10, 0, tzinfo=UTC),
    )
    assert info.email == "dave@example.com"
    assert info.principal_id == "obj-1234"
    assert info.source == "azure_activity"


# ── Driver dispatch (synthetic id shortcut) ────────────────────


def test_gcp_project_extractor_rejects_malformed_project_id() -> None:
    """Defence-in-depth: a corrupt ``resource_id`` ending in
    ``/projects/foo"`` (close-quote injection) must NOT yield a
    project token. The extractor's regex validates the candidate
    before returning, so a Logging-filter caller never interpolates
    an unvalidated value.
    """
    from app.integrations.drivers.gcp_cloud import GcpCloudDriver

    bad = [
        '//compute.googleapis.com/projects/foo"/zones/x',
        '//compute.googleapis.com/projects/X/x',  # uppercase
        '//compute.googleapis.com/projects/1bad/x',  # leading digit
        '//compute.googleapis.com/projects/-bad/x',  # leading hyphen
        '//compute.googleapis.com/projects//x',     # empty
    ]
    for rid in bad:
        assert (
            GcpCloudDriver._gcp_extract_project_from_resource_id(rid)
            is None
        ), rid
    # Happy path: real project id passes.
    assert GcpCloudDriver._gcp_extract_project_from_resource_id(
        "//compute.googleapis.com/projects/my-good-prj/zones/x/instances/y",
    ) == "my-good-prj"


@pytest.mark.asyncio
async def test_azure_filter_escapes_single_quote_in_resource_id(
    monkeypatch,
) -> None:
    """Defence-in-depth: an Azure ARM resource_id containing a
    single quote (rare but possible if a future driver bug or a
    hostile caller crafts one) must be escaped before
    interpolation into the OData ``$filter``.

    Doesn't hit Azure — verifies the filter STRING the driver
    would send by intercepting the HTTP call.
    """
    from app.integrations.drivers.azure_cloud import AzureCloudDriver

    captured: dict[str, str] = {}

    class _StubResp:
        status_code = 200
        text = "{}"
        def json(self):
            return {"value": []}

    class _StubClient:
        async def __aenter__(self):
            return self
        async def __aexit__(self, *args):
            return False
        async def get(self, url, params=None, headers=None):
            captured["filter"] = (params or {}).get("$filter", "")
            return _StubResp()

    driver = AzureCloudDriver.__new__(AzureCloudDriver)
    driver.config = {}
    driver._token_cache = {}
    monkeypatch.setattr(driver, "_token", lambda *_a, **_kw: "tok")
    monkeypatch.setattr(driver, "_http", lambda: _StubClient())

    rid = (
        "/subscriptions/12345678-1234-1234-1234-123456789abc"
        "/resourceGroups/foo'bar/providers/Microsoft.Compute"
        "/virtualMachines/test'vm"
    )

    async def _token(*_a, **_kw):
        return "tok"
    monkeypatch.setattr(driver, "_token", _token)

    await driver.get_resource_creator(rid)
    f = captured.get("filter", "")
    # Each ``'`` must have been doubled to ``''``. Original had 2
    # single-quotes; the escaped string contains 4 in pairs.
    assert "foo''bar" in f, f
    assert "test''vm" in f, f
    # And the start/end quotes wrapping the resourceUri value are
    # still single (one each).
    assert "resourceUri eq '" in f


# ── Aliyun ActionTrail v3 field-case regression (PR #506) ─────────


def test_aliyun_event_to_creator_info_handles_camelcase_v3() -> None:
    """Regression for PR #506: ActionTrail OpenAPI v3 (2020-07-06)
    returns event objects with **camelCase** keys (``eventName``,
    ``eventTime``, ``userIdentity.userName``) even though the
    top-level wrapper is PascalCase (``Events`` / ``NextToken``).
    Pre-fix the parser read ``ev.get("EventName")`` and every field
    came back None → every event filtered out → 3743/3743
    alicloud-intl rows stuck at ``not_found`` in prod for ~weeks.

    Both the PascalCase test above AND this camelCase variant must
    produce the same ``ResourceCreatorInfo``.
    """
    from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver

    event = {
        "eventName": "RunInstances",
        "eventId": "ev-1",
        "eventTime": "2026-05-01T10:00:00Z",
        "eventSource": "ecs.aliyuncs.com",
        "userIdentity": {
            "type": "RAMUser",
            "principalId": "12345:carol",
            "userName": "carol@example.com",
            "accountId": "12345",
        },
    }
    info = AliyunCloudDriver._aliyun_event_to_creator_info(event)
    assert info.email == "carol@example.com"
    assert info.principal_id == "12345:carol"
    assert info.source == "aliyun_actiontrail"
    assert info.created_at_cloud is not None
    # raw should preserve the snake_case'd subfields regardless of
    # input case.
    assert info.raw["event_name"] == "RunInstances"
    ui = info.raw["user_identity"]
    assert ui["principal_id"] == "12345:carol"
    assert ui["user_name"] == "carol@example.com"


@pytest.mark.asyncio
async def test_aliyun_get_resource_creator_reads_camelcase_events_in_paging_loop() -> None:
    """End-to-end of the field-case fix on the paging loop: the
    driver must accept camelCase event objects inside the
    PascalCase ``Events`` wrapper and select the earliest matching
    create event. Mirrors prod ActionTrail v3 response shape.
    """
    from app.integrations.drivers.alicloud_intl_cloud import (
        AlicloudIntlCloudDriver,
    )

    driver = AlicloudIntlCloudDriver.__new__(AlicloudIntlCloudDriver)
    driver.integration_id = "intg-test"
    driver.config = {
        "access_key_id": "ak", "access_key_secret": "sk",
    }

    async def fake_acs3_call(*, host, **_kwargs):
        if host == "actiontrail.ap-southeast-1.aliyuncs.com":
            # Real v3 wire shape: PascalCase wrapper + camelCase events.
            return {
                "Events": [
                    {
                        "eventName": "RunInstances",
                        "eventId": "ev-late",
                        "eventTime": "2026-05-10T10:00:00Z",
                        "eventSource": "ecs.aliyuncs.com",
                        "userIdentity": {
                            "type": "RAMUser",
                            "principalId": "uid-late",
                            "userName": "later@example.com",
                        },
                    },
                    {
                        "eventName": "RunInstances",
                        "eventId": "ev-early",
                        "eventTime": "2026-05-01T10:00:00Z",
                        "eventSource": "ecs.aliyuncs.com",
                        "userIdentity": {
                            "type": "RAMUser",
                            "principalId": "uid-early",
                            "userName": "creator@example.com",
                        },
                    },
                ],
                "NextToken": None,
            }
        # Subsequent regions also return empty so the picked event is
        # the earliest from ap-southeast-1.
        return {"Events": [], "NextToken": None}

    driver._acs3_call = fake_acs3_call  # type: ignore[assignment]

    info = await driver.get_resource_creator("i-camelcase")
    assert info is not None
    assert info.email == "creator@example.com"
    assert info.principal_id == "uid-early"


# ── Azure SP→owner Graph enrichment (PR #506) ──────────────────────


@pytest.mark.asyncio
async def test_azure_get_resource_creator_enriches_sp_owner_via_graph(
    monkeypatch,
) -> None:
    """Regression for PR #506: when the Activity Log create event's
    ``caller`` is a Service Principal GUID (Terraform / AKS managed
    identity / GitHub Actions), the static parse returns
    ``email=None``. The driver must call Microsoft Graph to resolve
    the SP's owner and stamp the row with the owner's email so the
    resource has a human attributable to it.
    """
    from app.integrations.drivers.azure_cloud import (
        _GRAPH_OID_CACHE,
        AzureCloudDriver,
    )

    _GRAPH_OID_CACHE.clear()

    sub_id = "11111111-2222-3333-4444-555555555555"
    sp_oid = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
    rid = (
        f"/subscriptions/{sub_id}/resourceGroups/rg/providers/"
        "Microsoft.Compute/disks/disk-1"
    )

    # Activity-Log response: ``caller`` is the SP's GUID; no UPN.
    activity_resp_body = {
        "value": [
            {
                "operationName": {"value": "Microsoft.Compute/disks/write"},
                "eventTimestamp": "2026-05-01T10:00:00Z",
                "caller": sp_oid,
                "correlationId": "abc",
                "resourceProviderName": {"value": "Microsoft.Compute"},
                "status": {"value": "Started"},
                "claims": {
                    "appid": "11111111-1111-1111-1111-111111111111",
                    "iss": "https://sts.windows.net/tenant/",
                },
            },
        ],
        "nextLink": None,
    }
    # Graph directoryObject → SP.
    dirobj_body = {
        "@odata.type": "#microsoft.graph.servicePrincipal",
        "id": sp_oid,
        "displayName": "tidb-cloud-terraform",
        "appId": "11111111-1111-1111-1111-111111111111",
    }
    # Graph SP owners → one user.
    owners_body = {
        "value": [
            {
                "@odata.type": "#microsoft.graph.user",
                "id": "owner-1",
                "mail": "alice@example.com",
                "displayName": "Alice",
            },
        ],
    }

    class _StubResp:
        def __init__(self, status, body):
            self.status_code = status
            self._body = body
            self.text = json.dumps(body)

        def json(self):
            return self._body

    class _StubClient:
        def __init__(self):
            self.urls: list[str] = []

        async def __aenter__(self):
            return self

        async def __aexit__(self, *args):
            return False

        async def get(self, url, params=None, headers=None):
            self.urls.append(url)
            if "providers/microsoft.insights" in url:
                return _StubResp(200, activity_resp_body)
            if f"directoryObjects/{sp_oid}" in url:
                return _StubResp(200, dirobj_body)
            if f"servicePrincipals/{sp_oid}/owners" in url:
                return _StubResp(200, owners_body)
            raise AssertionError(f"unexpected url {url}")

    stub_client = _StubClient()

    driver = AzureCloudDriver.__new__(AzureCloudDriver)
    driver.integration_id = "intg-azure-test"
    driver.config = {}
    driver._token_cache = {}

    async def _token(_resource):
        return "tok"

    monkeypatch.setattr(driver, "_token", _token)
    monkeypatch.setattr(driver, "_http", lambda: stub_client)

    info = await driver.get_resource_creator(rid)
    assert info is not None
    # Owner's email resolved via Graph.
    assert info.email == "alice@example.com"
    # display_name reflects both owner AND the SP that did the write.
    assert "Alice" in (info.display_name or "")
    assert "tidb-cloud-terraform" in (info.display_name or "")
    # principal_id stays the original SP oid for traceability.
    assert info.principal_id == sp_oid
    # raw retains the Graph enrichment metadata.
    assert info.raw["graph_sp_enrichment"]["sp_display_name"] == (
        "tidb-cloud-terraform"
    )
    _GRAPH_OID_CACHE.clear()


@pytest.mark.asyncio
async def test_azure_sp_enrichment_degrades_when_graph_returns_403(
    monkeypatch,
) -> None:
    """When Graph returns 403 (Directory.Read.All missing), the
    driver must NOT crash the cron — it returns the static-parsed
    info unchanged so the row at least lands as ``status='resolved'``
    with the SP GUID as principal_id (the pre-PR state). Operator
    sees the 403 in the worker logs and can grant the scope.
    """
    from app.integrations.drivers.azure_cloud import (
        _GRAPH_OID_CACHE,
        AzureCloudDriver,
    )

    _GRAPH_OID_CACHE.clear()

    sub_id = "11111111-2222-3333-4444-555555555555"
    sp_oid = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
    rid = (
        f"/subscriptions/{sub_id}/resourceGroups/rg/providers/"
        "Microsoft.Compute/disks/disk-1"
    )

    activity_resp_body = {
        "value": [
            {
                "operationName": {"value": "Microsoft.Compute/disks/write"},
                "eventTimestamp": "2026-05-01T10:00:00Z",
                "caller": sp_oid,
                "correlationId": "abc",
                "resourceProviderName": {"value": "Microsoft.Compute"},
                "status": {"value": "Started"},
                "claims": {"appid": "app-1"},
            },
        ],
        "nextLink": None,
    }

    class _StubResp:
        def __init__(self, status, body):
            self.status_code = status
            self._body = body
            self.text = json.dumps(body) if isinstance(body, dict) else body

        def json(self):
            return self._body if isinstance(self._body, dict) else {}

    class _StubClient:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *args):
            return False

        async def get(self, url, params=None, headers=None):
            if "providers/microsoft.insights" in url:
                return _StubResp(200, activity_resp_body)
            if f"directoryObjects/{sp_oid}" in url:
                return _StubResp(403, "forbidden")
            raise AssertionError(f"unexpected url {url}")

    driver = AzureCloudDriver.__new__(AzureCloudDriver)
    driver.integration_id = "intg-azure-test"
    driver.config = {}
    driver._token_cache = {}

    async def _token(_resource):
        return "tok"

    monkeypatch.setattr(driver, "_token", _token)
    monkeypatch.setattr(driver, "_http", lambda: _StubClient())

    info = await driver.get_resource_creator(rid)
    assert info is not None
    # Pre-PR behaviour preserved on 403 — no email, principal stays SP.
    assert info.email is None
    assert info.principal_id == sp_oid
    _GRAPH_OID_CACHE.clear()


@pytest.mark.asyncio
async def test_azure_sp_enrichment_skips_when_caller_is_already_email(
    monkeypatch,
) -> None:
    """When ``caller`` is already a UPN (e.g. ``alice@example.com``),
    the SP fallback must NOT fire — that's the pre-PR happy path
    for user callers (Azure Portal / CLI). One Graph call wasted
    per resource is a 200/tick QPS hit on a tenant that doesn't
    even need the lookup.
    """
    import json as _json

    from app.integrations.drivers.azure_cloud import (
        _GRAPH_OID_CACHE,
        AzureCloudDriver,
    )

    _GRAPH_OID_CACHE.clear()

    sub_id = "11111111-2222-3333-4444-555555555555"
    rid = (
        f"/subscriptions/{sub_id}/resourceGroups/rg/providers/"
        "Microsoft.Compute/disks/disk-1"
    )

    activity_resp_body = {
        "value": [
            {
                "operationName": {"value": "Microsoft.Compute/disks/write"},
                "eventTimestamp": "2026-05-01T10:00:00Z",
                "caller": "alice@example.com",
                "correlationId": "abc",
                "resourceProviderName": {"value": "Microsoft.Compute"},
                "status": {"value": "Started"},
                "claims": {"upn": "alice@example.com", "oid": "obj-1"},
            },
        ],
        "nextLink": None,
    }

    class _StubResp:
        status_code = 200

        def __init__(self, body):
            self._body = body
            self.text = _json.dumps(body)

        def json(self):
            return self._body

    class _StubClient:
        def __init__(self):
            self.graph_called = False

        async def __aenter__(self):
            return self

        async def __aexit__(self, *args):
            return False

        async def get(self, url, params=None, headers=None):
            from urllib.parse import urlsplit

            if urlsplit(url).hostname == "graph.microsoft.com":
                self.graph_called = True
                raise AssertionError(
                    "Graph must not be called when caller is already an email",
                )
            return _StubResp(activity_resp_body)

    stub = _StubClient()
    driver = AzureCloudDriver.__new__(AzureCloudDriver)
    driver.integration_id = "intg-azure-test"
    driver.config = {}
    driver._token_cache = {}

    async def _token(_resource):
        return "tok"

    monkeypatch.setattr(driver, "_token", _token)
    monkeypatch.setattr(driver, "_http", lambda: stub)

    info = await driver.get_resource_creator(rid)
    assert info is not None
    assert info.email == "alice@example.com"
    assert stub.graph_called is False


def test_aws_extract_lookup_name_filters_synthetic_ids() -> None:
    """Synthetic resource ids minted by the BQ-CUR ingest path
    (``aws-account:`` aggregates, ``aws-principal:`` IAM-tag rows)
    must short-circuit so we don't burn a CloudTrail call on
    them."""
    from app.integrations.drivers.aws_cloud import AwsCloudDriver

    assert AwsCloudDriver._cloudtrail_extract_lookup_name(
        "aws-account:178:Compute",
    ) is None
    assert AwsCloudDriver._cloudtrail_extract_lookup_name(
        "aws-principal:Owner=alice|service=EC2",
    ) is None
    # Real ARN still works.
    assert AwsCloudDriver._cloudtrail_extract_lookup_name(
        "arn:aws:s3:::my-bucket",
    ) == ("my-bucket", True)