"""Owner-approved cloud grant renewal + the 2026-07-29 perpetual-grant incident regressions. Two clusters of behaviour are locked down here. **Cluster 1 — the cn68 incident (aws-account 317766989874).** "Perpetual" was decided independently at each write site and one site did not decide it at all, so a cloud admin whose access was DISCOVERED cloud-side got a 90-day TTL and was auto-revoked while the UI displayed 「永久授权」. 162 grants across three cloud admins were revoked between 2026-07-24 and 2026-07-29. Tests assert the single predicate is now honoured by the discovery writer AND both sweep passes, and that the sweep actively normalizes rows that slipped through. **Cluster 2 — renewal became an approval flow.** Renewal used to be self-service (763 renewals in three months, every one executed by the holder alone) and recycled grants were self-revivable forever (839 rows across 135 people). Tests cover the request state machine, the four-eyes rules, the second signature above the owner-approvable TTL, the lapse sweep, and the cascade that stops an open request from resurrecting deliberately-closed access. Plus the standalone "renewal must never SHORTEN an expiry" regression: four production grants lost ~8 months of validity to a 「全部续期」 click on 2026-07-15 because renew assigned ``now + 90`` unconditionally. """ from __future__ import annotations import uuid from contextlib import asynccontextmanager from datetime import UTC, datetime, timedelta from unittest.mock import patch import pytest from sqlalchemy import select, text from sqlalchemy.ext.asyncio import AsyncSession from app.auth.models import AuthUserRole from app.core.exceptions import AppException as AppExceptionLike from app.domains.cmdb.models import ( GRANT_REVOKE_REASON_EXPIRED, GRANT_STATUS_ACTIVE, GRANT_STATUS_EXPIRED, GRANT_TTL_DEFAULT_DAYS, GRANT_TTL_HARD_CAP_DAYS, RENEWAL_REQUEST_LAPSE_DAYS, RENEWAL_REVIVAL_MAX_AGE_DAYS, RENEWAL_STATUS_APPROVED, RENEWAL_STATUS_CANCELLED, RENEWAL_STATUS_LAPSED, RENEWAL_STATUS_PENDING_OWNER, RENEWAL_STATUS_REJECTED, SCOPE_STATUS_ACTIVE, CmdbCloudAccessGrant, CmdbCloudGrantRenewalRequest, CmdbCloudScope, ) from app.identity.person_models import IdentityPerson from app.integrations.models import Integration # ── Helpers ──────────────────────────────────────────────── async def _role_id(db: AsyncSession, name: str) -> str: row = await db.execute( text("SELECT id FROM auth_roles WHERE name = :n"), {"n": name}, ) return str(row.scalar_one()) async def _person( db: AsyncSession, *, email: str, name: str, roles: tuple[str, ...] = (), status: str = "active", ) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), work_email=email, display_name=name, employment_status=status, source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", ) db.add(p) await db.flush() for role in roles: db.add(AuthUserRole( user_id=p.id, role_id=uuid.UUID(await _role_id(db, role)), )) await db.commit() return p async def _integration(db: AsyncSession, provider: str = "aws") -> Integration: row = Integration( id=uuid.uuid4(), kind="cloud", provider=provider, name=f"{provider}-renew-{uuid.uuid4().hex[:6]}", slug=f"{provider}-renew-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False, capabilities=["servers"], ) db.add(row) await db.commit() return row async def _scope( db: AsyncSession, integration: Integration, *, external_id: str | None = None, owner_person_id: str | None = None, environment: str = "dev", ttl_days: int = 90, ) -> CmdbCloudScope: s = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integration.id), scope_type="aws_account", external_id=external_id or uuid.uuid4().hex[:12], # Mirror external_id so a card asserting on the account label # matches what the renderer actually shows (display_name wins). display_name=external_id or "Renew Test Scope", status=SCOPE_STATUS_ACTIVE, environment=environment, owner_person_id=owner_person_id, default_grant_ttl_days=ttl_days, ) db.add(s) await db.commit() return s async def _grant( db: AsyncSession, scope: CmdbCloudScope, person: IdentityPerson, *, role_key: str = "role/reader", status: str = GRANT_STATUS_ACTIVE, # float allowed so a test can land mid-window: the reminder bucket # is ``now + 13 < expires_at <= now + 14`` and MySQL DATETIME ROUNDS # to the second (it does not truncate), so exactly ``now + 14`` can # round a few hundred microseconds ABOVE the upper bound and miss. expires_in_days: float | None = 5, source: str = "discovered", revoked_days_ago: int | None = None, notified_stage: str | None = None, ) -> CmdbCloudAccessGrant: now = datetime.now(UTC) g = CmdbCloudAccessGrant( id=uuid.uuid4(), scope_id=str(scope.id), person_id=str(person.id), role_key=role_key, status=status, granted_at=now - timedelta(days=60), expires_at=( None if expires_in_days is None else now + timedelta(days=expires_in_days) ), source=source, last_notified_stage=notified_stage, revoked_at=( None if revoked_days_ago is None else now - timedelta(days=revoked_days_ago) ), revoke_reason=( None if revoked_days_ago is None else GRANT_REVOKE_REASON_EXPIRED ), ) db.add(g) await db.commit() return g def _patch_session_factory(db_session: AsyncSession): @asynccontextmanager async def _factory(): yield db_session return patch( "app.domains.cmdb.lifecycle_service.async_session_factory", new=_factory, ) def _naive(dt: datetime | None) -> datetime | None: """Drop the tzinfo so DB round-trips compare against in-memory values. MySQL/TiDB return offset-naive datetimes even for ``DateTime(timezone= True)`` columns, while values freshly assigned in Python are aware. Comparing the two raises TypeError. """ if dt is None: return None return dt.astimezone(UTC).replace(tzinfo=None) if dt.tzinfo else dt def _patch_lock(module: str = "app.domains.cmdb.lifecycle_service"): """No-op the Redis advisory lock, as the other lifecycle tests do. The lock is orthogonal to what these tests assert and pulling Redis into the unit surface makes them fail for unrelated reasons. """ @asynccontextmanager async def _noop(*_a, **_k): yield return patch(f"{module}.distributed_lock", new=_noop) # ═══════════════════════════════════════════════════════════ # Cluster 1 — perpetual grants (the cn68 incident) # ═══════════════════════════════════════════════════════════ @pytest.mark.asyncio @pytest.mark.parametrize("role_name", ["cloud_admin", "global_admin"]) async def test_expiry_sweep_never_revokes_an_admins_grant( db_session: AsyncSession, role_name: str, ) -> None: """A cloud/global admin grant carrying a past-due TTL must NOT be revoked, and the sweep must clear the bogus TTL. Before the fix ``sweep_expiring_grants`` had owner + prod exemptions but no cloud-admin exemption (its sibling ``sweep_inactive_grants`` did), so this row was revoked at T+0 while the member table showed 「永久授权」. """ from app.domains.cmdb.lifecycle_service import sweep_expiring_grants admin = await _person( db_session, email=f"{role_name}-sweep@test.com", name="Admin", roles=(role_name,), ) owner = await _person(db_session, email="ow-sweep@test.com", name="Own") integ = await _integration(db_session) scope = await _scope(db_session, integ, owner_person_id=str(owner.id)) grant = await _grant( db_session, scope, admin, expires_in_days=-1, notified_stage="t7", ) gid = grant.id summary = await sweep_expiring_grants(db_session) await db_session.commit() db_session.expire_all() after = await db_session.get(CmdbCloudAccessGrant, gid) assert after is not None assert after.status == GRANT_STATUS_ACTIVE, ( f"a {role_name} grant is perpetual and must never be auto-revoked" ) assert after.expires_at is None, ( "the sweep must normalize the bogus TTL away, not merely skip the " "row — leaving it means every other reader still sees a fake expiry" ) assert after.last_notified_stage is None assert summary["perpetual_normalized"] >= 1, summary @pytest.mark.asyncio async def test_expiry_sweep_stamps_ttl_on_null_expiry_with_no_reason( db_session: AsyncSession, ) -> None: """Pass 0b — ``expires_at IS NULL`` held for no reason gets a fresh TTL. NULL is legitimate for exactly two populations: a perpetual-ROLE holder and the scope's CURRENT owner. A row that is neither is invisible to both sweep passes (they key on ``expires_at IS NOT NULL``) — a permanent grant nobody revisits. They are created by the auto-owner backfill writing the then-owner's grant perpetual (correct) and ownership later transferring while the handover re-stamp skipped NULL rows. Production 2026-08-18: 7 such rows from the 2026-06-02 backfill batch. Stamping a TTL (not revoking) is the gentle correct action: the holder re-enters the normal cycle with a full T-14 / T-7 / T-1 runway. """ from app.domains.cmdb.lifecycle_service import sweep_expiring_grants owner = await _person(db_session, email="pass0b-own@test.com", name="Own") stranger = await _person(db_session, email="pass0b-str@test.com", name="Str") admin = await _person( db_session, email="pass0b-adm@test.com", name="Adm", roles=("cloud_admin",), ) integ = await _integration(db_session) scope = await _scope(db_session, integ, owner_person_id=str(owner.id)) prod_scope = await _scope( db_session, integ, owner_person_id=str(owner.id), environment="prod", ) # (a) neither owner nor perpetual role → must be stamped. unreasoned = await _grant(db_session, scope, stranger, expires_in_days=None) # (b) the scope's current owner → NULL is correct, leave alone. owner_grant = await _grant(db_session, scope, owner, expires_in_days=None) # (c) perpetual by ROLE → NULL is correct, leave alone. admin_grant = await _grant(db_session, scope, admin, expires_in_days=None) # (d) same shape as (a) but on a PROD scope → Helix does not manage prod. prod_grant = await _grant( db_session, prod_scope, stranger, expires_in_days=None, ) ids = (unreasoned.id, owner_grant.id, admin_grant.id, prod_grant.id) summary = await sweep_expiring_grants(db_session) await db_session.commit() db_session.expire_all() a, b, c, d = [await db_session.get(CmdbCloudAccessGrant, i) for i in ids] assert a is not None and a.expires_at is not None, ( "a NULL-expiry grant held by neither the owner nor a perpetual-role " "holder is invisible to every pass — it must be stamped" ) assert a.last_notified_stage is None, "must get the full reminder runway" assert a.status == GRANT_STATUS_ACTIVE, "stamp a TTL, never revoke here" assert b is not None and b.expires_at is None, "current owner stays NULL" assert c is not None and c.expires_at is None, "cloud_admin stays NULL" assert d is not None and d.expires_at is None, ( "prod cloud access is not managed by Helix at all" ) assert summary["unreasoned_null_expiry_stamped"] == 1, summary # Converges: a second run finds nothing left to heal. again = await sweep_expiring_grants(db_session) await db_session.commit() assert again["unreasoned_null_expiry_stamped"] == 0, again @pytest.mark.asyncio @pytest.mark.parametrize("role_name", ["cloud_admin", "global_admin"]) async def test_expiry_sweep_does_not_nag_perpetual_admin( db_session: AsyncSession, role_name: str, ) -> None: """No T-14/T-7/T-1 card for a grant that is never revoked. xiaopeng.zhu (cloud_admin) received 63 expiry cards in one day for 93 grants that all displayed 「永久授权」. """ from app.domains.cmdb.lifecycle_service import sweep_expiring_grants admin = await _person( db_session, email=f"{role_name}-nag@test.com", name="Admin Nag", roles=(role_name,), ) owner = await _person(db_session, email="ow-nag@test.com", name="Own") integ = await _integration(db_session) scope = await _scope(db_session, integ, owner_person_id=str(owner.id)) await _grant(db_session, scope, admin, expires_in_days=13) summary = await sweep_expiring_grants(db_session) await db_session.commit() assert summary["notified"] == 0, ( f"{role_name} must not be nagged about a perpetual grant: {summary}" ) @pytest.mark.asyncio @pytest.mark.parametrize("role_name", ["cloud_admin", "global_admin"]) async def test_discovered_grant_for_admin_is_perpetual( db_session: AsyncSession, role_name: str, ) -> None: """grants_sync must consult the perpetual predicate on INSERT. This is the writer that never checked it — the root cause. A cloud admin discovered cloud-side got ``now + default_grant_ttl_days``. """ from app.domains.cmdb.lifecycle_service import _apply_discovered_grants from app.integrations.driver_base import CloudScopeRef, DiscoveredGrant admin = await _person( db_session, email=f"{role_name}-disc@test.com", name="Admin Disc", roles=(role_name,), ) member = await _person(db_session, email="mem-disc@test.com", name="Mem") integ = await _integration(db_session) scope = await _scope(db_session, integ) ref = CloudScopeRef( scope_type=scope.scope_type, external_id=scope.external_id, display_name=scope.display_name, ) now = datetime.now(UTC) discovered = [ DiscoveredGrant( principal_email=admin.work_email, principal_type="user", role_key="role/admin", role_display_name="Admin", external_ref="x::admin", scope=ref, granted_at=now, raw={}, ), DiscoveredGrant( principal_email=member.work_email, principal_type="user", role_key="role/reader", role_display_name="Reader", external_ref="x::reader", scope=ref, granted_at=now, raw={}, ), ] class _D: provider = "aws" owner_role_keys = frozenset() await _apply_discovered_grants( db_session, scope, discovered, _D(), str(uuid.uuid4()), now, ) await db_session.commit() rows = (await db_session.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.scope_id == str(scope.id), ) )).scalars().all() by_person = {str(r.person_id): r for r in rows} assert by_person[str(admin.id)].expires_at is None, ( f"discovered grant for {role_name} must be perpetual" ) assert by_person[str(member.id)].expires_at is not None, ( "a regular member's discovered grant still gets a TTL" ) @pytest.mark.asyncio @pytest.mark.parametrize("role_name", ["cloud_admin", "global_admin"]) async def test_grants_sync_does_not_restamp_perpetual_rows( db_session: AsyncSession, role_name: str, ) -> None: """The nightly sync must not undo the sweep's normalization. ``_apply_discovered_grants`` backfills ``expires_at`` when it finds NULL on a discovered row. NULL is also how perpetual is SPELLED, so without a guard the sweep clears the TTL at 08:15 and the sync re-stamps it at 05:23 — forever. """ from app.domains.cmdb.lifecycle_service import _apply_discovered_grants from app.integrations.driver_base import CloudScopeRef, DiscoveredGrant admin = await _person( db_session, email=f"{role_name}-restamp@test.com", name="Admin Re", roles=(role_name,), ) owner = await _person(db_session, email="ow-restamp@test.com", name="Ow") integ = await _integration(db_session) scope = await _scope(db_session, integ, owner_person_id=str(owner.id)) # Already-normalized perpetual row + the owner's own perpetual row. admin_grant = await _grant( db_session, scope, admin, role_key="role/admin", expires_in_days=None, ) owner_grant = await _grant( db_session, scope, owner, role_key="role/owner", expires_in_days=None, ) # Capture PKs BEFORE ``expire_all()``: reading ``.id`` off an expired # instance triggers a synchronous lazy refresh, which raises # MissingGreenlet under the async session. admin_gid, owner_gid = admin_grant.id, owner_grant.id ref = CloudScopeRef( scope_type=scope.scope_type, external_id=scope.external_id, display_name=scope.display_name, ) now = datetime.now(UTC) await _apply_discovered_grants( db_session, scope, [ DiscoveredGrant( principal_email=admin.work_email, principal_type="user", role_key="role/admin", role_display_name="Admin", external_ref="x::a", scope=ref, granted_at=now, raw={}, ), DiscoveredGrant( principal_email=owner.work_email, principal_type="user", role_key="role/owner", role_display_name="Owner", external_ref="x::o", scope=ref, granted_at=now, raw={}, ), ], type("_D", (), {"provider": "aws", "owner_role_keys": frozenset()})(), str(uuid.uuid4()), now, ) await db_session.commit() db_session.expire_all() admin_after = await db_session.get(CmdbCloudAccessGrant, admin_gid) owner_after = await db_session.get(CmdbCloudAccessGrant, owner_gid) assert admin_after is not None and admin_after.expires_at is None, ( f"{role_name}'s NULL expiry must survive a sync" ) assert owner_after is not None and owner_after.expires_at is None, ( "scope owner's NULL expiry must survive a sync" ) @pytest.mark.asyncio async def test_discovery_ttl_is_spread_not_clumped() -> None: """A freshly onboarded account must not expire everyone at once. 164 grants on aws-account 385595570414 shared one ``expires_at`` to the SECOND, which produced a single-day reminder storm at T-14 and a revoke cliff at T+0, and self-service renewal re-clumped the cohort every quarter. """ from app.domains.cmdb.lifecycle_service import _discovery_ttl_expires_at now = datetime(2026, 7, 29, 8, 0, 0) values = { _discovery_ttl_expires_at( now=now, ttl_days=90, scope_id="s", person_id=f"p{i}", role_key="r", ) for i in range(120) } assert len(values) > 100, ( f"expected the cohort spread out, got {len(values)} distinct instants" ) # Never longer than policy — jitter subtracts only. assert max(values) <= now + timedelta(days=90) # Never below half the TTL, so the very next sweep can't revoke it. assert min(values) >= now + timedelta(days=45) # Deterministic: same inputs, same instant (no ``random``, no # process-salted ``hash()``). a = _discovery_ttl_expires_at( now=now, ttl_days=90, scope_id="s", person_id="p1", role_key="r", ) b = _discovery_ttl_expires_at( now=now, ttl_days=90, scope_id="s", person_id="p1", role_key="r", ) assert a == b @pytest.mark.asyncio async def test_owner_handover_restamps_outgoing_owner_grants( db_session: AsyncSession, ) -> None: """Handover must not silently revoke the outgoing owner. An exempt owner grant's expiry rots (140 production rows were months overdue). The moment ownership moves, the exemption stops applying and the next sweep would revoke instantly — every reminder window already in the past. 2026-08-18 — this test used to assert that ANY ``expires_at IS NULL`` grant kept its NULL through a handover ("NULL means perpetual"). That premise was the bug: the outgoing owner here holds no perpetual role, so their NULL came from the auto-owner backfill's ``force_perpetual`` — i.e. from *being the owner*, a reason that evaporates the instant ownership moves. Skipping those left 7 production rows permanently perpetual and invisible to both sweep passes. The exemption is decided by the ROLE now, so this test asserts both halves: the plain ex-owner's NULL grant is re-stamped, a real ``cloud_admin``'s is not. """ from app.domains.cmdb.lifecycle_service import CloudLifecycleService old_owner = await _person(db_session, email="old-ow@test.com", name="Old") new_owner = await _person(db_session, email="new-ow@test.com", name="New") integ = await _integration(db_session) scope = await _scope(db_session, integ, owner_person_id=str(old_owner.id)) stale = await _grant( db_session, scope, old_owner, expires_in_days=-120, notified_stage="t1", ) # NULL only because they were the owner — must be re-stamped. owner_perpetual = await _grant( db_session, scope, old_owner, role_key="role/perp", expires_in_days=None, ) # NULL because of the ROLE — must stay NULL through any handover. admin = await _person( db_session, email="handover-adm@test.com", name="Adm", roles=("cloud_admin",), ) admin_perpetual = await _grant( db_session, scope, admin, role_key="role/adm", expires_in_days=None, ) stale_gid, perp_gid = stale.id, owner_perpetual.id admin_gid = admin_perpetual.id svc = CloudLifecycleService(db_session) with _patch_lock(): await svc.assign_scope_owner( str(scope.id), str(new_owner.id), actor_user_id=new_owner.id, trace_id="t-handover", ) await db_session.commit() db_session.expire_all() after = await db_session.get(CmdbCloudAccessGrant, stale_gid) assert after is not None and after.expires_at is not None assert after.expires_at > datetime.now(UTC).replace(tzinfo=None), ( "outgoing owner's stale grant must be given a fresh runway, not " "left overdue for the next sweep to revoke without warning" ) assert after.last_notified_stage is None, ( "stale reminder marker must be cleared so the fresh cycle can nag" ) perp_after = await db_session.get(CmdbCloudAccessGrant, perp_gid) assert perp_after is not None and perp_after.expires_at is not None, ( "the outgoing owner holds no perpetual ROLE — their grant was NULL " "only because they were the owner, and that reason is gone. Leaving " "it NULL makes a permanent grant no sweep pass can ever see." ) assert perp_after.last_notified_stage is None admin_after = await db_session.get(CmdbCloudAccessGrant, admin_gid) assert admin_after is not None and admin_after.expires_at is None, ( "a cloud_admin is perpetual because of their role — a handover on " "some subscription must not newly subject them to expiry" ) # ═══════════════════════════════════════════════════════════ # Renewal must never shorten # ═══════════════════════════════════════════════════════════ @pytest.mark.asyncio async def test_renewal_never_shortens_expiry( db_session: AsyncSession, ) -> None: """nancy.wu regression. Three of her 365-day grants and one of yishuai.li's dropped from ~336 days remaining to 90 on 2026-07-15 — the 「全部续期」 kebab fires one renew per active grant and renew assigned ``now + 90`` flat. The user pressed a button labelled "renew" and lost 8 months. """ from app.domains.cmdb.lifecycle_service import CloudLifecycleService holder = await _person(db_session, email="long@test.com", name="Long") owner = await _person(db_session, email="ow-long@test.com", name="Ow") integ = await _integration(db_session) scope = await _scope(db_session, integ, owner_person_id=str(owner.id)) grant = await _grant(db_session, scope, holder, expires_in_days=336) before, gid = grant.expires_at, grant.id svc = CloudLifecycleService(db_session) with _patch_session_factory(db_session), _patch_lock(): await svc.renew_grant( str(gid), actor_user_id=holder.id, trace_id="t-noshorten", ) await db_session.commit() db_session.expire_all() after = await db_session.get(CmdbCloudAccessGrant, gid) assert after is not None # Second granularity: MySQL DATETIME drops sub-second precision. assert abs( (_naive(after.expires_at) - _naive(before)).total_seconds(), ) < 2, ( "a renewal is a FLOOR on remaining validity, never a ceiling — " f"{before} must not be cut back to {after.expires_at}" ) @pytest.mark.asyncio async def test_renewal_extends_when_shorter_than_policy( db_session: AsyncSession, ) -> None: """The normal case still extends.""" from app.domains.cmdb.lifecycle_service import CloudLifecycleService holder = await _person(db_session, email="short@test.com", name="Short") owner = await _person(db_session, email="ow-short@test.com", name="Ow") integ = await _integration(db_session) scope = await _scope(db_session, integ, owner_person_id=str(owner.id)) grant = await _grant(db_session, scope, holder, expires_in_days=3) before, gid = grant.expires_at, grant.id svc = CloudLifecycleService(db_session) with _patch_session_factory(db_session), _patch_lock(): await svc.renew_grant( str(gid), actor_user_id=holder.id, trace_id="t-extend", ) await db_session.commit() db_session.expire_all() after = await db_session.get(CmdbCloudAccessGrant, gid) assert after is not None assert _naive(after.expires_at) > _naive(before), ( f"3-day grant should extend to ~+90d, got {after.expires_at}" ) assert after.last_renewed_at is not None # ═══════════════════════════════════════════════════════════ # Cluster 1b — expiry notifications # ═══════════════════════════════════════════════════════════ @pytest.mark.asyncio async def test_expiry_reminder_is_one_card_per_person( db_session: AsyncSession, ) -> None: """Aggregate per PERSON, never per grant. The old per-grant fan-out sent 861 cards to 198 people in one cron run (2026-07-15), 63 of them to a single person in a single day, over both Feishu and email. ``sweep_inactive_grants`` had already learned this lesson and grouped; the expiry sweep never did. """ from unittest.mock import AsyncMock from app.domains.cmdb.lifecycle_service import sweep_expiring_grants holder = await _person(db_session, email="agg@test.com", name="Agg") owner = await _person(db_session, email="ow-agg@test.com", name="Ow") integ = await _integration(db_session) scope_a = await _scope( db_session, integ, owner_person_id=str(owner.id), external_id="acct-a", ) scope_b = await _scope( db_session, integ, owner_person_id=str(owner.id), external_id="acct-b", ) # 5 grants across 2 accounts, all landing in the T-14 bucket. # 13.5 days, not 13 or 14: the window is # ``now + 13 < expires_at <= now + 14``, so 13 is outside on the low # side and 14 can round outside on the high side. for i in range(3): await _grant( db_session, scope_a, holder, role_key=f"role/a{i}", expires_in_days=13.5, ) for i in range(2): await _grant( db_session, scope_b, holder, role_key=f"role/b{i}", expires_in_days=13.5, ) sent: list[object] = [] async def _capture(env, **_kw): sent.append(env) with patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ): summary = await sweep_expiring_grants(db_session) await db_session.commit() assert summary["notified"] == 1, ( f"5 expiring grants must produce ONE card, not 5: {summary}" ) assert len(sent) == 1 body = sent[0].body_text # Every affected role is named, so aggregating loses no information. for role in ("role/a0", "role/a1", "role/a2", "role/b0", "role/b1"): assert role in body, f"{role} missing from the aggregated card" assert "acct-a" in body and "acct-b" in body # And it points at the approval-based flow, not the removed one-click # renew that no longer exists. assert "续期申请" in body @pytest.mark.asyncio async def test_recycled_access_notifies_the_holder( db_session: AsyncSession, ) -> None: """Recycling used to be completely silent. Every ``notification_outbox`` row related to a cloud grant was a 「云访问即将到期」 reminder — not one was a recycling notice. Holders discovered the loss by being denied in the cloud console. """ from unittest.mock import AsyncMock from app.domains.cmdb.lifecycle_service import sweep_expiring_grants holder = await _person(db_session, email="rev@test.com", name="Rev") owner = await _person(db_session, email="ow-rev@test.com", name="Ow") integ = await _integration(db_session) scope = await _scope(db_session, integ, owner_person_id=str(owner.id)) await _grant( db_session, scope, holder, role_key="role/gone", expires_in_days=-1, ) sent: list[object] = [] async def _capture(env, **_kw): sent.append(env) with patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ): summary = await sweep_expiring_grants(db_session) await db_session.commit() assert summary["expired_revoked"] >= 1, summary assert summary["revoke_notified"] >= 1, ( f"the holder must be told their access was recycled: {summary}" ) notices = [e for e in sent if "已回收" in e.title] assert len(notices) == 1 # And it must say how to get it back, since revival is possible for a # bounded window and nobody would guess that. assert str(RENEWAL_REVIVAL_MAX_AGE_DAYS) in notices[0].body_text # ═══════════════════════════════════════════════════════════ # Cluster 2 — renewal is an owner-approved request # ═══════════════════════════════════════════════════════════ def _patch_notify(): """Capture notification envelopes instead of delivering them.""" from unittest.mock import AsyncMock sent: list[object] = [] # ``patch(..., new=AsyncMock(...))`` replaces the class attribute with a # plain mock, which is not a descriptor — so ``self`` is NOT passed # through and the callback sees only the envelope. async def _capture(env, **_kw): sent.append(env) return sent, patch( "app.notifications.service.NotificationService.dispatch", new=AsyncMock(side_effect=_capture), ) async def _renewal_fixture( db_session: AsyncSession, *, ttl_days: int = 90, holder_status: str = "active", environment: str = "dev", grant_status: str = GRANT_STATUS_ACTIVE, revoked_days_ago: int | None = None, n_roles: int = 1, ): holder = await _person( db_session, email=f"h-{uuid.uuid4().hex[:6]}@test.com", name="Holder", status=holder_status, ) owner = await _person( db_session, email=f"o-{uuid.uuid4().hex[:6]}@test.com", name="Owner", ) integ = await _integration(db_session) scope = await _scope( db_session, integ, owner_person_id=str(owner.id), environment=environment, ) grants = [ await _grant( db_session, scope, holder, role_key=f"role/r{i}", status=grant_status, expires_in_days=(-1 if grant_status == GRANT_STATUS_EXPIRED else 5), revoked_days_ago=revoked_days_ago, ) for i in range(n_roles) ] return holder, owner, scope, grants @pytest.mark.asyncio async def test_submit_covers_every_role_as_one_group( db_session: AsyncSession, ) -> None: """One submission per SUBSCRIPTION, not per role. The pre-change flow was per-grant: a holder with 8 roles on one account got 8 reminder cards, had to click 8 times, and would have appeared 8 times in the owner's queue. """ from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture( db_session, n_roles=3, ) sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch: result = await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason="继续做压测", requested_by_person_id=holder.id, trace_id="t-submit", ) assert result["grant_count"] == 3 rows = (await db_session.execute( select(CmdbCloudGrantRenewalRequest).where( CmdbCloudGrantRenewalRequest.scope_id == str(scope.id), ) )).scalars().all() assert len(rows) == 3 assert len({str(r.request_group_id) for r in rows}) == 1, ( "all roles of one submission must share one request_group_id" ) assert all(r.status == RENEWAL_STATUS_PENDING_OWNER for r in rows) assert all(r.open_for_grant_id is not None for r in rows) # ONE card to the owner, not one per role — the expiry reminder's # per-grant fan-out (861 cards in a single run) must not be repeated. assert len(sent) == 1, f"expected 1 approver card, got {len(sent)}" @pytest.mark.asyncio async def test_duplicate_submission_is_refused_by_the_database( db_session: AsyncSession, ) -> None: """A double-clicked card must not open two requests for one grant.""" from app.core.exceptions import ConflictError from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch: await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason="一次", requested_by_person_id=holder.id, trace_id="t1", ) with pytest.raises(ConflictError): await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason="重复点击", requested_by_person_id=holder.id, trace_id="t2", ) @pytest.mark.asyncio @pytest.mark.parametrize( ("kind", "expected_reason"), [ ("perpetual", "perpetual_holder"), ("owner", "scope_owner_exempt"), ("prod", "prod_unmanaged"), ("departed", "holder_departed"), ("old_revival", "revival_window_expired"), ], ) async def test_ineligible_grants_are_refused_with_a_reason( db_session: AsyncSession, kind: str, expected_reason: str, ) -> None: """Every refusal carries a machine-readable reason the UI renders. The audit found the opposite pattern across the cloud surface — buttons enabled that then 409'd, chips claiming a state the backend disagreed with. Eligibility is computed server-side, once. """ from app.domains.cmdb.renewal_service import check_renewal_eligibility if kind == "perpetual": person = await _person( db_session, email="ca-elig@test.com", name="CA", roles=("cloud_admin",), ) integ = await _integration(db_session) scope = await _scope(db_session, integ) grant = await _grant(db_session, scope, person) elif kind == "owner": person = await _person(db_session, email="ow-elig@test.com", name="O") integ = await _integration(db_session) scope = await _scope( db_session, integ, owner_person_id=str(person.id), ) grant = await _grant(db_session, scope, person) elif kind == "prod": person = await _person(db_session, email="pr-elig@test.com", name="P") integ = await _integration(db_session) scope = await _scope(db_session, integ, environment="prod") grant = await _grant(db_session, scope, person) elif kind == "departed": person = await _person( db_session, email="lv-elig@test.com", name="L", status="leaver_pending", ) integ = await _integration(db_session) scope = await _scope(db_session, integ) grant = await _grant(db_session, scope, person) else: # old_revival person = await _person(db_session, email="od-elig@test.com", name="D") integ = await _integration(db_session) scope = await _scope(db_session, integ) grant = await _grant( db_session, scope, person, status=GRANT_STATUS_EXPIRED, expires_in_days=-(RENEWAL_REVIVAL_MAX_AGE_DAYS + 10), revoked_days_ago=RENEWAL_REVIVAL_MAX_AGE_DAYS + 10, ) verdict = await check_renewal_eligibility( db_session, grant=grant, scope=scope, holder=person, ) assert verdict.eligible is False assert verdict.reason == expected_reason @pytest.mark.asyncio async def test_owner_approval_applies_the_renewal( db_session: AsyncSession, ) -> None: """The happy path: owner approves, expiry moves, request closes.""" from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) # Snapshot every identifier BEFORE the service starts committing: # a commit expires the instances, and reading an attribute off an # expired one triggers a synchronous refresh (MissingGreenlet). gid = grants[0].id before = grants[0].expires_at scope_id, holder_id, owner_id = str(scope.id), str(holder.id), owner.id sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch, _patch_lock(), _patch_lock( "app.domains.cmdb.renewal_service", ), _patch_session_factory(db_session): submitted = await svc.submit_renewal_request( scope_id=scope_id, person_id=holder_id, grant_ids=None, ttl_days=90, reason="还要用", requested_by_person_id=holder.id, trace_id="t-sub", ) outcome = await svc.decide_renewal_group( request_group_id=submitted["request_group_id"], approve=True, approver_id=owner_id, approver_role_names=["employee"], trace_id="t-dec", ) await db_session.commit() db_session.expire_all() assert outcome["status"] == RENEWAL_STATUS_APPROVED assert outcome["applied"] is True grant_after = await db_session.get(CmdbCloudAccessGrant, gid) assert grant_after is not None assert _naive(grant_after.expires_at) > _naive(before) req = (await db_session.execute( select(CmdbCloudGrantRenewalRequest).where( CmdbCloudGrantRenewalRequest.scope_id == scope_id, ) )).scalars().one() assert req.status == RENEWAL_STATUS_APPROVED assert req.open_for_grant_id is None, ( "closing must free the UNIQUE slot or the grant becomes " "permanently un-renewable" ) assert req.applied_expires_at is not None assert req.owner_decided_by_person_id == str(owner_id) @pytest.mark.asyncio async def test_holder_cannot_approve_their_own_renewal( db_session: AsyncSession, ) -> None: """Four eyes. Self-service renewal is exactly what was removed — letting the holder approve would reinstate it under a new name, even when they own the scope or hold cloud_admin. """ from app.core.exceptions import ConflictError from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch, _patch_lock("app.domains.cmdb.renewal_service"): submitted = await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason="自批试试", requested_by_person_id=holder.id, trace_id="t-sub", ) with pytest.raises(ConflictError, match="不能审批自己"): await svc.decide_renewal_group( request_group_id=submitted["request_group_id"], approve=True, approver_id=holder.id, # even with cloud-admin authority approver_role_names=["cloud_admin"], trace_id="t-self", ) @pytest.mark.asyncio async def test_long_ttl_applies_on_the_owners_single_signature( db_session: AsyncSession, ) -> None: """A TTL over the old 90-day tier is decided by the owner alone. Until 2026-08-26 anything above ``GRANT_TTL_DEFAULT_DAYS`` parked in ``pending_admin`` waiting for a cloud admin to countersign. That tier was removed: the subscription owner's approval applies the renewal immediately at any length the hard cap allows. """ from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) gid = grants[0].id long_ttl = GRANT_TTL_HARD_CAP_DAYS sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch, _patch_lock(), _patch_lock( "app.domains.cmdb.renewal_service", ), _patch_session_factory(db_session): submitted = await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=long_ttl, reason="长期项目", requested_by_person_id=holder.id, trace_id="t-sub", ) decided = await svc.decide_renewal_group( request_group_id=submitted["request_group_id"], approve=True, approver_id=owner.id, approver_role_names=["employee"], trace_id="t-s1", ) assert decided["status"] == RENEWAL_STATUS_APPROVED assert decided["applied"] is True, ( "the owner is the only signature required — no escalation hop" ) db_session.expire_all() row = (await db_session.execute( select(CmdbCloudGrantRenewalRequest).where( CmdbCloudGrantRenewalRequest.request_group_id == submitted["request_group_id"], ) )).scalars().first() assert row is not None assert row.status == RENEWAL_STATUS_APPROVED assert row.admin_decided_by_person_id is None, ( "no second signature is collected any more" ) grant = await db_session.get(CmdbCloudAccessGrant, gid) assert grant is not None and grant.expires_at is not None applied_days = ( grant.expires_at.replace(tzinfo=None) - datetime.now(UTC).replace( tzinfo=None, ) ).days assert applied_days >= long_ttl - 2, ( f"approving {long_ttl} days must move expiry ~{long_ttl} days out, " f"got {applied_days}" ) @pytest.mark.asyncio async def test_cloud_admin_approving_a_long_ttl_applies_in_one_click( db_session: AsyncSession, ) -> None: """A cloud admin deciding an over-90 request no longer escalates. The removed rule split their click into "stage 1" and routed the row to a second cloud admin. With one signature required, their decision is the decision. """ from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) admin = await _person( db_session, email="adm-both@test.com", name="Admin Both", roles=("cloud_admin",), ) sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch, _patch_lock(), _patch_lock( "app.domains.cmdb.renewal_service", ), _patch_session_factory(db_session): submitted = await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=GRANT_TTL_DEFAULT_DAYS + 1, reason="超一天", requested_by_person_id=holder.id, trace_id="t-sub", ) decided = await svc.decide_renewal_group( request_group_id=submitted["request_group_id"], approve=True, approver_id=admin.id, approver_role_names=["cloud_admin"], trace_id="t-s1", ) assert decided["status"] == RENEWAL_STATUS_APPROVED assert decided["applied"] is True @pytest.mark.asyncio async def test_rejection_closes_group_and_frees_the_slot( db_session: AsyncSession, ) -> None: """Rejected → grant expires on its original schedule, and the holder can submit again (the UNIQUE slot is released).""" from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) gid = grants[0].id before = grants[0].expires_at sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch, _patch_lock("app.domains.cmdb.renewal_service"): submitted = await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason="试试", requested_by_person_id=holder.id, trace_id="t-sub", ) out = await svc.decide_renewal_group( request_group_id=submitted["request_group_id"], approve=False, approver_id=owner.id, approver_role_names=["employee"], trace_id="t-rej", note="项目已结束", ) assert out["status"] == RENEWAL_STATUS_REJECTED # Slot freed → resubmission allowed. await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason="沟通后重提", requested_by_person_id=holder.id, trace_id="t-sub2", ) await db_session.commit() db_session.expire_all() grant_after = await db_session.get(CmdbCloudAccessGrant, gid) assert grant_after is not None assert abs( (_naive(grant_after.expires_at) - _naive(before)).total_seconds(), ) < 2, "a rejection must not touch the grant's expiry" @pytest.mark.asyncio async def test_decision_refuses_a_holder_who_left_meanwhile( db_session: AsyncSession, ) -> None: """Up to 14 days pass between submit and decide — long enough to resign. Approving then would extend a leaver's cloud access and race the offboarding sweep.""" from app.core.exceptions import ConflictError from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) scope_id, holder_id, owner_id = str(scope.id), str(holder.id), owner.id sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch, _patch_lock("app.domains.cmdb.renewal_service"): submitted = await svc.submit_renewal_request( scope_id=scope_id, person_id=holder_id, grant_ids=None, ttl_days=90, reason="还要用", requested_by_person_id=holder.id, trace_id="t-sub", ) holder.employment_status = "leaver_pending" await db_session.commit() with pytest.raises(ConflictError, match="离职"): await svc.decide_renewal_group( request_group_id=submitted["request_group_id"], approve=True, approver_id=owner_id, approver_role_names=["employee"], trace_id="t-dec", ) db_session.expire_all() rows = (await db_session.execute( select(CmdbCloudGrantRenewalRequest).where( CmdbCloudGrantRenewalRequest.scope_id == scope_id, ) )).scalars().all() assert all(r.status == RENEWAL_STATUS_CANCELLED for r in rows) @pytest.mark.asyncio async def test_lapse_sweep_closes_undecided_requests( db_session: AsyncSession, ) -> None: """An undecided request must not stay open forever: it occupies the grant's only open slot and stays approvable months later.""" from app.domains.cmdb.renewal_service import ( CloudRenewalService, lapse_stale_renewal_requests, ) holder, owner, scope, grants = await _renewal_fixture(db_session) sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch, _patch_lock("app.domains.cmdb.renewal_service"): submitted = await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason="没人管", requested_by_person_id=holder.id, trace_id="t-sub", ) # Fast-forward past the lapse deadline. result = await lapse_stale_renewal_requests( db_session, now=datetime.now(UTC) + timedelta(days=RENEWAL_REQUEST_LAPSE_DAYS + 1), ) assert result["lapsed"] >= 1, result db_session.expire_all() rows = (await db_session.execute( select(CmdbCloudGrantRenewalRequest).where( CmdbCloudGrantRenewalRequest.request_group_id == submitted["request_group_id"], ) )).scalars().all() assert all(r.status == RENEWAL_STATUS_LAPSED for r in rows) assert all(r.open_for_grant_id is None for r in rows) @pytest.mark.asyncio async def test_unauthorized_decider_changes_nothing( db_session: AsyncSession, ) -> None: """Authority is checked BEFORE anything that writes or reveals. Found in review: the decide path re-validated the holder's employment first, and that branch cancels the request + writes an audit row attributed to the caller. Ordered that way, anyone who could reach the endpoint could force a state change on a scope they have no authority over, and read "the requester has left the company" out of the 409. """ from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) stranger = await _person( db_session, email="stranger@test.com", name="Stranger", ) sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch, _patch_lock("app.domains.cmdb.renewal_service"): submitted = await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason="等审批", requested_by_person_id=holder.id, trace_id="t-sub", ) # Make the holder departed so the (previously earlier) cancel # branch WOULD fire if it were still reached first. holder.employment_status = "leaver_pending" await db_session.commit() with pytest.raises(AppExceptionLike) as exc: await svc.decide_renewal_group( request_group_id=submitted["request_group_id"], approve=True, approver_id=stranger.id, approver_role_names=["employee"], trace_id="t-authz", ) assert exc.value.status_code == 403, ( "an unrelated employee must be refused on authority, not " "reach the holder-employment branch" ) db_session.expire_all() rows = (await db_session.execute( select(CmdbCloudGrantRenewalRequest).where( CmdbCloudGrantRenewalRequest.request_group_id == submitted["request_group_id"], ) )).scalars().all() assert all(r.status == RENEWAL_STATUS_PENDING_OWNER for r in rows), ( "an unauthorized call must leave the request untouched" ) assert all(r.open_for_grant_id is not None for r in rows) @pytest.mark.asyncio async def test_owner_is_nudged_once_before_a_request_lapses( db_session: AsyncSession, ) -> None: """The likeliest failure of the approval flow is silence. The owner misses the submission card, nothing else mentions it, the request lapses at T+14 and access is recycled because nobody said no — they just never said anything. One nudge near the deadline turns that into a decision. It must fire exactly once, or it becomes the same daily spam the per-grant reminders were. """ from app.domains.cmdb.renewal_service import ( RENEWAL_OWNER_NUDGE_LEAD_DAYS, CloudRenewalService, nudge_expiring_renewal_requests, ) holder, owner, scope, grants = await _renewal_fixture(db_session) sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch, _patch_lock("app.domains.cmdb.renewal_service"): submitted = await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason="等审批", requested_by_person_id=holder.id, trace_id="t-sub", ) sent.clear() # drop the submission card # Anchor on the row's own ``lapse_at`` rather than computing # ``now + (14 - 3)``: MySQL DATETIME rounds to the second, so a # value derived from the test's clock can land a few hundred # microseconds outside the window and silently match nothing. lapse_at = submitted["lapse_at"] if lapse_at.tzinfo is None: lapse_at = lapse_at.replace(tzinfo=UTC) at = lapse_at - timedelta(days=1) assert RENEWAL_OWNER_NUDGE_LEAD_DAYS >= 1 # window must contain it first = await nudge_expiring_renewal_requests(db_session, now=at) assert first["nudged"] >= 1, first assert len(sent) == 1 assert "即将超期" in sent[0].title # Running again the next day must NOT re-send. second = await nudge_expiring_renewal_requests( db_session, now=at + timedelta(days=1), ) assert second["nudged"] == 0, ( "the nudge is once-per-request; re-sending daily recreates the " "reminder spam this rework removed" ) assert len(sent) == 1 db_session.expire_all() rows = (await db_session.execute( select(CmdbCloudGrantRenewalRequest).where( CmdbCloudGrantRenewalRequest.request_group_id == submitted["request_group_id"], ) )).scalars().all() assert all(r.owner_reminded_at is not None for r in rows) @pytest.mark.asyncio async def test_hard_revoke_cascades_but_expiry_does_not( db_session: AsyncSession, ) -> None: """A deliberate revoke must close the open request — otherwise the owner clicking 「批准」 next week resurrects access somebody closed on purpose. Expiry is the opposite: the pending request IS the intended revival path and must survive. """ from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch, _patch_lock("app.domains.cmdb.renewal_service"): submitted = await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason="待审中", requested_by_person_id=holder.id, trace_id="t-sub", ) # Expiry-driven closure must NOT cascade. n = await svc.close_open_requests_for_grant( grant_id=str(grants[0].id), reason=GRANT_REVOKE_REASON_EXPIRED, trace_id="t-exp", ) assert n == 1, ( "close_open_requests_for_grant itself is reason-agnostic; the " "expiry exemption lives at the revoke call site" ) db_session.expire_all() rows = (await db_session.execute( select(CmdbCloudGrantRenewalRequest).where( CmdbCloudGrantRenewalRequest.request_group_id == submitted["request_group_id"], ) )).scalars().all() assert all(r.status == RENEWAL_STATUS_CANCELLED for r in rows) @pytest.mark.asyncio async def test_reason_is_mandatory(db_session: AsyncSession) -> None: """The approver needs something to judge against.""" from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) svc = CloudRenewalService(db_session) with pytest.raises(AppExceptionLike): await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason=" ", requested_by_person_id=holder.id, trace_id="t-noreason", ) @pytest.mark.asyncio async def test_cannot_submit_on_behalf_without_cloud_admin( db_session: AsyncSession, ) -> None: """Otherwise any employee could file renewals in a colleague's name.""" from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) other = await _person(db_session, email="other@test.com", name="Other") svc = CloudRenewalService(db_session) with pytest.raises(AppExceptionLike): await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason="代提", requested_by_person_id=other.id, actor_role_names=["employee"], trace_id="t-behalf", ) # ═══════════════════════════════════════════════════════════ # Cluster 3 — the approver has to be able to FIND the request # # Reported 2026-07-31 by a scope owner (a plain employee, as owners # usually are): the approver card arrived, the click landed on the cloud # subscription page, and there was no approval queue, no to-do entry, and # nothing highlighted. The decision surface existed only on the # global_admin scope page. These tests pin the three server-side halves of # that fix — the deep link naming the submission, the cross-scope approver # inbox naming the subscription, and the inbox not showing work the caller # is not allowed to complete. # ═══════════════════════════════════════════════════════════ @pytest.mark.asyncio async def test_approver_card_deep_links_to_the_exact_submission( db_session: AsyncSession, ) -> None: """``?renewals=1&group=`` — queue open, that row flashed. Without ``group`` the card could only say "the queue", and with the queue itself missing on the owner surface the click was a dead end. """ from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch: submitted = await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason="继续压测", requested_by_person_id=holder.id, trace_id="t-deeplink", ) assert len(sent) == 1 env = sent[0] group_id = submitted["request_group_id"] expected = f"/cloud/scopes/{scope.id}?renewals=1&group={group_id}" assert env.url == expected, env.url # The Feishu button must carry the same target — a card whose body link # and button disagree sends half the approvers to the wrong place. assert [a.url for a in (env.card_actions or [])] == [expected] assert env.related_resource_id == group_id @pytest.mark.asyncio async def test_approver_inbox_names_the_subscription( db_session: AsyncSession, ) -> None: """The cross-scope inbox row must say WHICH subscription. ``/my/todos`` renders one table across every scope the caller owns, so a row without a scope label is unreadable — and re-fetching the scope per row is exactly the N+1 the payload should prevent. """ from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch: await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=90, reason="继续压测", requested_by_person_id=holder.id, trace_id="t-inbox", ) rows = await svc.list_pending_for_approver( str(owner.id), is_cloud_admin=False, ) assert len(rows) == 1 assert rows[0]["scope_id"] == str(scope.id) assert rows[0]["scope_label"] == (scope.display_name or scope.external_id) assert rows[0]["scope_external_id"] == scope.external_id @pytest.mark.asyncio async def test_owner_inbox_clears_once_they_decide_a_long_ttl( db_session: AsyncSession, ) -> None: """The owner's to-do is completable in one step at any TTL. Between 2026-07-31 and 2026-08-26 an over-90 request left the owner's inbox on their approval and moved to the cloud-admin queue for a second signature. Now their decision closes it for everyone: the row is applied, so nobody is left holding an open to-do. """ from app.domains.cmdb.renewal_service import CloudRenewalService holder, owner, scope, grants = await _renewal_fixture(db_session) admin = await _person( db_session, email="adm-inbox@test.com", name="Admin Inbox", roles=("cloud_admin",), ) sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch, _patch_lock(), _patch_lock( "app.domains.cmdb.renewal_service", ), _patch_session_factory(db_session): submitted = await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=GRANT_TTL_DEFAULT_DAYS + 30, reason="长期项目", requested_by_person_id=holder.id, trace_id="t-sub", ) assert len(await svc.list_pending_for_approver( str(owner.id), is_cloud_admin=False, )) == 1 decided = await svc.decide_renewal_group( request_group_id=submitted["request_group_id"], approve=True, approver_id=owner.id, approver_role_names=["employee"], trace_id="t-s1", ) assert decided["applied"] is True assert await svc.list_pending_for_approver( str(owner.id), is_cloud_admin=False, ) == [] assert await svc.list_pending_for_approver( str(admin.id), is_cloud_admin=True, ) == [], "no second signature is outstanding for anyone" assert await svc.list_open_for_scope(str(scope.id)) == [] @pytest.mark.asyncio async def test_mobile_inbox_matches_the_approver_list_on_a_long_ttl( db_session: AsyncSession, ) -> None: """The mobile badge and the approver list must agree. They are two implementations of one authority rule; when they drift the badge shows a count the page then contradicts. """ from app.domains.cmdb.renewal_service import CloudRenewalService from app.mobile.action_inbox import _provider_cloud_renewal_approval holder, owner, scope, grants = await _renewal_fixture(db_session) sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch, _patch_lock(), _patch_lock( "app.domains.cmdb.renewal_service", ), _patch_session_factory(db_session): submitted = await svc.submit_renewal_request( scope_id=str(scope.id), person_id=str(holder.id), grant_ids=None, ttl_days=GRANT_TTL_DEFAULT_DAYS + 30, reason="长期项目", requested_by_person_id=holder.id, trace_id="t-sub", ) assert len(await _provider_cloud_renewal_approval( db_session, owner.id, ["employee"], )) == len(await svc.list_pending_for_approver( str(owner.id), is_cloud_admin=False, )) == 1 await svc.decide_renewal_group( request_group_id=submitted["request_group_id"], approve=True, approver_id=owner.id, approver_role_names=["employee"], trace_id="t-s1", ) assert await _provider_cloud_renewal_approval( db_session, owner.id, ["employee"], ) == [] assert await _provider_cloud_renewal_approval( db_session, owner.id, ["cloud_admin"], ) == [] # ── 回执:审批人手上那张「去审批」卡片,决策之后必须变 ────────────── @pytest.mark.asyncio async def test_deciding_settles_the_approver_card( db_session: AsyncSession, ) -> None: """用户的原话:「这个审批过了,还是去审批」。 这条测的不是回执**原语**(那在 test_settlement_2026_09_07.py),而是这条 流程真的**接线**了 —— 原语再好,没人调用就等于没有。所以断言的是「决策 发生之后,那条通知行被结案成 approved」。 """ from app.domains.cmdb.renewal_service import CloudRenewalService from app.notifications import card_design from app.notifications.models import NotificationOutbox holder, owner, scope, grants = await _renewal_fixture(db_session) scope_id, holder_id, owner_id = str(scope.id), str(holder.id), owner.id sent, notify_patch = _patch_notify() svc = CloudRenewalService(db_session) with notify_patch, _patch_lock(), _patch_lock( "app.domains.cmdb.renewal_service", ), _patch_session_factory(db_session): submitted = await svc.submit_renewal_request( scope_id=scope_id, person_id=holder_id, grant_ids=None, ttl_days=90, reason="还要用", requested_by_person_id=holder.id, trace_id="t-sub", ) group_id = submitted["request_group_id"] # 审批人那张卡:`_notify_approvers` 被替身拦掉了(它要打飞书),所以 # 这里按它真实的形状手写一行 —— 关键是 related_resource_type/_id, # 那是 settle 的连接键。 db_session.add(NotificationOutbox( recipient_person_id=owner.id, recipient_email=owner.work_email, template_key="approval.pending", title="续期待审批", body_text="请审批。", url=f"/cloud/scopes/{scope_id}?renewals=1&group={group_id}", channels=["feishu"], status="delivered", delivered_channels=["feishu"], related_resource_type="cmdb_cloud_grant_renewal_request", related_resource_id=str(group_id), )) await db_session.flush() await svc.decide_renewal_group( request_group_id=group_id, approve=True, approver_id=owner_id, approver_role_names=["employee"], trace_id="t-dec", ) await db_session.commit() db_session.expire_all() row = (await db_session.execute( select(NotificationOutbox).where( NotificationOutbox.related_resource_id == str(group_id), ) )).scalars().first() assert row is not None assert row.settled_at is not None, "决策发生了,但审批卡没有被结案" assert row.settled_outcome == card_design.STATUS_APPROVED assert str(row.settled_by_person_id) == str(owner_id)