"""Per-scope billing-detail endpoint tests.

Covers:

  * **Shape contract**: ``GET /cmdb/scopes/{id}/billing-detail`` returns
    scope identity + month summary + daily trend + monthly history +
    cost_by_service + top_resources + members. Front-end ``Scope
    BillingDetail`` interface tracks these — break the shape, break
    the drawer.
  * **AuthZ**: scope member / owner OR cloud_admin / it_admin /
    finance can read; everyone else gets 403. Closed-parent
    inheritance for RG scopes does NOT grant residual read on a
    surviving child (same tombstone safeguard ``/billing/daily``
    applies).
  * **Month parsing**: strict ``YYYY-MM`` (single-digit month rejected),
    defaults to current month when omitted, future month rejected by
    Query pattern.
  * **Past-vs-current month**: ``is_partial=True`` for current month,
    ``False`` for past months; daily trend zero-fills the whole
    range; monthly history walks back 6 months from the anchor.
  * **Tenant-vs-RG routing**: tenant scopes filter by ``scope_id``;
    RG-typed scopes filter by ``(integration_id, resource_group)``.
    Routing echo in response carries the chosen branch.
  * **Member redaction**: plain employee viewing a shared scope sees
    member display_names but NOT their cloud roles (same disclosure
    gate as ``/my-subscriptions``).

Hits the real test MySQL via the standard ``client`` fixture so SQL
joins exercise the actual indexes (the cost_facts month-window scan
relies on ``ix_facts_date_int``).
"""
from __future__ import annotations

import uuid
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.models import (
    GRANT_STATUS_ACTIVE,
    SCOPE_STATUS_ACTIVE,
    SCOPE_TYPE_AZURE_RESOURCE_GROUP,
    CmdbCloudAccessGrant,
    CmdbCloudScope,
    CmdbCostFact,
)
from app.integrations.models import Integration


def _now_naive() -> datetime:
    return datetime.now(UTC).replace(tzinfo=None)


async def _mk_integ(db: AsyncSession, provider: str = "aws") -> Integration:
    row = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider=provider,
        name=f"{provider}-{uuid.uuid4().hex[:4]}",
        slug=f"{provider}-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db.add(row)
    await db.commit()
    return row


async def _mk_scope(
    db: AsyncSession,
    integ: Integration,
    *,
    owner_person_id: str | None = None,
    scope_type: str | None = None,
    external_id: str | None = None,
) -> CmdbCloudScope:
    s = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integ.id),
        scope_type=scope_type or f"{integ.provider}_account",
        external_id=external_id or uuid.uuid4().hex[:16],
        display_name=f"scope-{uuid.uuid4().hex[:6]}",
        status=SCOPE_STATUS_ACTIVE,
        environment="dev",
        owner_person_id=owner_person_id,
        budget_usd_monthly=Decimal("1000"),
        first_seen_at=_now_naive(),
        last_synced_at=_now_naive(),
    )
    db.add(s)
    await db.commit()
    return s


async def _mk_fact(
    db: AsyncSession,
    *,
    integ_id: str,
    scope_id: str | None,
    fact_date: date,
    service: str = "ECS",
    amount: str = "10.00",
    resource_id: str | None = None,
    resource_group: str | None = None,
) -> CmdbCostFact:
    """Mint one cost-fact row with a deterministic fact_key."""
    import hashlib
    rid = resource_id or f"res-{uuid.uuid4().hex[:8]}"
    raw = f"test:{integ_id}:{scope_id}:{fact_date}:{rid}:{service}"
    fact_key = hashlib.sha256(raw.encode()).hexdigest()
    f = CmdbCostFact(
        fact_key=fact_key,
        date=fact_date,
        integration_id=integ_id,
        scope_id=scope_id,
        resource_id=rid,
        resource_name=f"name-{rid}",
        resource_type="instance",
        service=service,
        region="us-east-1",
        resource_group=resource_group,
        amount_usd=Decimal(amount),
        amount_native=Decimal(amount),
        currency="USD",
    )
    db.add(f)
    await db.commit()
    return f


# ── AuthZ ──────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_billing_detail_admin_can_read_any_scope(
    client, admin_user, db_session,
):
    _, token = admin_user
    integ = await _mk_integ(db_session)
    scope = await _mk_scope(db_session, integ)
    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/billing-detail",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text


@pytest.mark.asyncio
async def test_billing_detail_employee_no_membership_403(
    client, employee_user, db_session,
):
    _, token = employee_user
    integ = await _mk_integ(db_session)
    scope = await _mk_scope(db_session, integ)
    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/billing-detail",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_billing_detail_employee_with_grant_can_read(
    client, employee_user, db_session,
):
    user, token = employee_user
    integ = await _mk_integ(db_session)
    scope = await _mk_scope(db_session, integ)
    grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(user.id),
        role_key="Reader",
        status=GRANT_STATUS_ACTIVE,
        granted_at=_now_naive(),
        source="helix",
    )
    db_session.add(grant)
    await db_session.commit()
    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/billing-detail",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text


@pytest.mark.asyncio
async def test_billing_detail_unknown_scope_404(client, admin_user):
    _, token = admin_user
    fake = uuid.uuid4()
    resp = await client.get(
        f"/api/v1/cmdb/scopes/{fake}/billing-detail",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 404


# ── Month parsing ──────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_billing_detail_invalid_month_422(
    client, admin_user, db_session,
):
    _, token = admin_user
    integ = await _mk_integ(db_session)
    scope = await _mk_scope(db_session, integ)
    # Single-digit month — strict YYYY-MM pattern rejects.
    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/billing-detail",
        params={"month": "2026-5"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_billing_detail_future_month_rejected(
    client, admin_user, db_session,
):
    """Future months produce no useful data — gate them out as 400."""
    _, token = admin_user
    integ = await _mk_integ(db_session)
    scope = await _mk_scope(db_session, integ)
    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/billing-detail",
        params={"month": "2099-12"},
        headers={"Authorization": f"Bearer {token}"},
    )
    # ValidationError from parse_month_param maps to 422 (the
    # platform's standard validation status — see ValidationError in
    # app/core/exceptions.py).
    assert resp.status_code == 422, resp.text
    assert "future" in resp.text.lower()


# ── Shape + Aggregation ────────────────────────────────────────────


@pytest.mark.asyncio
async def test_billing_detail_shape_and_total(
    client, admin_user, db_session,
):
    _, token = admin_user
    integ = await _mk_integ(db_session)
    scope = await _mk_scope(db_session, integ)

    today = datetime.now(UTC).date()
    month_start = today.replace(day=1)
    await _mk_fact(
        db_session,
        integ_id=str(integ.id), scope_id=str(scope.id),
        fact_date=month_start, service="ECS", amount="10",
    )
    await _mk_fact(
        db_session,
        integ_id=str(integ.id), scope_id=str(scope.id),
        fact_date=month_start, service="OSS", amount="3",
    )

    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/billing-detail",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]

    for k in (
        "scope", "month", "month_window", "summary", "daily_trend",
        "monthly_history", "cost_by_service", "top_resources",
        "members", "routing",
    ):
        assert k in data, f"missing top-level key: {k}"

    assert data["routing"]["routing"] == "scope_id"
    # 10 + 3 = 13, settled today.
    assert pytest.approx(data["summary"]["total_usd"]) == 13.0
    # Two services in the breakdown.
    services = {s["service"] for s in data["cost_by_service"]}
    assert "ECS" in services and "OSS" in services
    # Current month — partial flag set, projection > total for a
    # 1-day-MTD that doesn't yet span the full month.
    assert data["month_window"]["is_partial"] is True
    assert data["summary"]["data_available"] is True


@pytest.mark.asyncio
async def test_billing_detail_rg_routing(
    client, admin_user, db_session,
):
    """RG-typed scopes route through (integration_id, resource_group)."""
    _, token = admin_user
    integ = await _mk_integ(db_session, provider="azure")
    rg_name = f"rg-{uuid.uuid4().hex[:6]}"
    scope = await _mk_scope(
        db_session, integ,
        scope_type=SCOPE_TYPE_AZURE_RESOURCE_GROUP,
        external_id=rg_name,
    )
    today = datetime.now(UTC).date().replace(day=1)
    await _mk_fact(
        db_session,
        integ_id=str(integ.id),
        scope_id=None,  # facts for RG scopes do NOT carry scope_id
        fact_date=today, service="Compute", amount="7.50",
        resource_group=rg_name,
    )
    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/billing-detail",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["routing"]["routing"] == "rg"
    assert data["routing"]["rg_name"] == rg_name
    assert pytest.approx(data["summary"]["total_usd"]) == 7.5


@pytest.mark.asyncio
async def test_billing_detail_past_month_settled(
    client, admin_user, db_session,
):
    """Past month: is_partial=False, projection==total."""
    _, token = admin_user
    integ = await _mk_integ(db_session)
    scope = await _mk_scope(db_session, integ)

    today = datetime.now(UTC).date()
    # Previous month's first day
    if today.month == 1:
        prev_y, prev_m = today.year - 1, 12
    else:
        prev_y, prev_m = today.year, today.month - 1
    past_month_date = date(prev_y, prev_m, 5)
    await _mk_fact(
        db_session,
        integ_id=str(integ.id), scope_id=str(scope.id),
        fact_date=past_month_date, service="ECS", amount="42",
    )
    past_month_str = f"{prev_y:04d}-{prev_m:02d}"
    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/billing-detail",
        params={"month": past_month_str},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["month"] == past_month_str
    assert data["month_window"]["is_partial"] is False
    assert pytest.approx(data["summary"]["total_usd"]) == 42.0
    # Past month: projection equals the settled total.
    assert pytest.approx(
        data["summary"]["month_end_projection_usd"],
    ) == 42.0


# ── Trailing zero-fill regression (2026-05-18) ─────────────────────


@pytest.mark.asyncio
async def test_billing_detail_truncates_trailing_missing_days(
    client, admin_user, db_session,
):
    """When the cost-facts cron lags behind today, the daily-trend
    series must NOT zero-fill the missing trailing days (a misleading
    "$0 spend" rendering) and the daily-average / projection must NOT
    divide by days-to-today (which understates both).

    Anchor on ``last_fact_date`` instead.

    Regression for 2026-05-18: ``aws-byoc-deployment`` showed a $0
    spike-down on 5/17 and 5/18 because cost_facts only had data up
    to 5/16; the daily-average then divided $710 by 18 days (to today)
    rather than 16 days (to last_fact_date), understating the daily
    by ~13% and the month-end projection by the same factor.
    """
    _, token = admin_user
    integ = await _mk_integ(db_session)
    scope = await _mk_scope(db_session, integ)

    today = datetime.now(UTC).date()
    month_start = today.replace(day=1)
    # Need at least 3 days into the month for the gap to be meaningful;
    # skip the test silently on day-1/day-2 of the month rather than
    # racing against the clock.
    if (today - month_start).days < 3:
        pytest.skip(
            "needs >=3 days into the current month to exercise lag",
        )
    # Insert facts for month_start and (today - 2), leaving (today - 1)
    # and today as "not yet ingested" trailing days. last_fact_date
    # should be (today - 2).
    last_with_data = today - timedelta(days=2)
    await _mk_fact(
        db_session,
        integ_id=str(integ.id), scope_id=str(scope.id),
        fact_date=month_start, service="ECS", amount="10",
    )
    await _mk_fact(
        db_session,
        integ_id=str(integ.id), scope_id=str(scope.id),
        fact_date=last_with_data, service="ECS", amount="20",
    )

    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/billing-detail",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    summary = data["summary"]
    trend = data["daily_trend"]

    # last_fact_date matches the latest fact we inserted.
    assert summary["last_fact_date"] == last_with_data.isoformat()

    # Daily series ends at last_fact_date — no $0 trailing bars for
    # the unsynced days.
    assert len(trend) > 0
    assert trend[-1]["date"] == last_with_data.isoformat()
    # No row in the series falls after last_fact_date.
    for d in trend:
        assert date.fromisoformat(d["date"]) <= last_with_data

    # days_elapsed is anchored on last_fact_date, not today.
    expected_days_elapsed = (last_with_data - month_start).days + 1
    assert summary["days_elapsed"] == expected_days_elapsed

    # daily_average = total / expected_days_elapsed (not / days-to-today).
    total = 10.0 + 20.0
    assert pytest.approx(
        summary["daily_average_usd"], rel=1e-3,
    ) == total / expected_days_elapsed


@pytest.mark.asyncio
async def test_billing_detail_past_month_uses_full_calendar_for_daily_avg(
    client, admin_user, db_session,
):
    """Past months: daily_average must use the FULL calendar month as
    denominator (conventional finance figure), not last_fact_date.

    Regression guard for Round-32 follow-up: the current-month fix
    anchors days_elapsed on last_fact_date to avoid diluting the daily
    average with un-ingested trailing days. That logic must NOT
    extend to past months — a partially-backfilled past month (e.g.
    one $20 fact on 3/5) should still render daily_avg = $20/31, not
    $20/5. The latter would massively inflate ranking views built on
    daily_average and mislead finance review of historical months.
    """
    _, token = admin_user
    integ = await _mk_integ(db_session)
    scope = await _mk_scope(db_session, integ)

    today = datetime.now(UTC).date()
    if today.month == 1:
        prev_y, prev_m = today.year - 1, 12
    else:
        prev_y, prev_m = today.year, today.month - 1
    import calendar as _cal
    days_in_prev = _cal.monthrange(prev_y, prev_m)[1]

    # Single fact on day 5 of the past month — leaves a 25+ day
    # trailing gap (well past the cron's freshness SLA).
    await _mk_fact(
        db_session,
        integ_id=str(integ.id), scope_id=str(scope.id),
        fact_date=date(prev_y, prev_m, 5),
        service="ECS", amount="20",
    )

    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/billing-detail",
        params={"month": f"{prev_y:04d}-{prev_m:02d}"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    summary = data["summary"]
    # is_partial=False for the past month.
    assert data["month_window"]["is_partial"] is False
    # days_elapsed anchors on the FULL calendar month (not on day 5).
    assert summary["days_elapsed"] == days_in_prev
    # daily_average = 20 / days_in_month, NOT 20 / 5.
    assert pytest.approx(
        summary["daily_average_usd"], rel=1e-3,
    ) == 20.0 / days_in_prev
    # Past-month projection equals the settled total (unchanged
    # behaviour — projection-vs-extrapolation only applies to
    # is_partial=True months).
    assert pytest.approx(summary["month_end_projection_usd"]) == 20.0


# ── /my-subscriptions month param ──────────────────────────────────


@pytest.mark.asyncio
async def test_my_subscriptions_month_param(
    client, employee_user, db_session,
):
    """cost_month_usd reflects the requested month, sourced from cost_facts."""
    user, token = employee_user
    integ = await _mk_integ(db_session)
    scope = await _mk_scope(db_session, integ)
    grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=str(user.id),
        role_key="Reader",
        status=GRANT_STATUS_ACTIVE,
        granted_at=_now_naive(),
        source="helix",
    )
    db_session.add(grant)
    await db_session.commit()

    today = datetime.now(UTC).date()
    month_start = today.replace(day=1)
    await _mk_fact(
        db_session,
        integ_id=str(integ.id), scope_id=str(scope.id),
        fact_date=month_start, service="ECS", amount="20",
    )
    resp = await client.get(
        "/api/v1/cmdb/my-subscriptions",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    rows = body["data"]
    assert len(rows) == 1
    assert rows[0]["scope_id"] == str(scope.id)
    assert pytest.approx(rows[0]["cost_month_usd"]) == 20.0
    # meta echoes the resolved month + partial flag.
    assert "month" in body["meta"]
    assert body["meta"]["month_is_partial"] is True


@pytest.mark.asyncio
async def test_my_subscriptions_include_requestable_directory(
    client, employee_user, db_session,
):
    """``include_requestable=true`` appends owner-only directory rows.

    The 我的云订阅 page lets a plain employee see EVERY requestable
    sub-account, not just the ones they're a member of. Rows the caller
    has a grant on come back ``accessible=true`` (full billing/members);
    the rest come back ``accessible=false`` with the owner exposed but
    cost zeroed and no member roster — the frontend renders those as
    "owner only + 快速申请".
    """
    from app.identity.person_models import IdentityPerson

    user, token = employee_user
    integ = await _mk_integ(db_session)

    # Scope A — the caller holds a grant → accessible row.
    scope_mine = await _mk_scope(db_session, integ)
    db_session.add(CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope_mine.id),
        person_id=str(user.id),
        role_key="Reader",
        status=GRANT_STATUS_ACTIVE,
        granted_at=_now_naive(),
        source="helix",
    ))

    # Scope B — owned by someone else, active + dev, caller has NO grant
    # → directory row (accessible=false) when include_requestable=true.
    owner = IdentityPerson(
        id=uuid.uuid4(),
        employee_no=f"E{uuid.uuid4().hex[:8]}",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        source_system="successfactors",
        display_name="Scope Owner",
        work_email="scope.owner@pingcap.com",
        employment_status="active",
    )
    db_session.add(owner)
    await db_session.commit()
    scope_other = await _mk_scope(
        db_session, integ, owner_person_id=str(owner.id),
    )
    await db_session.commit()

    # Default (no flag): only the caller's own subscription.
    resp = await client.get(
        "/api/v1/cmdb/my-subscriptions",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    rows = resp.json()["data"]
    assert {r["scope_id"] for r in rows} == {str(scope_mine.id)}
    assert rows[0]["accessible"] is True

    # With the flag: both scopes, B flagged non-accessible + owner shown.
    resp = await client.get(
        "/api/v1/cmdb/my-subscriptions?include_requestable=true",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    by_id = {r["scope_id"]: r for r in resp.json()["data"]}
    assert set(by_id) == {str(scope_mine.id), str(scope_other.id)}
    assert by_id[str(scope_mine.id)]["accessible"] is True
    directory = by_id[str(scope_other.id)]
    assert directory["accessible"] is False
    assert directory["owner_person_id"] == str(owner.id)
    assert "scope.owner@pingcap.com" in (directory["owner_label"] or "")
    assert directory["cost_month_usd"] == 0.0
    assert directory["members"] == []
    assert directory["my_grants"] == []


@pytest.mark.asyncio
async def test_my_subscriptions_directory_includes_prod_as_not_requestable(
    client, employee_user, db_session,
):
    """Prod scopes are discoverable but never self-service requestable.

    This lets an employee find and recognize a production subscription while
    preserving the production-access boundary in the picker and submit path.
    Directory rows still withhold billing, members, and detail access.
    """
    from app.identity.person_models import IdentityPerson

    _user, token = employee_user
    integ = await _mk_integ(db_session)
    owner = IdentityPerson(
        id=uuid.uuid4(),
        employee_no=f"E{uuid.uuid4().hex[:8]}",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        source_system="successfactors",
        display_name="Prod Owner",
        work_email="prod.owner@pingcap.com",
        employment_status="active",
    )
    db_session.add(owner)
    await db_session.commit()
    prod_scope = await _mk_scope(
        db_session, integ, owner_person_id=str(owner.id),
    )
    prod_scope.environment = "prod"
    await db_session.commit()

    resp = await client.get(
        "/api/v1/cmdb/my-subscriptions?include_requestable=true",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    by_id = {r["scope_id"]: r for r in resp.json()["data"]}
    directory = by_id[str(prod_scope.id)]
    assert directory["accessible"] is False
    assert directory["requestable"] is False
    assert directory["environment"] == "prod"
    assert directory["cost_month_usd"] == 0.0
    assert directory["members"] == []

    # The self-service picker remains a strictly smaller set: discovery
    # never becomes authority to request production access.
    picker = await client.get(
        "/api/v1/cmdb/scopes?requestable=true&page_size=500",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert picker.status_code == 200, picker.text
    picker_ids = {r["id"] for r in picker.json()["data"]}
    assert str(prod_scope.id) not in picker_ids


@pytest.mark.asyncio
async def test_billing_detail_top_resources_carry_creator_attribution(
    client, admin_user, db_session,
):
    """`top_resources` rows include `creator` from cmdb_resource_creators.

    The 4th SQL pass in ``_top_resources`` bulk-fetches creator rows
    keyed on (integration_id, resource_id) and joins identity_persons
    for canonical display_name. Locks the contract that the FE
    ``ScopeBillingResourceRow.creator`` field uses to render the
    「创建者」 column on both /admin and /my scope-detail pages.

    Three rows exercise the three render paths:
      * resolved + matched Person → display_name from Person
      * resolved + no Person match → display_name from creator_display_name
      * cron sweep hasn't processed → creator=None
    """
    from app.domains.cmdb.models import CmdbResourceCreator
    from app.identity.person_models import IdentityPerson

    _, token = admin_user
    integ = await _mk_integ(db_session, provider="azure")
    scope = await _mk_scope(db_session, integ)

    # Helix Person that will match by work_email — canonical
    # display_name should win over the audit-log-derived name.
    person = IdentityPerson(
        id=uuid.uuid4(),
        employee_no=f"E{uuid.uuid4().hex[:8]}",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        source_system="successfactors",
        display_name="Canonical Person",
        work_email="weiwei.cheng@pingcap.com",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.commit()

    today = datetime.now(UTC).date()
    month_start = today.replace(day=1)

    # Resource A — resolved, matched person.
    rid_a = "/subscriptions/abc/resourcegroups/auto-deploy/foo"
    await _mk_fact(
        db_session, integ_id=str(integ.id), scope_id=str(scope.id),
        fact_date=month_start, service="VM", amount="50",
        resource_id=rid_a,
    )
    db_session.add(CmdbResourceCreator(
        integration_id=str(integ.id), resource_id=rid_a,
        creator_email="weiwei.cheng@pingcap.com",
        creator_display_name="Weiwei Cheng",  # audit-log name
        creator_person_id=str(person.id),
        source="azure_activity", status="resolved",
        attempt_count=1,
    ))

    # Resource B — resolved but no Helix Person match (external).
    rid_b = "/subscriptions/abc/resourcegroups/external/bar"
    await _mk_fact(
        db_session, integ_id=str(integ.id), scope_id=str(scope.id),
        fact_date=month_start, service="VM", amount="30",
        resource_id=rid_b,
    )
    db_session.add(CmdbResourceCreator(
        integration_id=str(integ.id), resource_id=rid_b,
        creator_email="contractor@vendor.com",
        creator_display_name="External Contractor",
        creator_person_id=None,
        source="azure_activity", status="resolved",
        attempt_count=1,
    ))

    # Resource C — no cmdb_resource_creators row at all (cron sweep
    # hasn't processed). FE should render creator=null distinctly
    # from status='not_found'.
    rid_c = "/subscriptions/abc/resourcegroups/coldrow/baz"
    await _mk_fact(
        db_session, integ_id=str(integ.id), scope_id=str(scope.id),
        fact_date=month_start, service="VM", amount="20",
        resource_id=rid_c,
    )
    await db_session.commit()

    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/billing-detail",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text
    top = resp.json()["data"]["top_resources"]
    by_rid = {r["resource_id"]: r for r in top}

    assert rid_a in by_rid
    assert by_rid[rid_a]["integration_id"] == str(integ.id)
    creator_a = by_rid[rid_a]["creator"]
    assert creator_a is not None
    assert creator_a["status"] == "resolved"
    assert creator_a["email"] == "weiwei.cheng@pingcap.com"
    # Canonical Person.display_name wins over audit-log creator_display_name.
    assert creator_a["display_name"] == "Canonical Person"
    assert creator_a["person_id"] == str(person.id)
    assert creator_a["source"] == "azure_activity"

    creator_b = by_rid[rid_b]["creator"]
    assert creator_b is not None
    assert creator_b["status"] == "resolved"
    assert creator_b["person_id"] is None
    # No Person match → fall back to audit-log display_name.
    assert creator_b["display_name"] == "External Contractor"

    # No creator row → null (distinct from status='not_found').
    assert by_rid[rid_c]["creator"] is None


# ── _month_bounds: current-month partial flag (last-day edge) ──────


class _FixedDateTime(datetime):
    """datetime subclass whose ``now(tz)`` returns a frozen instant, so the
    date-sensitive ``_month_bounds`` branch can be tested deterministically
    on any calendar day (notably the last day of a month)."""

    _frozen: datetime

    @classmethod
    def now(cls, tz=None):  # noqa: D102 - mirrors datetime.now signature
        return cls._frozen if tz is None else cls._frozen.astimezone(tz)


def _freeze_today(monkeypatch, frozen_date: date) -> None:
    import app.domains.cmdb.scope_billing_detail as mod

    _FixedDateTime._frozen = datetime(
        frozen_date.year, frozen_date.month, frozen_date.day, 12, 0, tzinfo=UTC,
    )
    monkeypatch.setattr(mod, "datetime", _FixedDateTime)


@pytest.mark.parametrize(
    "today,year,month_num,exp_end,exp_partial",
    [
        # Last calendar day of the current month → STILL partial (regression:
        # full_month_end == today previously mis-flagged it as settled).
        (date(2026, 6, 30), 2026, 6, date(2026, 6, 30), True),
        # Mid current month → partial, clamped to today.
        (date(2026, 6, 15), 2026, 6, date(2026, 6, 15), True),
        # First day of the current month → partial.
        (date(2026, 6, 1), 2026, 6, date(2026, 6, 1), True),
        # A past month → fully settled (full month end, not partial).
        (date(2026, 6, 30), 2026, 5, date(2026, 5, 31), False),
    ],
)
def test_month_bounds_current_month_always_partial(
    monkeypatch, today, year, month_num, exp_end, exp_partial,
):
    from app.domains.cmdb.scope_billing_detail import _month_bounds

    _freeze_today(monkeypatch, today)
    month_start, month_end, is_partial = _month_bounds(year, month_num)
    assert month_start == date(year, month_num, 1)
    assert month_end == exp_end
    assert is_partial is exp_partial