"""Workstream D: deterministic Workspace projection planning (design D8, D14, D29, D30).""" import random from dataclasses import replace from uuid import uuid4 import pytest from src.domain.project_auth.membership_reconciliation import ( LEGACY, LOCAL, WORKSPACE, CurrentBinding, CurrentMembership, DesiredIdentity, OpKind, ProjectionError, plan_reconciliation, ) ADMIN = uuid4() FINANCE = uuid4() def owner(email="owner@x.test"): return DesiredIdentity(email, True, False) def admin(email): return DesiredIdentity(email, False, True) def member(email, *, source=WORKSPACE, is_owner=False, cloud=None, bindings=(), holds=()): return CurrentMembership( membership_id=uuid4(), email=email, source=source, is_owner=is_owner, cloud_app_user_id=cloud, bindings=frozenset(bindings), holds=frozenset(holds), ) def owner_m(): return member("owner@x.test", is_owner=True) def apply_to_state(current, plan): """Test-only applier: replays operations onto in-memory state, checking the owner invariant.""" state = {m.email: m for m in current} for op in plan.operations: if op.kind is OpKind.CREATE_MEMBERSHIP: state[op.email] = member(op.email, source=op.source, is_owner=bool(op.value)) continue m = state[op.email] if op.kind is OpKind.UPDATE_OWNER: m = replace(m, is_owner=bool(op.value)) elif op.kind is OpKind.ADD_SUSPENSION: m = replace(m, holds=m.holds | {op.source}) elif op.kind is OpKind.REMOVE_SUSPENSION: m = replace(m, holds=m.holds - {op.source}) elif op.kind is OpKind.GRANT_ROLE: m = replace(m, bindings=m.bindings | {CurrentBinding(op.role_id, op.source)}) elif op.kind is OpKind.REVOKE_ROLE: m = replace(m, bindings=m.bindings - {CurrentBinding(op.role_id, op.source)}) elif op.kind is OpKind.LINK_CLOUD_USER: m = replace(m, cloud_app_user_id=op.value) state[op.email] = m assert sum(1 for s in state.values() if s.is_owner) <= 1, "owner invariant broken mid-apply" if m.is_owner: assert not m.holds, "owner must never be held" return list(state.values()) def plan(desired, current, **kw): return plan_reconciliation(desired, current, admin_role_id=ADMIN, **kw) def converged(desired, current, **kw): """Apply once, then assert a second plan is empty (idempotency).""" first = plan(desired, current, **kw) after = apply_to_state(current, first) assert plan(desired, after, **kw).operations == () return first, after def test_empty_project_creates_owner_and_admin_then_converges(): desired = [owner(), admin("admin@x.test")] first, _ = converged(desired, []) kinds = {op.kind for op in first.operations} assert OpKind.CREATE_MEMBERSHIP in kinds and OpKind.GRANT_ROLE in kinds def test_owner_transfer_demotes_old_owner_before_promotion_and_never_has_two(): old = member("old@x.test", is_owner=True) desired = [owner("new@x.test"), admin("old@x.test")] result, after = converged(desired, [old]) demote = result.operations.index(next( op for op in result.operations if op.kind is OpKind.UPDATE_OWNER and op.email == "old@x.test" )) create = result.operations.index(next( op for op in result.operations if op.kind is OpKind.CREATE_MEMBERSHIP and op.email == "new@x.test" )) assert demote < create assert [m.email for m in after if m.is_owner] == ["new@x.test"] def test_removed_workspace_admin_loses_binding_and_keeps_local_authority(): alice = member("alice@x.test", bindings=[CurrentBinding(ADMIN, WORKSPACE), CurrentBinding(FINANCE, LOCAL)]) result, after = converged([owner()], [owner_m(), alice]) revokes = [op for op in result.operations if op.kind is OpKind.REVOKE_ROLE] assert [(op.role_id, op.source) for op in revokes] == [(ADMIN, WORKSPACE)] assert all(op.kind is not OpKind.ADD_SUSPENSION for op in result.operations) alice_after = next(m for m in after if m.email == "alice@x.test") assert alice_after.bindings == frozenset({CurrentBinding(FINANCE, LOCAL)}) assert not alice_after.holds def test_same_role_from_workspace_and_local_survives_workspace_removal(): alice = member("alice@x.test", bindings=[CurrentBinding(ADMIN, WORKSPACE), CurrentBinding(ADMIN, LOCAL)]) result, after = converged([owner()], [owner_m(), alice]) assert [(op.kind, op.role_id, op.source) for op in result.operations] == [ (OpKind.REVOKE_ROLE, ADMIN, WORKSPACE) ] alice_after = next(m for m in after if m.email == "alice@x.test") assert alice_after.bindings == frozenset({CurrentBinding(ADMIN, LOCAL)}) def test_removed_identity_without_local_authority_gets_workspace_hold(): gone = member("gone@x.test", bindings=[CurrentBinding(ADMIN, WORKSPACE)]) result, after = converged([owner()], [owner_m(), gone]) kinds = {(op.kind, op.email, op.source) for op in result.operations} assert (OpKind.ADD_SUSPENSION, "gone@x.test", WORKSPACE) in kinds gone_after = next(m for m in after if m.email == "gone@x.test") assert gone_after.holds == frozenset({WORKSPACE}) def test_removed_identity_with_local_binding_is_not_held(): gone = member("gone@x.test", bindings=[CurrentBinding(FINANCE, LOCAL)]) result, _ = converged([owner()], [owner_m(), gone]) assert all(op.kind is not OpKind.ADD_SUSPENSION for op in result.operations) def test_returning_identity_has_workspace_hold_lifted_and_local_hold_kept(): returning = member("back@x.test", holds=[WORKSPACE, LOCAL]) desired = [owner(), admin("back@x.test")] result, after = converged(desired, [owner_m(), returning]) kinds = {(o.kind, o.email) for o in result.operations} assert (OpKind.REMOVE_SUSPENSION, "back@x.test") in kinds back = next(m for m in after if m.email == "back@x.test") assert back.holds == frozenset({LOCAL}) assert "back@x.test:local" in result.held def test_removed_former_owner_is_demoted_then_held_and_converges(): former = member("former@x.test", is_owner=True) _, after = converged([owner("new@x.test")], [former]) former_after = next(m for m in after if m.email == "former@x.test") assert former_after.is_owner is False assert former_after.holds == frozenset({WORKSPACE}) def test_local_only_membership_is_never_touched(): local = member("local@x.test", source=LOCAL, bindings=[CurrentBinding(FINANCE, LOCAL)], holds=[LOCAL]) assert plan([owner()], [owner_m(), local]).operations == () def test_workspace_owner_loses_stale_admin_binding_and_keeps_local(): owner_member = member("owner@x.test", is_owner=True, bindings=[CurrentBinding(ADMIN, WORKSPACE), CurrentBinding(FINANCE, LOCAL)]) result = plan([owner("owner@x.test")], [owner_member]) assert [(op.kind, op.role_id) for op in result.operations] == [(OpKind.REVOKE_ROLE, ADMIN)] def test_legacy_hold_on_returning_admin_is_blocked_never_cleared(): legacy = member("admin@x.test", holds=[LEGACY]) result = plan([owner(), admin("admin@x.test")], [owner_m(), legacy]) assert all(op.kind is not OpKind.REMOVE_SUSPENSION for op in result.operations) assert result.blocked == ("legacy_hold_requires_reinstatement:admin@x.test",) def test_owner_held_locally_refuses_projection(): held_owner = member("owner@x.test", holds=[LOCAL]) with pytest.raises(ProjectionError, match="owner_held_by:local"): plan([owner("owner@x.test")], [held_owner]) def test_owner_with_workspace_hold_is_lifted_before_promotion(): held = member("new@x.test", holds=[WORKSPACE]) result, after = converged([owner("new@x.test")], [member("old@x.test", is_owner=True), held]) order = [(op.kind, op.email) for op in result.operations] lift = order.index((OpKind.REMOVE_SUSPENSION, "new@x.test")) promote = order.index((OpKind.UPDATE_OWNER, "new@x.test")) assert lift < promote assert [m.email for m in after if m.is_owner] == ["new@x.test"] def test_cloud_link_emitted_once_and_not_after_apply(): existing = member("admin@x.test", bindings=[CurrentBinding(ADMIN, WORKSPACE)]) desired = [owner(), admin("admin@x.test")] result = plan(desired, [owner_m(), existing], cloud_user_ids={"Admin@X.test": "usr_1"}) assert [op.kind for op in result.operations] == [OpKind.LINK_CLOUD_USER] linked = apply_to_state([owner_m(), existing], result) assert plan(desired, linked, cloud_user_ids={"admin@x.test": "usr_1"}).operations == () def test_missing_owner_refuses_projection(): with pytest.raises(ProjectionError, match="workspace_owner_missing"): plan([admin("admin@x.test")], []) def test_two_owners_refuse_projection(): with pytest.raises(ProjectionError): plan([owner("a@x.test"), DesiredIdentity("b@x.test", True, False)], []) def test_admin_without_admin_role_refuses_projection(): with pytest.raises(ProjectionError, match="admin_role_unavailable"): plan_reconciliation([owner(), admin("a@x.test")], [], admin_role_id=None) def test_duplicate_workspace_principal_refuses_projection(): with pytest.raises(ProjectionError, match="duplicate_workspace_principal"): plan([owner(), admin("A@x.test"), admin("a@x.test")], []) def test_duplicate_membership_refuses_projection(): with pytest.raises(ProjectionError, match="duplicate_membership"): plan([owner()], [member("a@x.test"), member("A@x.test")]) def test_plan_is_deterministic_regardless_of_input_order(): desired = [owner(), admin("b@x.test"), admin("a@x.test")] current = [owner_m(), member("c@x.test", bindings=[CurrentBinding(ADMIN, WORKSPACE)]), member("d@x.test", bindings=[CurrentBinding(ADMIN, WORKSPACE)])] base = plan(desired, current).operations rng = random.Random(7) for _ in range(5): d, c = desired[:], current[:] rng.shuffle(d) rng.shuffle(c) assert plan(d, c).operations == base def test_mixed_scenario_converges_in_one_pass(): current = [ member("old-owner@x.test", is_owner=True), member("kept-admin@x.test", bindings=[CurrentBinding(ADMIN, WORKSPACE)]), member("dropped@x.test", bindings=[CurrentBinding(ADMIN, WORKSPACE), CurrentBinding(FINANCE, LOCAL)]), member("local-only@x.test", source=LOCAL, bindings=[CurrentBinding(FINANCE, LOCAL)]), ] desired = [owner("new-owner@x.test"), admin("kept-admin@x.test"), admin("new-admin@x.test")] _, after = converged(desired, current) assert [m.email for m in after if m.is_owner] == ["new-owner@x.test"] local = next(m for m in after if m.email == "local-only@x.test") assert local.bindings == frozenset({CurrentBinding(FINANCE, LOCAL)})