"""Integration tests for ``software_seat.idle_reclaim``.

Real DB (seats / plans / connector / persons), respx-mocked Atlassian.
Covers the six regression cases the task calls out:

  ① a sandbox's ancient last_active must NOT make a daily prod user
     look idle (the bug the previous attempt shipped);
  ② a blank last_active is "undetermined" → never reclaimed;
  ③ an Atlassian administrator is never auto-reclaimed;
  ④ warn → buffer → keep-skips (a kept seat is not reclaimed);
  ⑤ buffer elapsed + still idle + not kept → reclaim: seat flipped +
     a ``target_type='product_access'`` provisioning task enqueued via
     the coordinator (the same pipeline self-cancel / admin-revoke use;
     the vendor-side group removals happen at task execution, NOT
     inside this job) — including multi-group holders, which the
     pre-2026-06-11 implementation punted to manual review forever;
  ⑥ idempotent — re-running mid-buffer doesn't re-stamp / re-act.

Notifications are patched to a no-op so the tests assert state-machine
behaviour, not the (separately-covered) notification plumbing.
"""
from __future__ import annotations

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

import httpx
import pytest
import respx
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import lazyload

from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
)
from app.domains.provisioning.models import (
    SOURCE_IDLE_RECLAIM,
    ProvisioningTask,
    SoftwareConnector,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration
from app.scheduled_jobs.idle_reclaim_handlers import (
    run_software_seat_idle_reclaim,
)

_SITE = "tidb.atlassian.net"
_BASE = f"https://{_SITE}/rest/api/3"
_ADMIN_BASE = "https://api.atlassian.com/admin/v1"
_ORG = "org-uuid-1234"
_GROUP = "jira-software-users"

# Job config: pin admin groups to a single group so respx only has to
# mock one group/member call for the admin guard.
_JOB_CONFIG = {"admin_group_names": ["jira-administrators"]}


async def _seed_jira(db: AsyncSession) -> tuple[AssetSoftwareCatalog, SoftwareLicensePlan]:
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(),
        name="Jira Software",
        asset_type="SaaS",
        status="active",
        offboarding_policy="auto_revoke",
    )
    db.add(sw)
    await db.flush()
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=sw.id, name="Jira Software · Standard",
        price=Decimal("7.50"), billing_cycle="monthly", status="active",
        max_seats=None,
    )
    db.add(plan)
    await db.flush()
    integration = Integration(
        id=uuid.uuid4(),
        kind="software_provisioning",
        provider="jira",
        name="Jira test",
        slug=f"jira-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=True,
        config_public={
            "site_url": _SITE,
            "admin_email": "it-admin@pingcap.com",
            "api_token": "ATATT-test-token",
            "org_id": _ORG,
            "org_admin_api_key": "ATCTT-org-admin-key",
        },
    )
    db.add(integration)
    await db.flush()
    db.add(SoftwareConnector(
        software_id=sw.id, integration_id=integration.id, is_active=True,
        # product_key present = the prod shape (required for the
        # coordinator's product_access route; verified on prod TiDB
        # 2026-06-11: all three real plans carry it).
        binding_config={
            "group_name": _GROUP, "action_on_offboard": "deactivate",
            "product_key": "jira-software", "site_urls": [_SITE],
        },
    ))
    await db.flush()
    return sw, plan


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


async def _seat(
    db: AsyncSession, *, plan_id: uuid.UUID, user_id: uuid.UUID,
    idle_warned_at: datetime | None = None,
    idle_kept_until: datetime | None = None,
) -> SoftwareSeatAllocation:
    seat = SoftwareSeatAllocation(
        plan_id=plan_id, user_id=user_id, status="active",
        idle_warned_at=idle_warned_at, idle_kept_until=idle_kept_until,
    )
    db.add(seat)
    await db.flush()
    return seat


def _iso_days_ago(days: int) -> str:
    return (datetime.now(UTC) - timedelta(days=days)).strftime(
        "%Y-%m-%dT%H:%M:%S.000000000Z",
    )


def _user(
    account_id: str, email: str, *,
    jira_software_last_active: str | None,
    extra_access: list[dict] | None = None,
    account_status: str = "active",
) -> dict:
    pa: list[dict] = []
    if jira_software_last_active is not None:
        pa.append({
            "key": "jira-software", "url": _SITE,
            "last_active": jira_software_last_active,
        })
    if extra_access:
        pa.extend(extra_access)
    return {
        "account_id": account_id,
        "email": email,
        "account_status": account_status,
        "product_access": pa,
    }


def _mock_directory(mock, users: list[dict]) -> None:
    mock.get(url__startswith=f"{_ADMIN_BASE}/orgs/{_ORG}/users").mock(
        return_value=httpx.Response(200, json={"data": users, "links": {}}),
    )


def _mock_admin_group(mock, members: list[dict] | None = None) -> None:
    mock.get(url__startswith=f"{_BASE}/group/member").mock(
        return_value=httpx.Response(
            200, json={"values": members or [], "isLast": True},
        ),
    )


async def _refetch(db: AsyncSession, seat_id: uuid.UUID) -> SoftwareSeatAllocation:
    # Detach everything from the session, then load a fresh row with all
    # relationships suppressed — avoids touching an expired/detached
    # instance's lazy attributes outside a greenlet (MissingGreenlet).
    db.expunge_all()
    return (await db.execute(
        select(SoftwareSeatAllocation)
        .where(SoftwareSeatAllocation.id == seat_id)
        .options(lazyload("*"))
    )).scalar_one()


# ── ① sandbox doesn't fool us → daily prod user not warned ─────────


@pytest.mark.asyncio
async def test_sandbox_old_date_not_misjudged(db_session: AsyncSession):
    _, plan = await _seed_jira(db_session)
    p = await _person(db_session, "hu.guo@pingcap.com")
    seat = await _seat(db_session, plan_id=plan.id, user_id=p.id)

    users = [_user(
        "acct-huguo", "hu.guo@pingcap.com",
        jira_software_last_active=_iso_days_ago(5),  # recent on prod
        extra_access=[{
            "key": "jira-software", "url": "tidb-sandbox.atlassian.net",
            "last_active": _iso_days_ago(800),  # ancient sandbox
        }],
    )]

    with patch(
        "app.notifications.service.NotificationService.dispatch", new=AsyncMock(),
    ), respx.mock(assert_all_called=False) as mock:
        _mock_directory(mock, users)
        _mock_admin_group(mock)
        summary = await run_software_seat_idle_reclaim(db_session, _JOB_CONFIG)

    assert summary["warned"] == 0
    assert summary["reclaimed"] == 0
    after = await _refetch(db_session, seat.id)
    assert after.status == "active"
    assert after.idle_warned_at is None


# ── ② blank last_active → never reclaimed ──────────────────────────


@pytest.mark.asyncio
async def test_blank_last_active_not_reclaimed(db_session: AsyncSession):
    _, plan = await _seed_jira(db_session)
    p = await _person(db_session, "blank@pingcap.com")
    # Pre-warned + buffer elapsed: even so, blank last_active must NOT
    # reclaim (undetermined wins over an existing warning).
    seat = await _seat(
        db_session, plan_id=plan.id, user_id=p.id,
        idle_warned_at=datetime.now(UTC) - timedelta(days=30),
    )
    users = [_user("acct-blank", "blank@pingcap.com", jira_software_last_active="")]

    with patch(
        "app.notifications.service.NotificationService.dispatch", new=AsyncMock(),
    ), respx.mock(assert_all_called=False) as mock:
        _mock_directory(mock, users)
        _mock_admin_group(mock)
        summary = await run_software_seat_idle_reclaim(db_session, _JOB_CONFIG)

    assert summary["reclaimed"] == 0
    assert summary["undetermined"] >= 1
    after = await _refetch(db_session, seat.id)
    assert after.status == "active"


# ── ③ admin never reclaimed ────────────────────────────────────────


@pytest.mark.asyncio
async def test_admin_not_reclaimed(db_session: AsyncSession):
    _, plan = await _seed_jira(db_session)
    p = await _person(db_session, "admin@pingcap.com")
    seat = await _seat(
        db_session, plan_id=plan.id, user_id=p.id,
        idle_warned_at=datetime.now(UTC) - timedelta(days=30),
    )
    users = [_user(
        "acct-admin", "admin@pingcap.com",
        jira_software_last_active=_iso_days_ago(400),  # very idle
    )]

    with patch(
        "app.notifications.service.NotificationService.dispatch", new=AsyncMock(),
    ), respx.mock(assert_all_called=False) as mock:
        _mock_directory(mock, users)
        # account is a member of jira-administrators → hard guardrail
        _mock_admin_group(mock, members=[{"accountId": "acct-admin"}])
        summary = await run_software_seat_idle_reclaim(db_session, _JOB_CONFIG)

    assert summary["admin_skipped"] >= 1
    assert summary["reclaimed"] == 0
    after = await _refetch(db_session, seat.id)
    assert after.status == "active"
    # admin guard also clears any stale warning
    assert after.idle_warned_at is None


# ── ④ warn → buffer → keep skips ───────────────────────────────────


@pytest.mark.asyncio
async def test_warn_then_keep_skips_reclaim(db_session: AsyncSession):
    _, plan = await _seed_jira(db_session)

    # First: a fresh idle seat gets warned (idle_warned_at stamped).
    p1 = await _person(db_session, "warnme@pingcap.com")
    seat1 = await _seat(db_session, plan_id=plan.id, user_id=p1.id)

    # Second: an idle seat past the buffer BUT kept → must be skipped.
    p2 = await _person(db_session, "keptme@pingcap.com")
    seat2 = await _seat(
        db_session, plan_id=plan.id, user_id=p2.id,
        idle_warned_at=datetime.now(UTC) - timedelta(days=30),
        idle_kept_until=datetime.now(UTC) + timedelta(days=60),
    )

    users = [
        _user("acct-warn", "warnme@pingcap.com", jira_software_last_active=_iso_days_ago(200)),
        _user("acct-kept", "keptme@pingcap.com", jira_software_last_active=_iso_days_ago(200)),
    ]

    with patch(
        "app.notifications.service.NotificationService.dispatch", new=AsyncMock(),
    ), respx.mock(assert_all_called=False) as mock:
        _mock_directory(mock, users)
        _mock_admin_group(mock)
        summary = await run_software_seat_idle_reclaim(db_session, _JOB_CONFIG)

    assert summary["warned"] == 1
    assert summary["kept_skipped"] == 1
    assert summary["reclaimed"] == 0

    after1 = await _refetch(db_session, seat1.id)
    assert after1.idle_warned_at is not None  # newly warned
    after2 = await _refetch(db_session, seat2.id)
    assert after2.status == "active"  # kept → not reclaimed


# ── ⑤ buffer elapsed, still idle, not kept → reclaim ───────────────


async def _product_access_tasks(
    db: AsyncSession, user_id: uuid.UUID,
) -> list[ProvisioningTask]:
    return list((await db.execute(
        select(ProvisioningTask)
        .where(
            ProvisioningTask.user_id == user_id,
            ProvisioningTask.target_type == "product_access",
        )
        .options(lazyload("*"))
    )).scalars().all())


@pytest.mark.asyncio
async def test_buffer_elapsed_reclaims(db_session: AsyncSession):
    _, plan = await _seed_jira(db_session)
    p = await _person(db_session, "reclaimme@pingcap.com")
    seat = await _seat(
        db_session, plan_id=plan.id, user_id=p.id,
        idle_warned_at=datetime.now(UTC) - timedelta(days=30),
    )
    users = [_user("acct-reclaim", "reclaimme@pingcap.com",
                   jira_software_last_active=_iso_days_ago(200))]

    with patch(
        "app.notifications.service.NotificationService.dispatch", new=AsyncMock(),
    ), respx.mock(assert_all_called=False) as mock:
        _mock_directory(mock, users)
        _mock_admin_group(mock)
        # No vendor write may happen inside the job — the group removal
        # belongs to the enqueued task's execution, not this cron.
        delete_route = mock.delete(url__startswith=f"{_BASE}/group/user").mock(
            return_value=httpx.Response(204),
        )
        summary = await run_software_seat_idle_reclaim(db_session, _JOB_CONFIG)

    assert summary["reclaimed"] == 1
    assert not delete_route.called
    after = await _refetch(db_session, seat.id)
    assert after.status == "revoked"
    assert after.revoked_at is not None
    assert "product_access revoke task" in (after.reason or "")

    tasks = await _product_access_tasks(db_session, p.id)
    assert len(tasks) == 1
    task = tasks[0]
    assert task.action == "revoke"
    assert task.status == "pending"
    assert task.source == SOURCE_IDLE_RECLAIM
    assert (task.context_data or {}).get("product_key") == "jira-software"
    # Failure must surface (seat back to active + admin DM), never a
    # silently-lying revoked row.
    assert (task.context_data or {}).get("revert_seat_on_failure") is True


@pytest.mark.asyncio
async def test_multi_group_holder_is_reclaimed_via_product_access(
    db_session: AsyncSession,
):
    """The 2026-06-11 fix: a holder who has the product via MORE groups
    than the bound one (jira-dev-engineers style) is reclaimed through
    the product_access pipeline like everyone else — the old behaviour
    punted these to admin manual review forever. The job itself never
    enumerates groups any more (that's ``revoke_product_access``'s job
    at task-execution time), so no group mock is needed."""
    _, plan = await _seed_jira(db_session)
    p = await _person(db_session, "multi@pingcap.com")
    seat = await _seat(
        db_session, plan_id=plan.id, user_id=p.id,
        idle_warned_at=datetime.now(UTC) - timedelta(days=30),
    )
    users = [_user("acct-multi", "multi@pingcap.com",
                   jira_software_last_active=_iso_days_ago(200))]

    async def _run():
        with patch(
            "app.notifications.service.NotificationService.dispatch",
            new=AsyncMock(),
        ), respx.mock(assert_all_called=False) as mock:
            _mock_directory(mock, users)
            _mock_admin_group(mock)
            return await run_software_seat_idle_reclaim(db_session, _JOB_CONFIG)

    s1 = await _run()
    after1 = await _refetch(db_session, seat.id)
    s2 = await _run()
    after2 = await _refetch(db_session, seat.id)

    assert s1["reclaimed"] == 1
    assert after1.status == "revoked" and after2.status == "revoked"
    # Legacy manual-review stamp cleared on reclaim; never re-stamped.
    assert after1.idle_manual_review_at is None
    # Second run: the seat is no longer ``active`` → not rescanned, no
    # second task, no double-count.
    assert s2["reclaimed"] == 0
    assert len(await _product_access_tasks(db_session, p.id)) == 1


@pytest.mark.asyncio
async def test_dry_run_previews_without_acting(db_session: AsyncSession):
    """dry_run must NOT write anything — no seat flip, no provisioning
    task, no vendor call — while counting every would-be reclaim."""
    _, plan = await _seed_jira(db_session)
    safe = await _person(db_session, "drysafe@pingcap.com")
    safe_seat = await _seat(
        db_session, plan_id=plan.id, user_id=safe.id,
        idle_warned_at=datetime.now(UTC) - timedelta(days=30),
    )
    multi = await _person(db_session, "drymulti@pingcap.com")
    multi_seat = await _seat(
        db_session, plan_id=plan.id, user_id=multi.id,
        idle_warned_at=datetime.now(UTC) - timedelta(days=30),
    )
    users = [
        _user("acct-drysafe", "drysafe@pingcap.com", jira_software_last_active=_iso_days_ago(200)),
        _user("acct-drymulti", "drymulti@pingcap.com", jira_software_last_active=_iso_days_ago(200)),
    ]

    with patch(
        "app.notifications.service.NotificationService.dispatch", new=AsyncMock(),
    ), respx.mock(assert_all_called=False) as mock:
        _mock_directory(mock, users)
        _mock_admin_group(mock)
        delete_route = mock.delete(url__startswith=f"{_BASE}/group/user").mock(
            return_value=httpx.Response(204),
        )
        summary = await run_software_seat_idle_reclaim(
            db_session, {**_JOB_CONFIG, "dry_run": True},
        )

    assert summary["dry_run"] is True
    assert summary["reclaimed"] == 2      # both idle seats would reclaim
    assert not delete_route.called        # dry-run performs no revoke
    # No state mutated, no task enqueued.
    after_safe = await _refetch(db_session, safe_seat.id)
    after_multi = await _refetch(db_session, multi_seat.id)
    assert after_safe.status == "active"
    assert after_multi.status == "active"
    assert await _product_access_tasks(db_session, safe.id) == []
    assert await _product_access_tasks(db_session, multi.id) == []


# ── ⑥ idempotent — re-run mid-buffer doesn't re-stamp / re-act ─────


@pytest.mark.asyncio
async def test_idempotent_within_buffer(db_session: AsyncSession):
    _, plan = await _seed_jira(db_session)
    p = await _person(db_session, "buffered@pingcap.com")
    warned_at = datetime.now(UTC) - timedelta(days=2)  # within 14-day buffer
    seat = await _seat(
        db_session, plan_id=plan.id, user_id=p.id, idle_warned_at=warned_at,
    )
    users = [_user("acct-buf", "buffered@pingcap.com",
                   jira_software_last_active=_iso_days_ago(200))]

    async def _run():
        with patch(
            "app.notifications.service.NotificationService.dispatch", new=AsyncMock(),
        ), respx.mock(assert_all_called=False) as mock:
            _mock_directory(mock, users)
            _mock_admin_group(mock)
            return await run_software_seat_idle_reclaim(db_session, _JOB_CONFIG)

    s1 = await _run()
    after1 = await _refetch(db_session, seat.id)
    warned_after_1 = after1.idle_warned_at
    s2 = await _run()
    after2 = await _refetch(db_session, seat.id)

    assert s1["in_buffer"] == 1 and s2["in_buffer"] == 1
    assert s1["reclaimed"] == 0 and s2["reclaimed"] == 0
    assert after1.status == "active" and after2.status == "active"
    # idle_warned_at is NOT re-stamped on subsequent runs
    assert warned_after_1 == after2.idle_warned_at


# ── Benign config skip must NOT be a raise-worthy error (review fix) ──


@pytest.mark.asyncio
async def test_unbound_plan_skips_not_errors(db_session: AsyncSession):
    """A Jira plan with no active connector binding is a benign skip —
    it must land in summary['skipped'], NOT summary['errors'], so a
    steady-state run with nobody idle doesn't false-page admins via
    raise_if_total_failure (which reads 'errors')."""
    # jira integration (so the driver loads) but a Jira plan with NO
    # connector binding.
    integration = Integration(
        id=uuid.uuid4(), kind="software_provisioning", provider="jira",
        name="Jira test", slug=f"jira-{uuid.uuid4().hex[:6]}",
        enabled=True, is_primary=True,
        config_public={
            "site_url": _SITE, "admin_email": "it-admin@pingcap.com",
            "api_token": "ATATT-test-token", "org_id": _ORG,
            "org_admin_api_key": "ATCTT-org-admin-key",
        },
    )
    db_session.add(integration)
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(), name="Jira Software", asset_type="SaaS",
        status="active", offboarding_policy="auto_revoke",
    )
    db_session.add(sw)
    await db_session.flush()
    db_session.add(SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=sw.id, name="Unbound plan",
        price=Decimal("7.50"), billing_cycle="monthly", status="active",
    ))
    await db_session.flush()

    with patch(
        "app.notifications.service.NotificationService.dispatch", new=AsyncMock(),
    ), respx.mock(assert_all_called=False) as mock:
        _mock_directory(mock, [])
        _mock_admin_group(mock)
        # Must NOT raise (the bug this locks would raise here).
        summary = await run_software_seat_idle_reclaim(db_session, _JOB_CONFIG)

    assert summary["errors"] == []
    assert any("no active connector binding" in s for s in summary["skipped"])
    assert summary["reclaimed"] == 0 and summary["warned"] == 0


@pytest.mark.asyncio
async def test_no_jira_integration_skips_not_errors(db_session: AsyncSession):
    """No Jira integration at all (driver is None) is a benign config state,
    not a failure: the early return must put 'not configured' in
    summary['skipped'] and leave summary['errors'] empty, so the daily run
    stays green and is never escalated by raise_if_total_failure."""
    # Nothing seeded — load_jira_driver returns None, so the handler takes
    # the early-return branch before any HTTP call.
    summary = await run_software_seat_idle_reclaim(db_session, _JOB_CONFIG)
    assert summary["errors"] == []
    assert any("not configured" in s for s in summary["skipped"])
    assert summary["reclaimed"] == 0 and summary["warned"] == 0


# ── Marketplace add-on plan (seat_revocable=false) must be skipped, ──
# ── while the real Jira Software plan still reclaims in the same run ──


async def _seed_addon_plan(db: AsyncSession) -> SoftwareLicensePlan:
    """A tier-billed add-on (Tempo) reusing the SAME jira-software-users
    group as a billing proxy, flagged seat_revocable=false. Reuses the
    integration ``_seed_jira`` already created so the one driver backs
    both plans (exactly the production shape)."""
    integ = (await db.execute(
        select(Integration).where(Integration.provider == "jira")
    )).scalars().first()
    assert integ is not None
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(), name="Jira Tempo Timesheets", asset_type="SaaS",
        status="active", offboarding_policy="auto_revoke",
    )
    db.add(sw)
    await db.flush()
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=sw.id,
        name="Jira Tempo Timesheets · per Jira user",
        price=Decimal("3.00"), billing_cycle="monthly", status="active",
        max_seats=None,
    )
    db.add(plan)
    await db.flush()
    db.add(SoftwareConnector(
        software_id=sw.id, integration_id=integ.id, is_active=True,
        binding_config={
            "group_name": _GROUP, "action_on_offboard": "deactivate",
            "product_key": "jira-software", "site_urls": [_SITE],
            "seat_revocable": False,
        },
    ))
    await db.flush()
    return plan


@pytest.mark.asyncio
async def test_addon_plan_skipped_real_plan_reclaimed(db_session: AsyncSession):
    """The footgun fix: an idle holder on the add-on plan (which reuses
    jira-software-users) is NEVER touched — removing them from that group
    would strip their real Jira Software access. The same holder being
    idle on the *real* Jira Software plan IS reclaimed, proving the skip
    is plan-scoped, not a blanket disable."""
    _, real_plan = await _seed_jira(db_session)
    addon_plan = await _seed_addon_plan(db_session)

    warned_30d = datetime.now(UTC) - timedelta(days=30)
    real_p = await _person(db_session, "realidle@pingcap.com")
    real_seat = await _seat(
        db_session, plan_id=real_plan.id, user_id=real_p.id,
        idle_warned_at=warned_30d,
    )
    addon_p = await _person(db_session, "addonidle@pingcap.com")
    addon_seat = await _seat(
        db_session, plan_id=addon_plan.id, user_id=addon_p.id,
        idle_warned_at=warned_30d,
    )

    users = [
        _user("acct-real", "realidle@pingcap.com",
              jira_software_last_active=_iso_days_ago(200)),
        _user("acct-addon", "addonidle@pingcap.com",
              jira_software_last_active=_iso_days_ago(200)),
    ]

    with patch(
        "app.notifications.service.NotificationService.dispatch", new=AsyncMock(),
    ), respx.mock(assert_all_called=False) as mock:
        _mock_directory(mock, users)
        _mock_admin_group(mock)
        delete_route = mock.delete(url__startswith=f"{_BASE}/group/user").mock(
            return_value=httpx.Response(204),
        )
        summary = await run_software_seat_idle_reclaim(db_session, _JOB_CONFIG)

    # Only the real plan was scanned; the add-on plan was skipped whole.
    assert summary["plans_scanned"] == 1
    assert summary["reclaimed"] == 1
    assert any("seat_revocable=false" in s for s in summary["skipped"])
    # No vendor write inside the job (the revoke runs at task execution);
    # the real holder got exactly one product_access task, the add-on
    # holder got none (that's the footgun this prevents).
    assert not delete_route.called
    assert len(await _product_access_tasks(db_session, real_p.id)) == 1
    assert await _product_access_tasks(db_session, addon_p.id) == []

    after_real = await _refetch(db_session, real_seat.id)
    after_addon = await _refetch(db_session, addon_seat.id)
    assert after_real.status == "revoked"
    assert after_addon.status == "active"
    # Add-on seat state machine untouched (not warned-cleared, not stamped).
    assert after_addon.idle_warned_at is not None