"""Reconciliation-identity tests for ``*.seat_sync`` handlers. The drawer renders:: vendor_license_holders = holders_matched + holders_external + holders_no_email + holders_leaver_residual + holders_vendor_inactive verbatim (``web/src/components/SeatReconciliationDrawer`` — ``t('seatRecon.equation6')``). ``holders_off_paid_site`` is recorded as anomaly observations for the drawer's audit tab but is **excluded from the licensed-seat total** — sandbox + personal-site holders are not part of the plan's paid subscription and the vendor does NOT bill them. See ``record_seat_reconciliation`` for the canonical statement. Two failure modes this module pins down: 1. **Off-by-N inside the snapshot row** — pre-fix, ``holders_matched`` was ``len(set_of_person_ids)`` while ``holders_external`` counted unique emails. Helix-side aliased emails (taichi.kusaka@ + japan_tidbcloud_taichi.kusaka@ both mapping to one person) made the identity break: 365 = 318 + 46 + 0 was actually 364. 2. **Helix count over-counted Atlassian's billing UI** — pre-2026-05-30, ``vendor_license_holders = len(all_holders)`` included sandbox + personal-site holders. JSM showed 85 on Helix vs 71 in Atlassian admin (15 off-paid-site agents); Confluence showed 134 vs ~105 (29 PingCAP-US-only holders surfaced on the tidb plan). The fix counts ORG-WIDE all-status all-sites and classifies each holder into exactly one of six mutually-exclusive buckets, then subtracts the off-paid-site bucket from the headline so the displayed number matches Atlassian's admin Apps page within Atlassian's own license_assigned vs product_access slip (1–12 holders, irreducible because no public Atlassian API exposes the SaaS Subscription license_used number directly). """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta from decimal import Decimal import httpx import pytest import respx from sqlalchemy.ext.asyncio import AsyncSession from app.domains.asset.models import ( AssetSoftwareCatalog, SoftwareExternalHolder, SoftwareLicensePlan, SoftwareSeatSyncStat, ) from app.domains.provisioning.models import SoftwareConnector from app.identity.person_models import IdentityPerson from app.integrations.models import Integration from app.scheduled_jobs._seat_sync_helpers import ( ExternalHolderObservation, SeatReconciliationIdentityError, record_seat_reconciliation, ) from app.scheduled_jobs.jira_handlers import run_jira_seat_sync _SITE = "tidb.atlassian.net" _SANDBOX = "tidb-sandbox.atlassian.net" _HIBERNATE = "hibernate.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 identity test", slug=f"jira-id-{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, ) -> 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", ) db.add(plan) await db.flush() db.add(SoftwareConnector( software_id=sw.id, integration_id=integ.id, is_active=True, binding_config={ "group_name": "jira-software-users", "product_key": "jira-software", "site_urls": [_SITE], "action_on_offboard": "deactivate", }, )) await db.flush() return plan async def _person( db: AsyncSession, work_email: str, *, email_aliases: list[str] | None = None, employment_status: str = "active", ) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), work_email=work_email, email_aliases=email_aliases, display_name=work_email.split("@")[0], employment_status=employment_status, source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", country="cn", ) db.add(p) await db.flush() return p def _user( account_id: str, email: str, access: list[tuple[str, str, str]], account_status: str = "active", ) -> dict: 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 approximateLicenseCount endpoint — default to 404 so the # handler falls back to the Helix-computed 5-bucket paid-site sum # (which is what these identity tests are written against). mock.get( url__regex=r"https://[^/]+/rest/api/3/license/approximateLicenseCount/product/.+", ).mock(return_value=httpx.Response(404, json={})) async def _load_stat( db: AsyncSession, plan_id: uuid.UUID, ) -> SoftwareSeatSyncStat: from sqlalchemy import select db.expunge_all() row = (await db.execute( select(SoftwareSeatSyncStat).where( SoftwareSeatSyncStat.plan_id == plan_id, ), )).scalar_one() return row async def _load_holders_by_category( db: AsyncSession, plan_id: uuid.UUID, ) -> dict[str, list[SoftwareExternalHolder]]: from sqlalchemy import select db.expunge_all() rows = (await db.execute( select(SoftwareExternalHolder).where( SoftwareExternalHolder.plan_id == plan_id, SoftwareExternalHolder.resolved == False, # noqa: E712 ), )).scalars().all() out: dict[str, list[SoftwareExternalHolder]] = {} for r in rows: out.setdefault(r.category, []).append(r) return out # ── A0. Atlassian's approximateLicenseCount is the headline source ── @pytest.mark.asyncio async def test_atlassian_approximate_license_count_overrides_bucket_sum( db_session: AsyncSession, ): """When Atlassian's ``/rest/api/3/license/approximateLicenseCount/product/{key}`` endpoint returns a value, the handler uses it as ``vendor_license_holders`` in preference to the Helix-computed 5-bucket paid-site sum. Atlassian's own license_assigned accounting may diverge from Helix's product_access count by a few holders; this is exactly the variance the new source is meant to capture. Prod 2026-05-30: Atlassian reports ``jira-software=367`` while Helix's 5-bucket sum is 361, so the workbench cell now reads 367 (closer to the Atlassian Admin Apps page's 372 than Helix's own 361).""" integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", ) await _person(db_session, "alice@pingcap.com") await _person(db_session, "bob@pingcap.com") users = [ _user("acct-a", "alice@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), _user("acct-b", "bob@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), _user("acct-ext", "ext@partner.example.com", [("jira-software", _SITE, _iso_days_ago(1))]), ] with respx.mock(assert_all_called=False) as mock: # Register the specific 200 route FIRST so it takes priority over # _mock_directory's catch-all 404 (respx is first-match-wins). # Atlassian reports 5 license-assigned users (higher than Helix's # product_access-derived 3) — common when license is held but # the user hasn't accessed the product yet. mock.get( url__regex=r"https://[^/]+/rest/api/3/license/approximateLicenseCount/product/jira-software", ).mock( return_value=httpx.Response(200, json={ "key": "license.jira-software.approximateUserCount", "value": "5", }), ) _mock_directory(mock, users) await run_jira_seat_sync(db_session, {}) stat = await _load_stat(db_session, plan.id) # Atlassian's number wins, not Helix's 5-bucket sum (=3). assert stat.vendor_license_holders == 5 # The per-bucket breakdown stays authoritative for the drawer tabs. assert stat.holders_matched == 2 assert stat.holders_external == 1 assert stat.holders_no_email == 0 assert stat.holders_leaver_residual == 0 assert stat.holders_vendor_inactive == 0 assert stat.holders_off_paid_site == 0 # ── A. healthy baseline ──────────────────────────────────────────── @pytest.mark.asyncio async def test_healthy_baseline_holds_identity(db_session: AsyncSession): """All holders are active employees on paid site — only ``holders_matched`` and ``holders_external`` are populated.""" integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", ) await _person(db_session, "alice@pingcap.com") await _person(db_session, "bob@pingcap.com") users = [ _user("acct-a", "alice@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), _user("acct-b", "bob@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), _user("acct-ext", "ext@partner.example.com", [("jira-software", _SITE, _iso_days_ago(1))]), ] with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) await run_jira_seat_sync(db_session, {}) stat = await _load_stat(db_session, plan.id) assert stat.vendor_license_holders == 3 assert stat.holders_matched == 2 assert stat.holders_external == 1 assert stat.holders_no_email == 0 assert stat.holders_leaver_residual == 0 assert stat.holders_vendor_inactive == 0 assert stat.holders_off_paid_site == 0 # Reconciliation identity: off_paid_site is excluded from # vendor_license_holders (it's anomaly-only — Atlassian does not # bill for sandbox / personal sites). assert stat.vendor_license_holders == ( stat.holders_matched + stat.holders_external + stat.holders_no_email + stat.holders_leaver_residual + stat.holders_vendor_inactive ) # ── B. leaver_residual — offboarding leak (prod 2026-05-29 case) ─── @pytest.mark.asyncio async def test_leaver_residual_bucket(db_session: AsyncSession): """A Helix-leaver whose Atlassian account is still active on the paid site shows up in ``holders_leaver_residual`` — the highest- severity bucket. This is the offboarding leak prod surfaced on 2026-05-29 (1 leaver still holding a Jira Software seat on tidb).""" integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", ) leaver = await _person( db_session, "leaver@pingcap.com", employment_status="left", ) await _person(db_session, "alice@pingcap.com") users = [ _user("acct-a", "alice@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), _user("acct-lv", "leaver@pingcap.com", [("jira-software", _SITE, _iso_days_ago(60))]), ] with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) await run_jira_seat_sync(db_session, {}) stat = await _load_stat(db_session, plan.id) assert stat.vendor_license_holders == 2 assert stat.holders_matched == 1 assert stat.holders_leaver_residual == 1 assert stat.holders_external == 0 assert stat.holders_vendor_inactive == 0 holders = await _load_holders_by_category(db_session, plan.id) assert "leaver_residual" in holders [row] = holders["leaver_residual"] assert row.email == "leaver@pingcap.com" assert row.person_id == leaver.id # deep-link target for the drawer # ── C. vendor_inactive (the lifu.wu prod case) ───────────────────── @pytest.mark.asyncio async def test_vendor_inactive_bucket(db_session: AsyncSession): """Vendor account_status is NOT ``active`` (Atlassian ``inactive``). Even though the email maps to a Helix-active person (HRIS hasn't propagated the departure yet), the holder is bucketed by precedence rule into ``vendor_inactive`` — the deactivated state outweighs the leaver/matched signal because the account isn't usable as a seat anymore.""" integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", ) # lifu.wu's prod state on 2026-05-29: Atlassian deactivated, Helix # employment_status still 'active' (HRIS sync lag). await _person(db_session, "lifu.wu@pingcap.com") await _person(db_session, "alice@pingcap.com") users = [ _user("acct-a", "alice@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), _user("acct-lf", "lifu.wu@pingcap.com", [("jira-software", _SITE, _iso_days_ago(120))], account_status="inactive"), ] with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) await run_jira_seat_sync(db_session, {}) stat = await _load_stat(db_session, plan.id) assert stat.vendor_license_holders == 2 assert stat.holders_matched == 1 assert stat.holders_vendor_inactive == 1 assert stat.holders_leaver_residual == 0 assert stat.holders_external == 0 # ── D. off_paid_site (qizhi.wang on hibernate, etc.) ─────────────── @pytest.mark.asyncio async def test_off_paid_site_bucket(db_session: AsyncSession): """Vendor account is active and holds the product but only on a NON-paid site (sandbox / personal). Surfaces in ``holders_off_paid_site`` for the drawer's audit tab, but does NOT contribute to ``vendor_license_holders`` — Atlassian admin's Apps page only bills for paid-production sites, and the headline must agree (JSM prod 2026-05-30: 85→70 after this rule, matching Atlassian's 71 within 1 holder).""" integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", ) await _person(db_session, "alice@pingcap.com") await _person(db_session, "qizhi.wang@pingcap.com") users = [ _user("acct-a", "alice@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), _user("acct-qz", "qizhi.wang@pingcap.com", [("jira-software", _HIBERNATE, _iso_days_ago(7))]), ] with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) await run_jira_seat_sync(db_session, {}) stat = await _load_stat(db_session, plan.id) # Only the paid-site holder counts toward the licensed-seat total. assert stat.vendor_license_holders == 1 assert stat.holders_matched == 1 # off_paid_site bucket still recorded for audit (drawer tab). assert stat.holders_off_paid_site == 1 holders = await _load_holders_by_category(db_session, plan.id) [row] = holders["off_paid_site"] assert row.email == "qizhi.wang@pingcap.com" # The drawer shows the actual non-paid site host so the operator # knows where the seat lives. assert _HIBERNATE in (row.site or "") # ── E. alias-collapse (the prod taichi.kusaka root cause) ─────────── @pytest.mark.asyncio async def test_alias_collapse_counts_per_account(db_session: AsyncSession): """One Helix person, two Atlassian accounts via email alias — BOTH count toward ``holders_matched`` because Atlassian charges per account (each is a billable seat). Pre-fix the matched count was a ``set`` keyed by person.id, collapsing the two to one; the new semantics count by raw account so the displayed total matches Atlassian's billing UI.""" integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", ) await _person( db_session, "taichi.kusaka@pingcap.com", email_aliases=["japan_tidbcloud_taichi.kusaka@pingcap.com"], ) await _person(db_session, "solo@pingcap.com") users = [ _user("acct-tk-primary", "taichi.kusaka@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), _user("acct-tk-alias", "japan_tidbcloud_taichi.kusaka@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), _user("acct-solo", "solo@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), ] with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) await run_jira_seat_sync(db_session, {}) stat = await _load_stat(db_session, plan.id) assert stat.vendor_license_holders == 3 # Both aliased accounts of Taichi count as employees_active — they # ARE both billable Atlassian seats. assert stat.holders_matched == 3 assert stat.holders_external == 0 assert stat.holders_leaver_residual == 0 # ── F. mixed prod-shaped scenario — every bucket non-zero ─────────── @pytest.mark.asyncio async def test_mixed_buckets_full_identity(db_session: AsyncSession): """End-to-end shape mirroring the 2026-05-29 prod jira-software breakdown (scaled-down): every row-persisted bucket non-zero, licensed-seat identity holds exactly (off_paid_site is recorded but EXCLUDED from vendor_license_holders since Atlassian does not bill for sandbox / personal sites).""" integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", ) await _person(db_session, "alice@pingcap.com") await _person(db_session, "bob@pingcap.com") await _person( db_session, "leaver@pingcap.com", employment_status="left", ) await _person(db_session, "lifu.wu@pingcap.com") await _person(db_session, "qizhi.wang@pingcap.com") users = [ # matched (×2) _user("acct-a", "alice@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), _user("acct-b", "bob@pingcap.com", [("jira-software", _SITE, _iso_days_ago(1))]), # external (×1) _user("acct-ext", "ext@partner.example.com", [("jira-software", _SITE, _iso_days_ago(1))]), # leaver_residual (×1) _user("acct-lv", "leaver@pingcap.com", [("jira-software", _SITE, _iso_days_ago(60))]), # vendor_inactive (×1) _user("acct-lf", "lifu.wu@pingcap.com", [("jira-software", _SITE, _iso_days_ago(120))], account_status="inactive"), # off_paid_site (×1) — Helix employee but only on hibernate _user("acct-qz", "qizhi.wang@pingcap.com", [("jira-software", _HIBERNATE, _iso_days_ago(7))]), ] with respx.mock(assert_all_called=False) as mock: _mock_directory(mock, users) await run_jira_seat_sync(db_session, {}) stat = await _load_stat(db_session, plan.id) # 5 paid-site billable seats (matched 2 + ext 1 + leaver 1 + vendor_inactive 1). # off_paid_site=1 still recorded for the drawer's audit tab but # EXCLUDED from the licensed-seat headline. assert stat.vendor_license_holders == 5 assert stat.holders_matched == 2 assert stat.holders_external == 1 assert stat.holders_leaver_residual == 1 assert stat.holders_vendor_inactive == 1 assert stat.holders_off_paid_site == 1 assert stat.holders_no_email == 0 # Identity (NON-NEGOTIABLE) — the helper enforces it at write time, # but assert at the read side too so a regression that bypasses # the helper would still fail this test. assert stat.vendor_license_holders == ( stat.holders_matched + stat.holders_external + stat.holders_no_email + stat.holders_leaver_residual + stat.holders_vendor_inactive ) # ── G. helper rejects bad inputs (defence in depth) ────────────────── @pytest.mark.asyncio async def test_record_seat_reconciliation_raises_on_identity_violation( db_session: AsyncSession, ): """The helper refuses to persist a snapshot when ``vendor_license_holders`` diverges from the 5-bucket paid-site sum beyond the tolerated drift. Small drift (1–5 holders) is expected when the vendor's authoritative endpoint is the source — Atlassian's own license_assigned vs Helix's product_access count don't perfectly agree. Gross divergence indicates caller-side miscount and is rejected loudly. Two reject branches: 1. ``vendor_license_holders < holders_matched`` — vendor can't bill for fewer licenses than Helix has matched employees to. 2. ``|vendor - 5bucket_sum| > max(50, 30% of sum)`` — gross drift. """ integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", ) await db_session.flush() # Branch 1: vendor < holders_matched is always invalid. with pytest.raises(SeatReconciliationIdentityError) as exc: await record_seat_reconciliation( db_session, holder_model=SoftwareExternalHolder, stat_model=SoftwareSeatSyncStat, integration_id=integ.id, software_id=plan.software_id, plan_id=plan.id, provider="jira", product_key="jira-software", site=_SITE, vendor_license_holders=300, # < holders_matched=318 holders_matched=318, holders_no_email=0, externals=[ ExternalHolderObservation( email=f"ext{i}@partner.example.com", category="external", ) for i in range(46) ], now=datetime.now(UTC), ) assert "300" in str(exc.value) assert "holders_matched" in str(exc.value) # Branch 2: gross drift from 5-bucket sum (>30% over). with pytest.raises(SeatReconciliationIdentityError) as exc: await record_seat_reconciliation( db_session, holder_model=SoftwareExternalHolder, stat_model=SoftwareSeatSyncStat, integration_id=integ.id, software_id=plan.software_id, plan_id=plan.id, provider="jira", product_key="jira-software", site=_SITE, # 5-bucket sum = 318 + 46 + 0 + 0 + 0 = 364. # Drift limit = max(50, int(364*0.30)=109) = 109. # 364 + 200 = 564 → drift 200 > 109 → reject. vendor_license_holders=564, holders_matched=318, holders_no_email=0, externals=[ ExternalHolderObservation( email=f"ext{i}@partner.example.com", category="external", ) for i in range(46) ], now=datetime.now(UTC), ) assert "564" in str(exc.value) assert "differs from" in str(exc.value) or "differ" in str(exc.value) @pytest.mark.asyncio async def test_record_seat_reconciliation_raises_on_bucket_disagreement( db_session: AsyncSession, ): """``holders_leaver_residual`` count passed by the caller MUST match the number of ``category='leaver_residual'`` observations — otherwise the headline number on the drawer disagrees with the per-tab list it points at, which would mystify the operator.""" integ = await _integration(db_session) plan = await _plan( db_session, integ, catalog_name="Jira Software", plan_name="Jira Software (full member)", ) await db_session.flush() # Caller says 2 leavers, but only sends 1 observation. Identity # sum balances (vendor=3 = matched=1 + ext=0 + no_email=0 + # leaver_residual=2 + vendor_inactive=0 + off_paid_site=0) — so # the top-line identity passes BUT the per-bucket count disagrees # with the observations. The helper catches this mismatch. with pytest.raises(SeatReconciliationIdentityError) as exc: await record_seat_reconciliation( db_session, holder_model=SoftwareExternalHolder, stat_model=SoftwareSeatSyncStat, integration_id=integ.id, software_id=plan.software_id, plan_id=plan.id, provider="jira", product_key="jira-software", site=_SITE, vendor_license_holders=3, holders_matched=1, holders_no_email=0, holders_leaver_residual=2, externals=[ ExternalHolderObservation( email="leaver@pingcap.com", category="leaver_residual", ), ], now=datetime.now(UTC), ) assert "leaver_residual" in str(exc.value)