"""Tests for cloud-scope owner governance — semiannual confirm + zero-spend.

Covers:

* :func:`OwnerConfirmationService.list_scopes_due_semiannual` — only
  selects non-prod active scopes whose owners are past 180 days and
  have no outstanding pending row.
* :func:`OwnerConfirmationService.list_scopes_with_zero_spend_last_month`
  — selects scopes with $0 spend in the named billing month,
  excluding brand-new scopes and scopes with an outstanding pending
  row from the same month.
* Confirm endpoint flips pending → responded and bumps
  ``last_owner_confirmed_at``.
* Zero-spend respond endpoint requires ``reason`` on the ``keep`` action.
* :func:`OwnerConfirmationService.remind_unanswered_owners` re-sends an
  unanswered prompt to its OWNER (IT escalation retired 2026-08-24 —
  those cases now live in
  ``test_owner_governance_timeline_2026_08_24.py``).
* ``assign_scope_owner`` now dispatches a notification to the new owner
  (verified via the ``notification_outbox`` row).
"""
from __future__ import annotations

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

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

from app.domains.cmdb.models import (
    OWNER_CONFIRMATION_KIND_OWNER_TRANSFERRED,
    OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING,
    OWNER_CONFIRMATION_KIND_SEMIANNUAL_TRANSFERRED,
    OWNER_CONFIRMATION_KIND_ZERO_SPEND_PENDING,
    OWNER_CONFIRMATION_STATUS_PENDING,
    OWNER_CONFIRMATION_STATUS_RESPONDED,
    SCOPE_STATUS_ACTIVE,
    CmdbCloudScope,
    CmdbCloudScopeOwnerConfirmation,
    CmdbCostFact,
)
from app.domains.cmdb.owner_confirmation_service import (
    OWNER_CONFIRMATION_FIRST_OBSERVED_KEY,
    OwnerConfirmationService,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration, PlatformSetting
from app.notifications.models import NotificationOutbox


async def _make_person(
    db: AsyncSession,
    *,
    email: str,
    employment_status: str = "active",
    employee_no: str | None = None,
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=email.split("@")[0],
        employee_no=employee_no,
        employment_status=employment_status,
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db.add(p)
    await db.flush()
    return p


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


async def _make_scope(
    db: AsyncSession,
    integration: Integration,
    *,
    owner_id: uuid.UUID | None,
    environment: str = "dev",
    last_owner_confirmed_at: datetime | None = None,
    created_days_ago: int = 365,
    created_at: datetime | None = None,
) -> CmdbCloudScope:
    # ``created_at`` (absolute) overrides ``created_days_ago`` (relative to
    # now). Tests that pin an explicit target month MUST use the absolute
    # form — a relative ``created_days_ago`` silently rots as the calendar
    # advances (e.g. 400d-ago crept forward to == a fixed 2025-06 month_start,
    # flipping the ``created_at < month_start`` filter and emptying the result).
    _created = created_at if created_at is not None else (
        datetime.now(UTC) - timedelta(days=created_days_ago)
    )
    s = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integration.id),
        scope_type="aws_account",
        external_id=uuid.uuid4().hex[:16],
        display_name=f"scope-{uuid.uuid4().hex[:6]}",
        status=SCOPE_STATUS_ACTIVE,
        owner_person_id=str(owner_id) if owner_id else None,
        environment=environment,
        last_owner_confirmed_at=last_owner_confirmed_at,
        first_seen_at=_created,
        last_synced_at=datetime.now(UTC),
        created_at=_created,
    )
    db.add(s)
    await db.flush()
    return s


# ── list_scopes_due_semiannual ───────────────────────────────────


@pytest.mark.asyncio
async def test_semiannual_selects_only_overdue_non_prod_with_owner(
    db_session: AsyncSession,
):
    integ = await _make_integration(db_session)
    owner = await _make_person(db_session, email="owner@test.com")
    long_ago = datetime.now(UTC) - timedelta(days=200)
    fresh = datetime.now(UTC) - timedelta(days=30)

    # ✓ eligible: non-prod, owner set, confirmed 200 days ago
    overdue = await _make_scope(
        db_session, integ, owner_id=owner.id,
        environment="dev", last_owner_confirmed_at=long_ago,
    )
    # ✗ never confirmed but very new (< 30d) — skipped
    await _make_scope(
        db_session, integ, owner_id=owner.id,
        environment="dev", last_owner_confirmed_at=None,
        created_days_ago=5,
    )
    # ✗ prod — exempt
    await _make_scope(
        db_session, integ, owner_id=owner.id,
        environment="prod", last_owner_confirmed_at=long_ago,
    )
    # ✗ no owner — exempt
    await _make_scope(
        db_session, integ, owner_id=None,
        environment="dev", last_owner_confirmed_at=long_ago,
    )
    # ✗ recently confirmed (30d ago) — exempt
    await _make_scope(
        db_session, integ, owner_id=owner.id,
        environment="staging", last_owner_confirmed_at=fresh,
    )
    await db_session.commit()

    svc = OwnerConfirmationService(db_session)
    eligible = await svc.list_scopes_due_semiannual()
    assert [s.id for s in eligible] == [overdue.id]


@pytest.mark.asyncio
async def test_semiannual_min_age_days_zero_picks_up_brand_new_scopes(
    db_session: AsyncSession,
):
    """Init-sweep mode: a tenant whose scopes were bulk-imported in
    the last 30 days needs ``min_age_days=0`` to receive cards on the
    first sweep. Default 30-day newborn guard otherwise hides them."""
    integ = await _make_integration(db_session)
    owner = await _make_person(db_session, email="initsweep@test.com")
    # ✓ 5-day-old never-confirmed scope — hidden by default guard,
    # surfaced when admin passes min_age_days=0.
    young = await _make_scope(
        db_session, integ, owner_id=owner.id,
        environment="dev", last_owner_confirmed_at=None,
        created_days_ago=5,
    )
    await db_session.commit()

    svc = OwnerConfirmationService(db_session)
    # Default guard hides it.
    assert await svc.list_scopes_due_semiannual() == []
    # Init-sweep override picks it up.
    eligible = await svc.list_scopes_due_semiannual(min_age_days=0)
    assert [s.id for s in eligible] == [young.id]


@pytest.mark.asyncio
async def test_semiannual_min_age_days_negative_rejected(
    db_session: AsyncSession,
):
    svc = OwnerConfirmationService(db_session)
    with pytest.raises(ValueError, match="min_age_days"):
        await svc.list_scopes_due_semiannual(min_age_days=-1)


@pytest.mark.asyncio
async def test_semiannual_skips_scopes_with_outstanding_pending(
    db_session: AsyncSession,
):
    """Re-running the cron the same day must not produce a duplicate
    prompt — the existing pending row blocks re-selection."""
    integ = await _make_integration(db_session)
    owner = await _make_person(db_session, email="owner2@test.com")
    scope = await _make_scope(
        db_session, integ, owner_id=owner.id,
        environment="dev",
        last_owner_confirmed_at=datetime.now(UTC) - timedelta(days=300),
    )
    db_session.add(CmdbCloudScopeOwnerConfirmation(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        owner_person_id=str(owner.id),
        kind=OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING,
        status=OWNER_CONFIRMATION_STATUS_PENDING,
        notified_at=datetime.now(UTC),
    ))
    await db_session.commit()

    svc = OwnerConfirmationService(db_session)
    eligible = await svc.list_scopes_due_semiannual()
    assert eligible == []


# ── list_scopes_with_zero_spend_last_month ───────────────────────


@pytest.mark.asyncio
async def test_zero_spend_selects_only_zero_spend_with_owner(
    db_session: AsyncSession,
):
    integ = await _make_integration(db_session)
    owner = await _make_person(db_session, email="ownerz@test.com")

    # Pick a month deep enough in the past that no other test fixtures
    # interfere. Last month relative to "today" is the canonical case
    # but the function takes an explicit year/month so we can pin.
    target_year = 2025
    target_month = 6  # June 2025
    month_start = date(target_year, target_month, 1)

    # Round-51 L3-P0 first-observation cooldown: the service stamps a
    # platform-wide ``first_observed_at`` on its first run and returns []
    # for 30 days. Seed it well in the past so this unit exercises the
    # eligibility query, not the cold-start guard.
    db_session.add(PlatformSetting(
        key=OWNER_CONFIRMATION_FIRST_OBSERVED_KEY,
        value={"at": (datetime.now(UTC) - timedelta(days=60)).isoformat()},
    ))

    # Round-51 L3-P0 ingest-presence guard: a scope is only eligible if
    # cost data is demonstrably flowing for it (≥1 fact row in the 60d
    # before the billing month). A presence fact dated in the prior
    # month sits OUTSIDE the [month_start, next_month_start) spend window,
    # so it proves ingest without contributing to the zero-spend total.
    presence_date = month_start - timedelta(days=20)  # 2025-05-12
    # Absolute "created before month start" stamp — pinned, not now-relative,
    # so the ``created_at < month_start`` filter can't rot as time advances.
    before_month = datetime(target_year, 1, 1, tzinfo=UTC)  # 2025-01-01

    # ✓ zero-spend scope, created before month start, with ingest presence
    zero_scope = await _make_scope(
        db_session, integ, owner_id=owner.id,
        environment="dev", created_at=before_month,
    )
    db_session.add(CmdbCostFact(
        fact_key=f"fk-{uuid.uuid4().hex}",
        date=presence_date,
        integration_id=str(integ.id),
        scope_id=str(zero_scope.id),
        service="ec2",
        amount_usd=Decimal("4.00"),
        amount_native=Decimal("4.00"),
        currency="USD",
        fetched_at=datetime.now(UTC),
    ))
    # ✓ but with non-zero spend in the billing month — exclude. Also give
    # it ingest presence so its exclusion is genuinely attributable to
    # spend, not to the ingest-presence pre-filter.
    spent_scope = await _make_scope(
        db_session, integ, owner_id=owner.id,
        environment="dev", created_at=before_month,
    )
    db_session.add(CmdbCostFact(
        fact_key=f"fk-{uuid.uuid4().hex}",
        date=presence_date,
        integration_id=str(integ.id),
        scope_id=str(spent_scope.id),
        service="ec2",
        amount_usd=Decimal("1.00"),
        amount_native=Decimal("1.00"),
        currency="USD",
        fetched_at=datetime.now(UTC),
    ))
    db_session.add(CmdbCostFact(
        fact_key=f"fk-{uuid.uuid4().hex}",
        date=date(target_year, target_month, 5),
        integration_id=str(integ.id),
        scope_id=str(spent_scope.id),
        service="ec2",
        amount_usd=Decimal("12.50"),
        amount_native=Decimal("12.50"),
        currency="USD",
        fetched_at=datetime.now(UTC),
    ))
    # ✗ no owner
    await _make_scope(
        db_session, integ, owner_id=None,
        environment="dev", created_at=before_month,
    )
    # ✗ prod
    await _make_scope(
        db_session, integ, owner_id=owner.id,
        environment="prod", created_at=before_month,
    )
    # ✗ created AFTER month start
    too_new = await _make_scope(
        db_session, integ, owner_id=owner.id,
        environment="dev", created_days_ago=1,
    )
    # Make sure "too_new" is past month_start clearly.
    too_new.created_at = datetime(
        target_year, target_month, 15, tzinfo=UTC,
    )

    await db_session.commit()

    svc = OwnerConfirmationService(db_session)
    eligible = await svc.list_scopes_with_zero_spend_last_month(
        billing_year=target_year, billing_month=target_month,
    )
    eligible_ids = {s.id for s in eligible}
    assert zero_scope.id in eligible_ids
    assert spent_scope.id not in eligible_ids
    assert too_new.id not in eligible_ids


# ── Confirm endpoint flips pending → responded + resets clock ─────


@pytest.mark.asyncio
async def test_confirm_endpoint_flips_pending_and_resets_clock(
    client: AsyncClient, db_session: AsyncSession,
):
    integ = await _make_integration(db_session)
    owner = await _make_person(db_session, email="confirm_owner@test.com")
    scope = await _make_scope(
        db_session, integ, owner_id=owner.id,
        environment="dev",
        last_owner_confirmed_at=datetime.now(UTC) - timedelta(days=200),
    )

    pending = CmdbCloudScopeOwnerConfirmation(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        owner_person_id=str(owner.id),
        kind=OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING,
        status=OWNER_CONFIRMATION_STATUS_PENDING,
        notified_at=datetime.now(UTC),
    )
    db_session.add(pending)
    await db_session.commit()

    from app.core.security import create_access_token

    token = create_access_token(str(owner.id), ["employee"])
    resp = await client.post(
        f"/api/v1/cmdb/scopes/{scope.id}/owner-confirmations/"
        f"{pending.id}/semiannual-confirm",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, resp.text

    await db_session.refresh(pending)
    assert pending.status == OWNER_CONFIRMATION_STATUS_RESPONDED
    assert pending.response_kind == "semiannual_confirmed"

    refreshed_scope = await db_session.get(CmdbCloudScope, scope.id)
    assert refreshed_scope.last_owner_confirmed_at is not None
    age = datetime.now(UTC) - refreshed_scope.last_owner_confirmed_at.replace(
        tzinfo=UTC if refreshed_scope.last_owner_confirmed_at.tzinfo is None
        else refreshed_scope.last_owner_confirmed_at.tzinfo
    )
    assert age < timedelta(minutes=5)


# ── Zero-spend respond: 'keep' requires reason ────────────────────


@pytest.mark.asyncio
async def test_zero_spend_keep_requires_reason(
    client: AsyncClient, db_session: AsyncSession,
):
    integ = await _make_integration(db_session)
    owner = await _make_person(db_session, email="zskeep_owner@test.com")
    scope = await _make_scope(
        db_session, integ, owner_id=owner.id, environment="dev",
    )
    pending = CmdbCloudScopeOwnerConfirmation(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        owner_person_id=str(owner.id),
        kind=OWNER_CONFIRMATION_KIND_ZERO_SPEND_PENDING,
        status=OWNER_CONFIRMATION_STATUS_PENDING,
        notified_at=datetime.now(UTC),
    )
    db_session.add(pending)
    await db_session.commit()

    from app.core.security import create_access_token

    token = create_access_token(str(owner.id), ["employee"])
    # keep without reason → 400
    resp = await client.post(
        f"/api/v1/cmdb/scopes/{scope.id}/owner-confirmations/"
        f"{pending.id}/zero-spend-respond",
        headers={"Authorization": f"Bearer {token}"},
        json={"action": "keep"},
    )
    assert resp.status_code == 400, resp.text
    assert resp.json()["errors"][0]["code"] == "OWNER_CONFIRMATION_REASON_REQUIRED"

    # keep with reason → 200 and reason persisted
    resp = await client.post(
        f"/api/v1/cmdb/scopes/{scope.id}/owner-confirmations/"
        f"{pending.id}/zero-spend-respond",
        headers={"Authorization": f"Bearer {token}"},
        json={"action": "keep", "reason": "下季度迁移过来用"},
    )
    assert resp.status_code == 200, resp.text
    await db_session.refresh(pending)
    assert pending.response_kind == "zero_spend_keep"
    assert pending.reason == "下季度迁移过来用"


# ── Escalation sweep: RETIRED 2026-08-24 ─────────────────────────
#
# ``escalate_overdue`` and its two tests lived here. The policy it
# implemented — 14 days of owner silence hands the prompt to every IT
# admin — was replaced by re-sending the prompt to the OWNER daily.
# The tests are gone rather than adapted for a reason worth recording:
# both seeded an ``it_admin`` AuthRole themselves, and production has no
# such row (the canonical name is ``global_admin``). So they passed for
# three months against a roster query that matched nobody in prod —
# they encoded the bug instead of catching it. The replacement coverage
# is in ``test_owner_governance_timeline_2026_08_24.py``, whose fixtures
# grant only roles production actually has.


# ── Transfer notifies the new owner via outbox ────────────────────


@pytest.mark.asyncio
async def test_assign_scope_owner_writes_outbox_for_new_owner(
    client: AsyncClient, db_session: AsyncSession,
):
    """``assign_scope_owner`` writes a notification_outbox row for the
    new owner so the welcome-card path is exercised end-to-end."""
    from app.core.security import create_access_token

    integ = await _make_integration(db_session)
    a = await _make_person(db_session, email="from_a@test.com")
    b = await _make_person(db_session, email="to_b@test.com")
    scope = await _make_scope(
        db_session, integ, owner_id=a.id, environment="dev",
    )
    await db_session.commit()

    a_token = create_access_token(str(a.id), ["employee"])
    resp = await client.patch(
        f"/api/v1/cmdb/scopes/{scope.id}",
        json={"owner_person_id": str(b.id)},
        headers={"Authorization": f"Bearer {a_token}"},
    )
    assert resp.status_code == 200, resp.text

    res = await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.recipient_person_id == b.id,
            NotificationOutbox.template_key == "cloud.owner_transferred_to_you",
        )
    )
    outbox_rows = res.scalars().all()
    assert len(outbox_rows) == 1, (
        "transfer must trigger exactly one welcome-card outbox row for "
        "the new owner — see assign_scope_owner patch in PR #168."
    )

    # last_owner_confirmed_at bumped to "now" by the transfer.
    refreshed_scope = await db_session.get(CmdbCloudScope, scope.id)
    assert refreshed_scope.last_owner_confirmed_at is not None
    delta = (
        datetime.now(UTC) - refreshed_scope.last_owner_confirmed_at.replace(
            tzinfo=UTC,
        )
        if refreshed_scope.last_owner_confirmed_at.tzinfo is None
        else datetime.now(UTC) - refreshed_scope.last_owner_confirmed_at
    )
    assert delta < timedelta(minutes=5)


# ── governance-timeline transfer row (regression coverage) ─────────


@pytest.mark.asyncio
async def test_owner_transfer_without_pending_writes_governance_row(
    client: AsyncClient, db_session: AsyncSession,
):
    """A self-initiated transfer with no outstanding semiannual prompt
    writes a standalone ``owner_transferred`` row so the governance
    timeline reflects the transfer.

    Without this row, an owner / admin viewing "Owner 与治理记录" on the
    scope detail page sees nothing (the audit_log row alone is not
    rendered there), and the transfer is effectively invisible.
    """
    from app.core.security import create_access_token

    integ = await _make_integration(db_session)
    a = await _make_person(db_session, email="from_a@test.com")
    b = await _make_person(db_session, email="to_b@test.com")
    scope = await _make_scope(
        db_session, integ, owner_id=a.id, environment="dev",
    )
    await db_session.commit()

    a_token = create_access_token(str(a.id), ["employee"])
    resp = await client.patch(
        f"/api/v1/cmdb/scopes/{scope.id}",
        json={"owner_person_id": str(b.id)},
        headers={"Authorization": f"Bearer {a_token}"},
    )
    assert resp.status_code == 200, resp.text

    res = await db_session.execute(
        select(CmdbCloudScopeOwnerConfirmation).where(
            CmdbCloudScopeOwnerConfirmation.scope_id == str(scope.id),
        )
    )
    rows = list(res.scalars().all())
    assert len(rows) == 1, (
        "transfer with no outstanding semiannual prompt must write "
        "exactly one owner_transferred row"
    )
    row = rows[0]
    assert row.kind == OWNER_CONFIRMATION_KIND_OWNER_TRANSFERRED
    assert row.status == OWNER_CONFIRMATION_STATUS_RESPONDED
    assert row.owner_person_id == str(a.id), "prev owner snapshot"
    assert row.transferred_to_person_id == str(b.id), "new owner"
    assert row.responded_by_person_id == str(a.id), (
        "actor is the original owner (self-transfer path)"
    )
    assert row.response_at is not None


@pytest.mark.asyncio
async def test_owner_transfer_with_pending_does_not_double_write(
    client: AsyncClient, db_session: AsyncSession,
):
    """When a semiannual_pending row IS outstanding, the transfer flips
    it to ``semiannual_transferred`` and does NOT write an extra
    ``owner_transferred`` row — one event, one timeline entry.
    """
    from app.core.security import create_access_token

    integ = await _make_integration(db_session)
    a = await _make_person(db_session, email="from_a@test.com")
    b = await _make_person(db_session, email="to_b@test.com")
    scope = await _make_scope(
        db_session, integ, owner_id=a.id, environment="dev",
    )
    # Pre-seed an outstanding semiannual_pending prompt for the current
    # owner — mimics the cron having dispatched the prompt to A.
    pending = CmdbCloudScopeOwnerConfirmation(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        owner_person_id=str(a.id),
        kind=OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING,
        status=OWNER_CONFIRMATION_STATUS_PENDING,
        notified_at=datetime.now(UTC) - timedelta(days=1),
        created_at=datetime.now(UTC) - timedelta(days=1),
        updated_at=datetime.now(UTC) - timedelta(days=1),
    )
    db_session.add(pending)
    await db_session.commit()

    a_token = create_access_token(str(a.id), ["employee"])
    resp = await client.patch(
        f"/api/v1/cmdb/scopes/{scope.id}",
        json={"owner_person_id": str(b.id)},
        headers={"Authorization": f"Bearer {a_token}"},
    )
    assert resp.status_code == 200, resp.text

    res = await db_session.execute(
        select(CmdbCloudScopeOwnerConfirmation).where(
            CmdbCloudScopeOwnerConfirmation.scope_id == str(scope.id),
        )
    )
    rows = list(res.scalars().all())
    kinds = sorted(r.kind for r in rows)
    statuses = sorted(r.status for r in rows)
    # Exactly one row (the original pending, now flipped). No extra
    # owner_transferred leaf — the flipped row IS the transfer record.
    assert len(rows) == 1, (
        "pending-flip path must not double-write a governance row"
    )
    assert kinds == [OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING], (
        "kind stays semiannual_pending; only status/response_kind flip"
    )
    assert statuses == [OWNER_CONFIRMATION_STATUS_RESPONDED]
    flipped = rows[0]
    assert flipped.response_kind == OWNER_CONFIRMATION_KIND_SEMIANNUAL_TRANSFERRED
    assert flipped.transferred_to_person_id == str(b.id)
    # ``responded_by_person_id`` must stamp the ACTOR (here A — the
    # current owner who clicked transfer), NOT the new owner B. The
    # detail-page UI relies on comparing this against the prev owner
    # to distinguish self-transfer from admin-initiated; stamping B
    # would make every transfer look like a self-transfer.
    assert flipped.responded_by_person_id == str(a.id)


@pytest.mark.asyncio
async def test_owner_transfer_admin_initiated_stamps_admin_as_actor(
    client: AsyncClient, db_session: AsyncSession,
):
    """Admin-initiated rebind: A is current owner, B is the new owner,
    GLOBAL ADMIN G initiates the PATCH. The governance leaf must stamp
    ``responded_by_person_id = G``, NOT A or B. Pins the
    self-vs-admin distinguishing claim documented in the
    ``assign_scope_owner`` patch.
    """
    from app.auth.models import AuthRole, AuthUserRole
    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson

    integ = await _make_integration(db_session)
    a = await _make_person(db_session, email="from_a_adm@test.com")
    b = await _make_person(db_session, email="to_b_adm@test.com")
    # Global admin actor
    g_id = uuid.uuid4()
    g = IdentityPerson(
        id=g_id,
        work_email="actor-admin@test.com",
        display_name="Global Admin Actor",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-adm-{uuid.uuid4().hex[:6]}",
    )
    db_session.add(g)
    await db_session.flush()
    admin_role_res = await db_session.execute(
        select(AuthRole.id).where(AuthRole.name == "global_admin")
    )
    admin_role_id = admin_role_res.scalar_one()
    db_session.add(AuthUserRole(user_id=g_id, role_id=admin_role_id))
    scope = await _make_scope(
        db_session, integ, owner_id=a.id, environment="dev",
    )
    await db_session.commit()

    g_token = create_access_token(str(g_id), ["global_admin"])
    resp = await client.patch(
        f"/api/v1/cmdb/scopes/{scope.id}",
        json={"owner_person_id": str(b.id)},
        headers={"Authorization": f"Bearer {g_token}"},
    )
    assert resp.status_code == 200, resp.text

    res = await db_session.execute(
        select(CmdbCloudScopeOwnerConfirmation).where(
            CmdbCloudScopeOwnerConfirmation.scope_id == str(scope.id),
        )
    )
    rows = list(res.scalars().all())
    assert len(rows) == 1
    row = rows[0]
    assert row.kind == OWNER_CONFIRMATION_KIND_OWNER_TRANSFERRED
    assert row.owner_person_id == str(a.id), "prev owner snapshot"
    assert row.transferred_to_person_id == str(b.id), "new owner"
    assert row.responded_by_person_id == str(g_id), (
        "actor is the ADMIN, not the prev owner or new owner — the UI "
        "uses this to render admin-initiated vs self-transfer distinctly"
    )


# ── Handler: min_age_days config override (init sweep) ────────────


@pytest.mark.asyncio
async def test_handler_passes_min_age_days_override_through_to_service(
    db_session: AsyncSession,
):
    """Operator-facing contract: admin triggers the cron via
    ``POST /scheduled-jobs/{id}/trigger`` with
    ``config_override={"min_age_days": 0}`` to issue first-time cards
    for a tenant whose scopes are all < 30 days old. This test pins
    the wire: handler reads ``config['min_age_days']`` and forwards it
    verbatim to ``OwnerConfirmationService.list_scopes_due_semiannual``.

    The full eligibility behaviour for ``min_age_days=0`` is covered by
    ``test_semiannual_min_age_days_zero_picks_up_brand_new_scopes`` at
    the service level. This handler-level test stays mock-based to
    avoid coupling to the global DB state of other tests' scopes."""
    from unittest.mock import AsyncMock, patch

    from app.scheduled_jobs.cloud_owner_handlers import (
        run_owner_semiannual_confirmation,
    )

    captured_kwargs: list[dict] = []

    async def _fake_list(self, **kwargs):
        captured_kwargs.append(kwargs)
        return []  # empty -> handler short-circuits past the per-scope loop

    with (
        patch(
            "app.domains.cmdb.owner_confirmation_service."
            "OwnerConfirmationService.list_scopes_due_semiannual",
            new=_fake_list,
        ),
        patch(
            "app.domains.cmdb.owner_confirmation_service."
            "OwnerConfirmationService.remind_unanswered_owners",
            new=AsyncMock(return_value={"reminded": 0, "errors": 0}),
        ),
    ):
        # 1. Default (no override) → min_age_days=30 (the prod cron path).
        await run_owner_semiannual_confirmation(db_session, {})
        assert captured_kwargs[-1] == {"min_age_days": 30}, (
            "scheduled tick must keep the 30-day newborn guard"
        )

        # 2. Init-sweep override → min_age_days=0.
        await run_owner_semiannual_confirmation(
            db_session, {"min_age_days": 0},
        )
        assert captured_kwargs[-1] == {"min_age_days": 0}, (
            "admin init-sweep override must reach the service"
        )

        # 3. String int (JSON-from-frontend shape) coerces.
        await run_owner_semiannual_confirmation(
            db_session, {"min_age_days": "0"},
        )
        assert captured_kwargs[-1] == {"min_age_days": 0}


@pytest.mark.asyncio
async def test_handler_min_age_days_negative_rejected(
    db_session: AsyncSession,
):
    """Defence-in-depth: a malformed override (negative / non-int)
    raises before the sweep starts so the failure-notifier pages
    instead of silently skipping every scope."""
    from app.scheduled_jobs.cloud_owner_handlers import (
        run_owner_semiannual_confirmation,
    )

    with pytest.raises(ValueError, match="min_age_days"):
        await run_owner_semiannual_confirmation(
            db_session, {"min_age_days": -1},
        )
    with pytest.raises(ValueError, match="min_age_days"):
        await run_owner_semiannual_confirmation(
            db_session, {"min_age_days": "not-an-int"},
        )


@pytest.mark.asyncio
async def test_list_owner_confirmations_populates_owner_leaver_indicators(
    client: AsyncClient, db_session: AsyncSession, admin_headers,
):
    """The owner-confirmation timeline carries owner_employee_no /
    owner_employment_status (owner only) so OwnerGovernanceTab can render
    the owner cell through <PersonTag> with the 已离职 treatment. The
    responder is intentionally NOT leaver-enriched (name-only)."""
    integ = await _make_integration(db_session)
    owner = await _make_person(
        db_session, email="gone_owner@test.com",
        employment_status="left", employee_no="P0911",
    )
    responder = await _make_person(db_session, email="responder@test.com")
    scope = await _make_scope(
        db_session, integ, owner_id=owner.id, environment="dev",
    )
    conf = CmdbCloudScopeOwnerConfirmation(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        owner_person_id=str(owner.id),
        kind=OWNER_CONFIRMATION_KIND_SEMIANNUAL_PENDING,
        status=OWNER_CONFIRMATION_STATUS_PENDING,
        responded_by_person_id=str(responder.id),
        notified_at=datetime.now(UTC),
    )
    db_session.add(conf)
    await db_session.commit()

    resp = await client.get(
        f"/api/v1/cmdb/scopes/{scope.id}/owner-confirmations",
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    row = resp.json()["data"][0]
    assert row["owner_employee_no"] == "P0911"
    assert row["owner_employment_status"] == "left"
    # Responder is name-only — no leaver enrichment fields for it.
    assert row["responded_by_person_id"] == str(responder.id)
    assert "responded_by_employee_no" not in row