"""Tests for the API Key shared-access roster (migration 200). End-to-end coverage of the four behaviour contracts the feature guarantees: 1. Owner can add / remove a teammate, with idempotency and "active employee" validation. 2. A member can read the same surfaces the Owner reads (spend / detail / reveal / list visibility); a non-owner non-member non-admin cannot. 3. Owner transfer auto-promotes the previous Owner to a member, and removes the new Owner from the roster if they happened to already be on it. 4. `LeaverBlockGuard.flip_to_left` sweeps every roster row that references the leaver, audited per row. Real-DB tests; no service-layer mocks. Mirrors the `test_assignment_spend.py` fixture-as-helper style so each test reads end-to-end without bouncing into shared fixtures. """ from __future__ import annotations import uuid from decimal import Decimal from types import SimpleNamespace import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException, ForbiddenError from app.domains.apikey.models import ( ApiKeyAssignment, ApiKeyAssignmentMember, ApiKeyProvider, ) from app.domains.apikey.service import ApiKeyService from app.identity.person_models import IdentityPerson # ── Helpers ────────────────────────────────────────────────────────── async def _person( db: AsyncSession, *, email: str, status: str = "active", ) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", display_name=email.split("@")[0], work_email=email, employment_status=status, ) db.add(p) await db.commit() return p async def _provider(db: AsyncSession) -> ApiKeyProvider: p = ApiKeyProvider( id=uuid.uuid4(), name="OpenAI", description="", is_active=True, sort_order=0, ) db.add(p) await db.commit() return p async def _assignment( db: AsyncSession, *, provider: ApiKeyProvider, owner: IdentityPerson, status: str = "active", ) -> ApiKeyAssignment: a = ApiKeyAssignment( id=uuid.uuid4(), provider_id=provider.id, assigned_to=owner.id, status=status, request_number=f"AK-{uuid.uuid4().hex[:8].upper()}", project_name="Team SDK", reason="team-shared usage", monthly_budget_usd=Decimal("500.00"), ) db.add(a) await db.commit() return a def _auth( person: IdentityPerson, *, admin: bool = False, ) -> SimpleNamespace: return SimpleNamespace( id=person.id, display_name=person.display_name, work_email=person.work_email, # Mirror the real IdentityPerson columns the response serializer # now reads off the current-user shortcut (2026-06-18 leaver # display: assignee_employee_no / _employment_status). employee_no=person.employee_no, employment_status=person.employment_status, role_names=["asset_admin"] if admin else ["employee"], ) async def _member_rows( db: AsyncSession, assignment_id: uuid.UUID, ) -> list[ApiKeyAssignmentMember]: return list( ( await db.execute( select(ApiKeyAssignmentMember).where( ApiKeyAssignmentMember.assignment_id == assignment_id, ) ) ).scalars().all() ) # ── Add / remove (Owner-only authz, idempotent, active-only) ───────── @pytest.mark.asyncio async def test_owner_can_add_and_remove_member(db_session: AsyncSession): """Owner adds a teammate → row exists; remove → row gone.""" provider = await _provider(db_session) owner = await _person(db_session, email="alice@x.com") member = await _person(db_session, email="bob@x.com") assignment = await _assignment( db_session, provider=provider, owner=owner, ) svc = ApiKeyService(db_session) out = await svc.add_member( assignment.id, person_id=member.id, current_user=_auth(owner), trace_id="t1", ip_address=None, user_agent=None, ) assert out.person_id == member.id assert out.display_name == "bob" rows = await _member_rows(db_session, assignment.id) assert len(rows) == 1 assert rows[0].person_id == member.id assert rows[0].added_by == owner.id await svc.remove_member( assignment.id, person_id=member.id, current_user=_auth(owner), trace_id="t2", ip_address=None, user_agent=None, ) rows = await _member_rows(db_session, assignment.id) assert rows == [] @pytest.mark.asyncio async def test_owner_add_member_is_idempotent(db_session: AsyncSession): """Re-adding the same person is a no-op (same row returned, no duplicate insert).""" provider = await _provider(db_session) owner = await _person(db_session, email="alice@x.com") member = await _person(db_session, email="bob@x.com") assignment = await _assignment( db_session, provider=provider, owner=owner, ) svc = ApiKeyService(db_session) first = await svc.add_member( assignment.id, person_id=member.id, current_user=_auth(owner), trace_id="t1", ip_address=None, user_agent=None, ) second = await svc.add_member( assignment.id, person_id=member.id, current_user=_auth(owner), trace_id="t2", ip_address=None, user_agent=None, ) assert first.id == second.id rows = await _member_rows(db_session, assignment.id) assert len(rows) == 1 @pytest.mark.asyncio async def test_owner_cannot_add_leaver(db_session: AsyncSession): """Adding a person who has flipped to ``left`` is rejected — the Owner shouldn't be able to silently grant access to a former employee.""" provider = await _provider(db_session) owner = await _person(db_session, email="alice@x.com") leaver = await _person( db_session, email="exemployee@x.com", status="left", ) assignment = await _assignment( db_session, provider=provider, owner=owner, ) with pytest.raises(AppException) as exc: await ApiKeyService(db_session).add_member( assignment.id, person_id=leaver.id, current_user=_auth(owner), trace_id="t", ip_address=None, user_agent=None, ) assert exc.value.code == "MEMBER_NOT_ACTIVE" @pytest.mark.asyncio async def test_owner_cannot_add_self(db_session: AsyncSession): """Self-add is rejected — the applicant is already the Owner; a self-add row would surface as a duplicate in the panel.""" provider = await _provider(db_session) owner = await _person(db_session, email="alice@x.com") assignment = await _assignment( db_session, provider=provider, owner=owner, ) with pytest.raises(AppException) as exc: await ApiKeyService(db_session).add_member( assignment.id, person_id=owner.id, current_user=_auth(owner), trace_id="t", ip_address=None, user_agent=None, ) assert exc.value.code == "CANNOT_ADD_SELF" @pytest.mark.asyncio async def test_non_owner_cannot_manage_members(db_session: AsyncSession): """Even admins do NOT have add_member / remove_member privileges — membership is an Owner concern. Admins manage keys through the rebind / revoke / IAM tools instead.""" provider = await _provider(db_session) owner = await _person(db_session, email="alice@x.com") admin = await _person(db_session, email="ada@x.com") candidate = await _person(db_session, email="bob@x.com") assignment = await _assignment( db_session, provider=provider, owner=owner, ) with pytest.raises(ForbiddenError): await ApiKeyService(db_session).add_member( assignment.id, person_id=candidate.id, current_user=_auth(admin, admin=True), trace_id="t", ip_address=None, user_agent=None, ) @pytest.mark.asyncio async def test_remove_nonexistent_member_is_noop( db_session: AsyncSession, ): """Removing a person who isn't on the roster returns successfully with no side effect — the Owner's bulk-remove UX shouldn't have to special-case "already gone".""" provider = await _provider(db_session) owner = await _person(db_session, email="alice@x.com") stranger = await _person(db_session, email="bob@x.com") assignment = await _assignment( db_session, provider=provider, owner=owner, ) # No throw, no side effect. await ApiKeyService(db_session).remove_member( assignment.id, person_id=stranger.id, current_user=_auth(owner), trace_id="t", ip_address=None, user_agent=None, ) assert await _member_rows(db_session, assignment.id) == [] # ── Read-side authz: members see, strangers don't ──────────────────── @pytest.mark.asyncio async def test_member_can_view_assignment_spend(db_session: AsyncSession): """A member sees the spend endpoint where pre-feature only the Owner could. The bucket happens to be ``unknown`` (no bindings on this fixture's key) — what we're asserting is that authz passed, not the cost math.""" provider = await _provider(db_session) owner = await _person(db_session, email="alice@x.com") member = await _person(db_session, email="bob@x.com") assignment = await _assignment( db_session, provider=provider, owner=owner, ) svc = ApiKeyService(db_session) await svc.add_member( assignment.id, person_id=member.id, current_user=_auth(owner), trace_id="t", ip_address=None, user_agent=None, ) # Owner — should not raise. out_owner = await svc.get_assignment_spend( assignment.id, current_user=_auth(owner), ) assert out_owner["assignment_id"] == str(assignment.id) # Member — should also not raise post-feature. out_member = await svc.get_assignment_spend( assignment.id, current_user=_auth(member), ) assert out_member["assignment_id"] == str(assignment.id) @pytest.mark.asyncio async def test_stranger_cannot_view_assignment_spend( db_session: AsyncSession, ): """A non-Owner non-member non-admin gets 403.""" provider = await _provider(db_session) owner = await _person(db_session, email="alice@x.com") stranger = await _person(db_session, email="eve@x.com") assignment = await _assignment( db_session, provider=provider, owner=owner, ) with pytest.raises(ForbiddenError): await ApiKeyService(db_session).get_assignment_spend( assignment.id, current_user=_auth(stranger), ) @pytest.mark.asyncio async def test_get_my_api_keys_includes_member_rows( db_session: AsyncSession, ): """A member sees the Owner's key on /apikeys/my — the whole point of the feature is that the team's panel agrees.""" provider = await _provider(db_session) owner = await _person(db_session, email="alice@x.com") member = await _person(db_session, email="bob@x.com") a1 = await _assignment(db_session, provider=provider, owner=owner) # Second key the member does NOT belong to — must not surface. other_owner = await _person(db_session, email="charlie@x.com") a2 = await _assignment( db_session, provider=provider, owner=other_owner, ) await ApiKeyService(db_session).add_member( a1.id, person_id=member.id, current_user=_auth(owner), trace_id="t", ip_address=None, user_agent=None, ) rows = await ApiKeyService(db_session).get_my_api_keys(_auth(member)) ids = {r.id for r in rows} assert a1.id in ids assert a2.id not in ids # Viewer role on the member's payload is 'member' — drives the # frontend identity Tag. for r in rows: if r.id == a1.id: assert r.viewer_role == "member" @pytest.mark.asyncio async def test_list_members_filters_leavers_inline( db_session: AsyncSession, ): """A member whose status flipped to a terminal offboarding state (``leaver_pending`` or ``left``) after being added is filtered out at read time, even if the row still exists in the DB. This is defence-in-depth on top of the leaver sweep — race window between flip and sweep is closed at read. We exercise ``leaver_pending`` here because flipping directly to ``left`` would trip the ``LeaverBlockedGuard`` SQLAlchemy event listener (which requires an authorized via ``flip_to_left`` call); ``leaver_pending`` is the realistic mid-offboarding state a member could be in while their session / PAT is still live, so it's also the right state to test the read-time gate against. """ provider = await _provider(db_session) owner = await _person(db_session, email="alice@x.com") teammate = await _person(db_session, email="bob@x.com") assignment = await _assignment( db_session, provider=provider, owner=owner, ) svc = ApiKeyService(db_session) await svc.add_member( assignment.id, person_id=teammate.id, current_user=_auth(owner), trace_id="t", ip_address=None, user_agent=None, ) # Mid-offboarding state; ``is_member`` filters this AND ``left``. teammate.employment_status = "leaver_pending" await db_session.flush() rows = await svc.list_members( assignment.id, current_user=_auth(owner), ) # Note: ``list_members`` itself only filters ``left`` at the SQL # join level (matching the user-visible "former colleagues # disappear from the panel" semantics). The ``is_member`` # read-gate filters BOTH; let's exercise that one directly so # the more strict invariant has coverage even if the panel # representation changes. assert not await svc.is_member(assignment.id, teammate.id) # And panel still renders the in-progress row until the final # flip to ``left`` (visible to Owner so they know to remove # before fully losing the audit trail). assert any(r.person_id == teammate.id for r in rows) # ── Transfer auto-handover ─────────────────────────────────────────── @pytest.mark.asyncio async def test_transfer_promotes_old_owner_to_member( db_session: AsyncSession, ): """apply_transfer flips ``assigned_to`` AND auto-adds the old Owner as a member so the handover doesn't strip read access on the colleague who may still need to consult spend during the transition.""" provider = await _provider(db_session) old_owner = await _person(db_session, email="alice@x.com") new_owner = await _person(db_session, email="bob@x.com") assignment = await _assignment( db_session, provider=provider, owner=old_owner, ) await ApiKeyService(db_session).apply_transfer( assignment_id=assignment.id, new_owner_id=new_owner.id, transfer_request_id=uuid.uuid4(), offboarding_process_id=None, ) await db_session.refresh(assignment) assert assignment.assigned_to == new_owner.id members = await _member_rows(db_session, assignment.id) member_pids = {m.person_id for m in members} assert old_owner.id in member_pids # New owner is NOT a member of themselves. assert new_owner.id not in member_pids @pytest.mark.asyncio async def test_transfer_drops_new_owner_from_roster( db_session: AsyncSession, ): """If the new Owner was already a member of the key, accepting the transfer drops the membership row — avoid the "I'm both owner and member of myself" UI confusion.""" provider = await _provider(db_session) old_owner = await _person(db_session, email="alice@x.com") new_owner = await _person(db_session, email="bob@x.com") assignment = await _assignment( db_session, provider=provider, owner=old_owner, ) # Add the prospective new owner as a member first. await ApiKeyService(db_session).add_member( assignment.id, person_id=new_owner.id, current_user=_auth(old_owner), trace_id="t", ip_address=None, user_agent=None, ) await ApiKeyService(db_session).apply_transfer( assignment_id=assignment.id, new_owner_id=new_owner.id, transfer_request_id=uuid.uuid4(), offboarding_process_id=None, ) members = await _member_rows(db_session, assignment.id) member_pids = {m.person_id for m in members} # Old owner is now a member; new owner is NOT. assert old_owner.id in member_pids assert new_owner.id not in member_pids # ── Leaver sweep ───────────────────────────────────────────────────── @pytest.mark.asyncio async def test_cleanup_leaver_memberships_strips_every_row( db_session: AsyncSession, ): """When a person flips to ``left``, every roster row that references them is removed. The sweep is idempotent and emits one audit row per (assignment, leaver) pair.""" provider = await _provider(db_session) owner_a = await _person(db_session, email="alice@x.com") owner_b = await _person(db_session, email="bob@x.com") leaver = await _person(db_session, email="charlie@x.com") a1 = await _assignment(db_session, provider=provider, owner=owner_a) a2 = await _assignment(db_session, provider=provider, owner=owner_b) svc = ApiKeyService(db_session) await svc.add_member( a1.id, person_id=leaver.id, current_user=_auth(owner_a), trace_id="t", ip_address=None, user_agent=None, ) await svc.add_member( a2.id, person_id=leaver.id, current_user=_auth(owner_b), trace_id="t", ip_address=None, user_agent=None, ) removed = await svc.cleanup_leaver_memberships( person_id=leaver.id, trace_id="leaver-1", actor_user_id=None, via="hris_sync", ) assert removed == 2 # Idempotent — second call removes zero. removed_again = await svc.cleanup_leaver_memberships( person_id=leaver.id, trace_id="leaver-1", actor_user_id=None, via="hris_sync", ) assert removed_again == 0 # Neither assignment still has the leaver on the roster. for assignment_id in (a1.id, a2.id): rows = await _member_rows(db_session, assignment_id) assert leaver.id not in {r.person_id for r in rows} # ── viewer_role resolution on response ─────────────────────────────── @pytest.mark.asyncio async def test_viewer_role_owner_member_admin_stranger( db_session: AsyncSession, ): """The denormalised ``viewer_role`` on the assignment response matches the caller's relationship to the row. Precedence: owner > admin > member > none.""" provider = await _provider(db_session) owner = await _person(db_session, email="alice@x.com") member = await _person(db_session, email="bob@x.com") admin = await _person(db_session, email="ada@x.com") assignment = await _assignment( db_session, provider=provider, owner=owner, ) svc = ApiKeyService(db_session) await svc.add_member( assignment.id, person_id=member.id, current_user=_auth(owner), trace_id="t", ip_address=None, user_agent=None, ) # Owner sees viewer_role='owner' owner_list = await svc.get_my_api_keys(_auth(owner)) assert owner_list[0].viewer_role == "owner" # Member sees viewer_role='member' member_list = await svc.get_my_api_keys(_auth(member)) assert any( r.id == assignment.id and r.viewer_role == "member" for r in member_list ) # Admin running list_assignments sees viewer_role='admin' admin_rows, _ = await svc.list_assignments( current_user=_auth(admin, admin=True), assigned_to=owner.id, ) matched = [r for r in admin_rows if r.id == assignment.id] assert matched and matched[0].viewer_role == "admin"