"""Integration tests for ``jira.seat_sync`` — the product_access-based seat reconcile. Real DB (catalogs / plans / connectors / persons), respx-mocked Atlassian Org Admin API. Covers the regression cases the task calls out: ① a sandbox-only license must NOT count as a paid seat (sandbox is not in the plan's production ``site_urls``); ② a holder with a BLANK ``last_active`` still counts — a license is a license regardless of recency (admins / API-only users), the exact opposite of the idle-reclaim "blank → undetermined" rule; ③ the two Confluence subscriptions (tidb vs pingcap-us) are counted SEPARATELY — a holder on one site doesn't leak into the other plan; ④ "in the group but no license" does NOT count — counting is by product_access, so an account holding a *different* product (or no JSM product) never lands a JSM seat; ⑤ "has the license but not in the default group" DOES count — counting never consults group membership at all (no group endpoint is hit); ⑥ idempotent — re-running yields zero diffs; ⑦ drift — an existing seat whose holder lost the license is revoked (Helix bookkeeping), guarded by the ``SEAT_SYNC_REVOKE_GRACE_HOURS`` grace (25h as of 2026-05-28) + ratio floor. """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta from decimal import Decimal 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 SoftwareConnector from app.identity.person_models import IdentityPerson from app.integrations.models import Integration from app.scheduled_jobs.jira_handlers import run_jira_seat_sync _SITE = "tidb.atlassian.net" _US = "pingcap-us.atlassian.net" _SANDBOX = "tidb-sandbox.atlassian.net" _ADMIN_BASE = "https://api.atlassian.com/admin/v1" _ORG = "org-uuid-1234" def _iso_days_ago(days: int) -> str: return (datetime.now(UTC) - timedelta(days=days)).strftime( "%Y-%m-%dT%H:%M:%S.000000000Z", ) async def _integration(db: AsyncSession) -> Integration: integ = 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(integ) await db.flush() return integ async def _plan( db: AsyncSession, integ: Integration, *, catalog_name: str, plan_name: str, binding_config: dict, ) -> SoftwareLicensePlan: sw = AssetSoftwareCatalog( id=uuid.uuid4(), name=catalog_name, 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=plan_name, price=Decimal("7.50"), 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=binding_config, )) await db.flush() return 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, allocated_at: datetime | None = None, ) -> SoftwareSeatAllocation: seat = SoftwareSeatAllocation( plan_id=plan_id, user_id=user_id, status="active", allocated_at=allocated_at or datetime.now(UTC), ) db.add(seat) await db.flush() return seat def _user(account_id: str, email: str, access: list[tuple[str, str, str]], account_status: str = "active") -> dict: """``access`` = list of (product_key, site_host, last_active).""" return { "account_id": account_id, "email": email, "account_status": account_status, "product_access": [ {"key": k, "url": h, "last_active": la} for k, h, la in access ], } 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": {}}), ) # Atlassian's /rest/api/3/license/approximateLicenseCount/product/{key} # is called by the handler to source ``vendor_license_holders``. Tests # that don't set their own expectation get 404 → handler falls back # to the Helix-computed 5-bucket paid-site sum. mock.get( url__regex=r"https://[^/]+/rest/api/3/license/approximateLicenseCount/product/.+", ).mock(return_value=httpx.Response(404, json={})) async def _active_seat_user_ids( db: AsyncSession, plan_id: uuid.UUID, ) -> set[uuid.UUID]: db.expunge_all() rows = (await db.execute( select(SoftwareSeatAllocation) .where( SoftwareSeatAllocation.plan_id == plan_id, SoftwareSeatAllocation.status == "active", ) .options(lazyload("*")) )).scalars().all() return {s.user_id for s in rows} # ── ① sandbox-only license is NOT a paid seat ────────────────────── @pytest.mark.asyncio async def test_sandbox_only_off_paid_site_bucket(db_session: AsyncSession): """Sandbox-only holder is recorded in ``holders_off_paid_site`` for audit visibility (drawer tab), but is NOT counted toward ``license_holders`` — Atlassian admin's Apps / Subscriptions page only bills for paid-production sites, and the workbench top stat must agree. Prod 2026-05-30 motivation: JSM showed 85 on Helix vs 71 in Atlassian admin (13 sandbox + 2 personal-site agents inflated the headline). Also NOT imported as an active Helix allocation (the import-side filter still requires paid site).""" integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", binding_config={ "group_name": "jira-software-users", "product_key": "jira-software", "site_urls": [_SITE], "action_on_offboard": "deactivate", }, ) prod_p = await _person(db_session, "prod@pingcap.com") sandbox_p = await _person(db_session, "sandbox@pingcap.com") users = [ _user("acct-prod", "prod@pingcap.com", [("jira-software", _SITE, _iso_days_ago(3))]), # sandbox-only holder — counts in the off_paid_site bucket but # does NOT result in an active Helix seat allocation NOR in the # license_holders billing total. _user("acct-sandbox", "sandbox@pingcap.com", [("jira-software", _SANDBOX, _iso_days_ago(3))]), ] with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) summary = await run_jira_seat_sync(db_session, {}) # Sandbox holder does NOT become an active Helix seat. assert summary["seats_imported"] == 1 holders = await _active_seat_user_ids(db_session, plan.id) assert prod_p.id in holders assert sandbox_p.id not in holders pp = summary["per_plan"][0] # license_holders excludes sandbox — only the paid-site holder # counts toward the headline (matches Atlassian's Apps page). assert pp["license_holders"] == 1 # The matched-to-employee bucket holds only the paid-site holder. assert pp["holders_matched_to_employee"] == 1 # The sandbox holder lives in the off_paid_site bucket — the # operator can audit them via the drawer tab. assert pp["holders_off_paid_site"] == 1 # ── ② blank last_active still counts as a holder ─────────────────── @pytest.mark.asyncio async def test_blank_last_active_still_counts(db_session: AsyncSession): integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", binding_config={ "group_name": "jira-software-users", "product_key": "jira-software", "site_urls": [_SITE], "action_on_offboard": "deactivate", }, ) admin_p = await _person(db_session, "neverlogsin@pingcap.com") users = [ # blank last_active — an admin / API user who holds the license but # has no recorded activity. Idle-reclaim treats this as # "undetermined"; the seat COUNT must still include them. _user("acct-admin", "neverlogsin@pingcap.com", [("jira-software", _SITE, "")]), ] with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) summary = await run_jira_seat_sync(db_session, {}) assert summary["seats_imported"] == 1 holders = await _active_seat_user_ids(db_session, plan.id) assert admin_p.id in holders # ── ③ Confluence tidb vs pingcap-us counted separately ───────────── @pytest.mark.asyncio async def test_confluence_two_sites_counted_separately(db_session: AsyncSession): integ = await _integration(db_session) tidb_plan = await _plan( db_session, integ, catalog_name="Confluence", plan_name="Confluence (full member)", binding_config={ "group_name": "confluence-users", "product_key": "confluence", "site_urls": [_SITE], "action_on_offboard": "deactivate", }, ) us_plan = await _plan( db_session, integ, catalog_name="Confluence (PingCAP US)", plan_name="Confluence US (full member)", binding_config={ "group_name": "confluence-users", "product_key": "confluence", "site_urls": [_US], "action_on_offboard": "deactivate", }, ) only_tidb = await _person(db_session, "ct@pingcap.com") only_us = await _person(db_session, "cu@pingcap.com") both = await _person(db_session, "cb@pingcap.com") users = [ _user("acct-ct", "ct@pingcap.com", [("confluence", _SITE, _iso_days_ago(5))]), _user("acct-cu", "cu@pingcap.com", [("confluence", _US, _iso_days_ago(5))]), _user("acct-cb", "cb@pingcap.com", [ ("confluence", _SITE, _iso_days_ago(5)), ("confluence", _US, _iso_days_ago(5)), ]), ] with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) summary = await run_jira_seat_sync(db_session, {}) tidb_holders = await _active_seat_user_ids(db_session, tidb_plan.id) us_holders = await _active_seat_user_ids(db_session, us_plan.id) assert tidb_holders == {only_tidb.id, both.id} assert us_holders == {only_us.id, both.id} assert only_us.id not in tidb_holders assert only_tidb.id not in us_holders assert summary["seats_imported"] == 4 # 2 per plan # ── ④ in-group / other-product but no license → not counted ──────── @pytest.mark.asyncio async def test_other_product_not_counted_as_jsm(db_session: AsyncSession): integ = await _integration(db_session) jsm_plan = await _plan( db_session, integ, catalog_name="Jira Service Management", plan_name="Jira Service Management (full member)", binding_config={ # The CORRECT JSM grant group (migration 205). Counting ignores # it; this asserts the product_key is what drives the count. "group_name": "jira-servicemanagement-users-pingcap-ticket", "product_key": "jira-servicedesk", "site_urls": [_SITE], "action_on_offboard": "deactivate", }, ) real_agent = await _person(db_session, "agent@pingcap.com") not_agent = await _person(db_session, "softwareonly@pingcap.com") users = [ # real JSM agent (Atlassian's prod product key is hyphenated) _user("acct-agent", "agent@pingcap.com", [("jira-service-desk", _SITE, _iso_days_ago(2))]), # holds Jira Software only — in plenty of groups, but NO JSM # license → must not land a JSM seat _user("acct-sw", "softwareonly@pingcap.com", [("jira-software", _SITE, _iso_days_ago(2))]), ] with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) summary = await run_jira_seat_sync(db_session, {}) holders = await _active_seat_user_ids(db_session, jsm_plan.id) assert holders == {real_agent.id} assert not_agent.id not in holders assert summary["seats_imported"] == 1 # ── ⑤ license counted regardless of group membership ─────────────── @pytest.mark.asyncio async def test_counts_holders_without_consulting_groups(db_session: AsyncSession): integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", binding_config={ "group_name": "jira-software-users", "product_key": "jira-software", "site_urls": [_SITE], "action_on_offboard": "deactivate", }, ) people = [await _person(db_session, f"sw{i}@pingcap.com") for i in range(3)] users = [ _user(f"acct-{i}", f"sw{i}@pingcap.com", [("jira-software", _SITE, _iso_days_ago(i + 1))]) for i in range(3) ] # NO group endpoint is mocked. assert_all_called=True would fail if the # handler tried to hit /group/member or /user/groups — proving the # count is purely product_access driven. with respx.mock(assert_all_called=True) as mock: _mock_directory(mock, users) summary = await run_jira_seat_sync(db_session, {}) holders = await _active_seat_user_ids(db_session, plan.id) assert holders == {p.id for p in people} assert summary["seats_imported"] == 3 # ── ⑥ idempotent ─────────────────────────────────────────────────── @pytest.mark.asyncio async def test_idempotent_rerun(db_session: AsyncSession): integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", binding_config={ "group_name": "jira-software-users", "product_key": "jira-software", "site_urls": [_SITE], "action_on_offboard": "deactivate", }, ) await _person(db_session, "a@pingcap.com") await _person(db_session, "b@pingcap.com") users = [ _user("acct-a", "a@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), _user("acct-b", "b@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), ] async def _run(): with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) return await run_jira_seat_sync(db_session, {}) s1 = await _run() s2 = await _run() assert s1["seats_imported"] == 2 assert s2["seats_imported"] == 0 assert s2["seats_revoked"] == 0 assert len(await _active_seat_user_ids(db_session, plan.id)) == 2 # ── ⑦ drift: holder lost the license → seat revoked ──────────────── @pytest.mark.asyncio async def test_drift_revoke_when_license_lost(db_session: AsyncSession): integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", binding_config={ "group_name": "jira-software-users", "product_key": "jira-software", "site_urls": [_SITE], "action_on_offboard": "deactivate", }, ) keep1 = await _person(db_session, "keep1@pingcap.com") keep2 = await _person(db_session, "keep2@pingcap.com") keep3 = await _person(db_session, "keep3@pingcap.com") gone = await _person(db_session, "gone@pingcap.com") # Pre-existing seats (allocated > 2h ago so the fresh-grant grace # doesn't protect them). old = datetime.now(UTC) - timedelta(hours=3) for p in (keep1, keep2, keep3, gone): await _seat(db_session, plan_id=plan.id, user_id=p.id, allocated_at=old) # Directory: keep1/2/3 still hold the license; gone does not. users = [ _user("acct-k1", "keep1@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), _user("acct-k2", "keep2@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), _user("acct-k3", "keep3@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), ] with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) summary = await run_jira_seat_sync(db_session, {}) assert summary["seats_revoked"] == 1 holders = await _active_seat_user_ids(db_session, plan.id) assert holders == {keep1.id, keep2.id, keep3.id} assert gone.id not in holders # ── safety: empty directory must NOT mass-revoke paid seats ───────── @pytest.mark.asyncio async def test_empty_directory_refuses_mass_revoke(db_session: AsyncSession): integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", binding_config={ "group_name": "jira-software-users", "product_key": "jira-software", "site_urls": [_SITE], "action_on_offboard": "deactivate", }, ) old = datetime.now(UTC) - timedelta(hours=3) for i in range(6): p = await _person(db_session, f"paid{i}@pingcap.com") await _seat(db_session, plan_id=plan.id, user_id=p.id, allocated_at=old) # Directory returns zero holders (credentials revoked / outage). The # ratio guard must REFUSE to revoke all 6 paid seats. with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, []) summary = await run_jira_seat_sync(db_session, {}) assert summary["seats_revoked"] == 0 assert len(await _active_seat_user_ids(db_session, plan.id)) == 6 assert any("refusing to mass-revoke" in e for e in summary["errors"]) # ── seat_revocable=false (add-on plan) still counts by product_key ── @pytest.mark.asyncio async def test_seat_revocable_false_does_not_affect_counting( db_session: AsyncSession, ): """A Marketplace add-on plan (Tempo / ScriptRunner / eazyBI) reuses ``jira-software-users`` purely as a billing proxy and carries ``seat_revocable: false`` so idle_reclaim skips it. ``jira.seat_sync`` must NOT read that flag — it still counts the add-on's per-head seats by ``product_key=jira-software`` exactly like the real Jira Software plan (acceptance criterion: statistics口径 unaffected by the idle-reclaim opt-out).""" integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Tempo Timesheets", plan_name="Jira Tempo Timesheets (per Jira user)", binding_config={ "group_name": "jira-software-users", "product_key": "jira-software", "site_urls": [_SITE], "action_on_offboard": "deactivate", "seat_revocable": False, # idle-reclaim opt-out }, ) matched = await _person(db_session, "matched@pingcap.com") users = [ _user("acct-m", "matched@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), # external / service account holding a Jira license but with no # Helix person — the bill-vs-seat gap we want surfaced. _user("acct-ext", "contractor@partner.example.com", [("jira-software", _SITE, _iso_days_ago(1))]), ] with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) summary = await run_jira_seat_sync(db_session, {}) # Counting happened despite seat_revocable=false: the matched holder # got a seat, both holders are counted, the unmatched one is sampled. assert summary["seats_imported"] == 1 holders = await _active_seat_user_ids(db_session, plan.id) assert holders == {matched.id} pp = next(p for p in summary["per_plan"] if p.get("product") == "jira-software") assert pp["license_holders"] == 2 assert pp["holders_matched_to_employee"] == 1 assert pp["holders_external_unmatched"] == 1 assert pp["holders_external_unmatched_samples"] == [ "contractor@partner.example.com", ] # ── Round-56: a reconciliation identity violation on one plan must NOT # abort the whole daily sync (reachable via duplicate-email accounts) ─ @pytest.mark.asyncio async def test_reconciliation_identity_error_isolated_to_plan( db_session: AsyncSession, monkeypatch, ): """When ``record_seat_reconciliation`` raises ``SeatReconciliationIdentityError`` (e.g. the org directory legitimately holds two managed account_id rows with the same email — handled as a real case in ``atlassian_activity.fetch_atlassian_activity``), the handler must catch it per-plan, surface it in ``summary['errors']``, and NOT propagate (which would skip the audit row + every remaining plan and page on-call every tick). The import/drift seat work for the plan still commits; only that plan's snapshot is skipped. """ from app.scheduled_jobs._seat_sync_helpers import ( SeatReconciliationIdentityError, ) integ = await _integration(db_session) await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", binding_config={ "group_name": "jira-software-users", "product_key": "jira-software", "site_urls": [_SITE], "action_on_offboard": "deactivate", }, ) await _person(db_session, "prod@pingcap.com") users = [ _user("acct-prod", "prod@pingcap.com", [("jira-software", _SITE, _iso_days_ago(3))]), ] async def _boom(*_a, **_k): raise SeatReconciliationIdentityError( "forced duplicate-email identity mismatch" ) monkeypatch.setattr( "app.scheduled_jobs.jira_handlers.record_seat_reconciliation", _boom, ) with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) # Must NOT raise — the per-plan guard isolates the violation. summary = await run_jira_seat_sync(db_session, {}) # The import side still ran (seat imported before the recon step). assert summary["seats_imported"] == 1 # The violation is surfaced, not swallowed silently. assert any("identity" in e for e in summary["errors"]), summary["errors"] assert any( p.get("reconciliation_error") == "identity_mismatch" for p in summary["per_plan"] ), summary["per_plan"]