"""Unit tests for the Atlassian production-site activity directory. The previous idle-seat attempt mis-computed "90 days unused" by trusting a sandbox site's frozen ``last_active``. These tests pin the *correct* behaviour: per (person, product) take the MAX ``last_active`` across the paid PRODUCTION sites only, ignore sandbox / external sites, and treat a blank ``last_active`` as "undetermined" (never idle). All HTTP is mocked via respx (driver methods); the parsing helpers are pure and need neither HTTP nor DB. """ from __future__ import annotations from datetime import date import httpx import pytest import respx from app.integrations.atlassian_activity import ( DEFAULT_PRODUCTION_SITES, PRODUCT_CONFLUENCE, PRODUCT_JIRA_SERVICEDESK, PRODUCT_JIRA_SOFTWARE, build_account_activity, fetch_admin_account_ids, fetch_atlassian_activity, normalize_site_host, parse_last_active_date, product_family_for_access_key, product_family_for_group, ) from app.integrations.drivers.jira_software import ( _ADMIN_API_BASE, JiraSoftwareDriver, ) _SITE = "tidb.atlassian.net" _BASE = f"https://{_SITE}/rest/api/3" _ORG = "org-uuid-1234" _CONFIG = { "site_url": _SITE, "admin_email": "it-admin@example.com", "api_token": "ATATT-test-token", "org_id": _ORG, "org_admin_api_key": "ATCTT-org-admin-key", } def _driver(cfg: dict | None = None) -> JiraSoftwareDriver: return JiraSoftwareDriver(integration_id="jira-test", config=cfg or _CONFIG) # ── Pure classification helpers ──────────────────────────────────── def test_product_family_for_group(): assert product_family_for_group("jira-software-users") == PRODUCT_JIRA_SOFTWARE assert product_family_for_group("jira-servicedesk-users") == PRODUCT_JIRA_SERVICEDESK assert product_family_for_group("confluence-users") == PRODUCT_CONFLUENCE # servicedesk must not be swallowed by a broad "software" match assert product_family_for_group("jira-servicedesk-team") == PRODUCT_JIRA_SERVICEDESK assert product_family_for_group("random-group") is None assert product_family_for_group("") is None def test_product_family_for_access_key(): assert product_family_for_access_key("jira-software") == PRODUCT_JIRA_SOFTWARE assert product_family_for_access_key("jira-servicedesk") == PRODUCT_JIRA_SERVICEDESK assert product_family_for_access_key("confluence") == PRODUCT_CONFLUENCE # out of scope keys assert product_family_for_access_key("jira-core") is None assert product_family_for_access_key("jira-product-discovery") is None assert product_family_for_access_key(None) is None def test_parse_last_active_date_nanoseconds(): # nanosecond UTC — datetime.fromisoformat would raise; we take the date assert parse_last_active_date("2026-05-26T14:24:25.994292581Z") == date(2026, 5, 26) assert parse_last_active_date("2026-01-02") == date(2026, 1, 2) assert parse_last_active_date("") is None assert parse_last_active_date(None) is None assert parse_last_active_date("not-a-date") is None assert parse_last_active_date(12345) is None def test_normalize_site_host(): assert normalize_site_host("https://tidb.atlassian.net/") == "tidb.atlassian.net" assert normalize_site_host("tidb.atlassian.net") == "tidb.atlassian.net" assert normalize_site_host("http://pingcap-us.atlassian.net/wiki") == "pingcap-us.atlassian.net" assert normalize_site_host("") == "" # ── build_account_activity: the MAX-across-prod-sites core ───────── def test_sandbox_does_not_fool_idle_computation(): """Case ①: a sandbox's ancient last_active must NOT win over the real production last_active (the exact bug the last version had).""" rec = { "account_id": "acct-huguo", "email": "Hu.Guo@pingcap.com", "account_status": "active", "last_active": "2024-01-01T00:00:00Z", "product_access": [ {"key": "jira-software", "url": "https://tidb-sandbox.atlassian.net", "last_active": "2023-02-02T00:00:00Z"}, {"key": "jira-software", "url": "https://tidb.atlassian.net", "last_active": "2026-05-20T10:00:00.123456789Z"}, ], } acct = build_account_activity(rec, dict(DEFAULT_PRODUCTION_SITES)) assert acct.email == "hu.guo@pingcap.com" assert acct.is_active # prod (2026-05-20) wins over sandbox (2023) assert acct.last_active_for(PRODUCT_JIRA_SOFTWARE) == date(2026, 5, 20) def test_confluence_max_across_two_prod_sites(): rec = { "account_id": "a1", "email": "x@p.com", "account_status": "active", "product_access": [ {"key": "confluence", "url": "tidb.atlassian.net", "last_active": "2026-05-01T00:00:00Z"}, {"key": "confluence", "url": "pingcap-us.atlassian.net", "last_active": "2026-05-25T00:00:00Z"}, # external personal site must be ignored even though newer {"key": "confluence", "url": "yanyanqing.atlassian.net", "last_active": "2026-05-27T00:00:00Z"}, ], } acct = build_account_activity(rec, dict(DEFAULT_PRODUCTION_SITES)) assert acct.last_active_for(PRODUCT_CONFLUENCE) == date(2026, 5, 25) def test_blank_last_active_is_undetermined(): """Case ②: a blank last_active resolves to None (never idle).""" rec = { "account_id": "a2", "email": "y@p.com", "account_status": "active", "product_access": [ {"key": "jira-software", "url": "tidb.atlassian.net", "last_active": ""}, ], } acct = build_account_activity(rec, dict(DEFAULT_PRODUCTION_SITES)) assert acct.last_active_for(PRODUCT_JIRA_SOFTWARE) is None def test_sandbox_only_yields_none(): rec = { "account_id": "a3", "email": "z@p.com", "account_status": "active", "product_access": [ {"key": "jira-software", "url": "tidb-sandbox.atlassian.net", "last_active": "2026-05-26T00:00:00Z"}, ], } acct = build_account_activity(rec, dict(DEFAULT_PRODUCTION_SITES)) assert acct.last_active_for(PRODUCT_JIRA_SOFTWARE) is None def test_org_site_admin_role_flag(): rec = { "account_id": "a4", "email": "admin@p.com", "account_status": "active", "roles": ["org-admin"], "product_access": [], } acct = build_account_activity(rec, dict(DEFAULT_PRODUCTION_SITES)) assert acct.is_org_or_site_admin def test_inactive_account_status(): rec = {"account_id": "a5", "email": "q@p.com", "account_status": "inactive"} acct = build_account_activity(rec, dict(DEFAULT_PRODUCTION_SITES)) assert not acct.is_active # ── Driver HTTP: list_org_users_with_activity + get_user_groups ──── @pytest.mark.asyncio async def test_list_org_users_with_activity_paginates(): drv = _driver() url = f"{_ADMIN_API_BASE}/orgs/{_ORG}/users" with respx.mock(assert_all_called=True) as mock: page1 = { "data": [{"account_id": "a1", "email": "a@p.com"}], "links": {"next": f"{url}?cursor=2"}, } page2 = { "data": [{"account_id": "a2", "email": "b@p.com"}], "links": {}, } route = mock.get(url__startswith=url) route.side_effect = [ httpx.Response(200, json=page1), httpx.Response(200, json=page2), ] users = await drv.list_org_users_with_activity() assert [u["account_id"] for u in users] == ["a1", "a2"] @pytest.mark.asyncio async def test_list_org_users_requires_org_id(): drv = _driver({**_CONFIG, "org_id": ""}) with pytest.raises(RuntimeError, match="org_id not configured"): await drv.list_org_users_with_activity() @pytest.mark.asyncio async def test_list_org_users_requires_org_admin_key(): cfg = {k: v for k, v in _CONFIG.items() if k != "org_admin_api_key"} drv = _driver(cfg) with pytest.raises(RuntimeError, match="org_admin_api_key"): await drv.list_org_users_with_activity() @pytest.mark.asyncio async def test_list_org_users_auth_failure_raises(): drv = _driver() url = f"{_ADMIN_API_BASE}/orgs/{_ORG}/users" with respx.mock: respx.get(url__startswith=url).mock( return_value=httpx.Response(403, text="forbidden"), ) with pytest.raises(RuntimeError, match="403"): await drv.list_org_users_with_activity() @pytest.mark.asyncio async def test_get_user_groups_ok_and_404(): drv = _driver() with respx.mock: respx.get(f"{_BASE}/user/groups").mock( return_value=httpx.Response(200, json=[ {"name": "jira-software-users", "groupId": "g1"}, {"name": "confluence-users", "groupId": "g2"}, ]), ) groups = await drv.get_user_groups("acct-1") assert groups == ["jira-software-users", "confluence-users"] with respx.mock: respx.get(f"{_BASE}/user/groups").mock( return_value=httpx.Response(404, text="no such user"), ) assert await drv.get_user_groups("acct-x") == [] # ── fetch_atlassian_activity + fetch_admin_account_ids ───────────── class _FakeDB: """Minimal stand-in: _load_production_sites only calls db.get.""" async def get(self, *_a, **_k): return None class _FakeDriver: def __init__(self, users, groups_by_member=None): self._users = users self._groups_by_member = groups_by_member or {} async def list_org_users_with_activity(self): return self._users async def list_group_members(self, group_name): return self._groups_by_member.get(group_name, []) @pytest.mark.asyncio async def test_fetch_activity_indexes_by_email(): users = [ {"account_id": "a1", "email": "Alice@p.com", "account_status": "active", "product_access": [ {"key": "jira-software", "url": "tidb.atlassian.net", "last_active": "2026-05-20T00:00:00Z"}]}, ] directory = await fetch_atlassian_activity(_FakeDB(), driver=_FakeDriver(users)) assert directory.total_accounts == 1 hit = directory.lookup_by_emails("ALICE@p.com") assert hit is not None assert hit.last_active_for(PRODUCT_JIRA_SOFTWARE) == date(2026, 5, 20) # alias fallback assert directory.lookup_by_emails(None, "alice@p.com") is hit assert directory.lookup_by_emails("nobody@p.com") is None @pytest.mark.asyncio async def test_fetch_admin_account_ids_unions_and_skips_missing(): members = { "jira-administrators": [ {"accountId": "admin-1"}, {"accountId": "admin-2"}, ], } drv = _FakeDriver([], groups_by_member=members) ids = await fetch_admin_account_ids( drv, ["jira-administrators", "nonexistent-group"], ) assert ids == {"admin-1", "admin-2"} @pytest.mark.asyncio async def test_fetch_admin_account_ids_skips_404_group(): class _DriverWith404: async def list_group_members(self, group_name): raise RuntimeError(f"Jira group {group_name!r} not found (HTTP 404).") ids = await fetch_admin_account_ids(_DriverWith404(), ["jira-administrators"]) assert ids == set() @pytest.mark.asyncio async def test_fetch_admin_account_ids_reraises_hard_failure(): class _DriverWithAuthErr: async def list_group_members(self, group_name): raise RuntimeError("Jira group/member HTTP 401: bad token") with pytest.raises(RuntimeError, match="401"): await fetch_admin_account_ids(_DriverWithAuthErr(), ["jira-administrators"])