"""Workstream E: canonical Core → Cloud membership projection (pure).""" from dataclasses import replace from uuid import uuid4 from src.domain.project_auth import permissions as perms from src.domain.project_auth.cloud_projection import ( BindingSnapshot, MembershipProjection, MembershipSnapshot, ProjectSnapshot, RoleSnapshot, build_project_projections, build_projection, plan_projection_writes, projection_from_cloud_payload, to_cloud_payload, ) WS = "workspace" LOCAL = "local" ADMIN = RoleSnapshot(uuid4(), "admin", "admin", False, frozenset(perms.ADMIN_PRESET)) STAFF = RoleSnapshot(uuid4(), "staff", "staff", False, frozenset({perms.STAFF})) MEMBER = RoleSnapshot(uuid4(), "member", "member", False, frozenset()) FINANCE_STAFF = RoleSnapshot(uuid4(), "finance", "staff", False, frozenset({perms.BILLING})) UNCLASSIFIED = RoleSnapshot(uuid4(), "legacy_custom", None, False, frozenset({perms.BILLING})) PROTECTED = RoleSnapshot(uuid4(), "super_admin", None, True, frozenset(perms.OWNER_PRESET)) OWNER_PERMS = frozenset(perms.OWNER_PRESET) def snapshot(*memberships): roles = {r.role_id: r for r in (ADMIN, STAFF, MEMBER, FINANCE_STAFF, UNCLASSIFIED, PROTECTED)} return ProjectSnapshot(roles=roles, protected_owner_permissions=OWNER_PERMS, memberships=tuple(memberships)) def member(email, *bindings, owner=False, holds=(), version=1, cloud=None): return MembershipSnapshot( membership_id=uuid4(), email=email, is_owner=owner, cloud_app_user_id=cloud, authorization_version=version, bindings=frozenset(BindingSnapshot(r.role_id, s) for r, s in bindings), holds=frozenset(holds), ) def project(*memberships): return build_project_projections(snapshot(*memberships)) def only(projections): [p] = projections return p def test_workspace_admin_only_projects_admin(): p = only(project(member("a@x.test", (ADMIN, WS)))) assert p.project_roles == ("admin",) assert p.legacy_role == "admin" def test_local_admin_only_projects_admin(): p = only(project(member("a@x.test", (ADMIN, LOCAL)))) assert p.project_roles == ("admin",) assert p.legacy_role == "admin" assert set(p.project_permissions) == set(perms.ADMIN_PRESET) def test_workspace_and_local_same_role_is_one_effective_role(): p = only(project(member("a@x.test", (ADMIN, WS), (ADMIN, LOCAL)))) assert p.project_roles == ("admin",) assert len(p.project_roles) == 1 def test_staff_plus_member_projects_staff_legacy(): p = only(project(member("a@x.test", (STAFF, WS), (MEMBER, LOCAL)))) assert set(p.project_roles) == {"staff", "member"} assert p.legacy_role == "staff" def test_admin_plus_staff_projects_admin_legacy(): p = only(project(member("a@x.test", (ADMIN, WS), (STAFF, LOCAL)))) assert p.legacy_role == "admin" def test_owner_with_other_roles_projects_super_admin_legacy_and_protected_permissions(): p = only(project(member("o@x.test", (ADMIN, LOCAL), owner=True))) assert p.is_owner is True assert p.legacy_role == "super_admin" assert OWNER_PERMS <= set(p.project_permissions) def test_owner_without_bindings_still_gets_protected_permissions(): p = only(project(member("o@x.test", owner=True))) assert set(p.project_permissions) == OWNER_PERMS assert p.project_roles == () def test_suspended_membership_projects_inactive_with_no_usable_authority(): p = only(project(member("a@x.test", (ADMIN, WS), holds=["workspace"]))) assert p.status == "suspended" assert p.project_roles == () and p.project_permissions == () assert p.is_owner is False assert p.legacy_role == "member" def test_suspended_owner_is_not_projected_as_owner(): p = only(project(member("o@x.test", owner=True, holds=["legacy"]))) assert p.status == "suspended" and p.is_owner is False assert p.project_permissions == () def test_unclassified_binding_is_excluded_never_elevated(): p = only(project(member("a@x.test", (UNCLASSIFIED, LOCAL)))) assert p.project_roles == () assert p.project_permissions == () assert p.excluded_roles == ("legacy_custom",) assert p.legacy_role == "member" def test_protected_role_bound_through_a_binding_is_excluded_not_granted(): p = only(project(member("a@x.test", (PROTECTED, LOCAL)))) assert p.is_owner is False assert p.project_roles == () assert p.project_permissions == () assert p.excluded_roles == ("super_admin",) def test_unknown_role_id_is_excluded_not_elevated(): stranger = RoleSnapshot(uuid4(), "ghost", "admin", False, frozenset(perms.ADMIN_PRESET)) m = MembershipSnapshot(uuid4(), "a@x.test", False, None, 1, frozenset({BindingSnapshot(stranger.role_id, LOCAL)}), frozenset()) p = build_projection(ProjectSnapshot(roles={}, protected_owner_permissions=frozenset(), memberships=(m,)), m) assert p.project_permissions == () and p.legacy_role == "member" assert p.excluded_roles == (str(stranger.role_id),) def test_finance_staff_role_projects_staff_legacy_with_its_own_permissions(): p = only(project(member("a@x.test", (FINANCE_STAFF, LOCAL)))) assert p.project_roles == ("finance",) assert p.project_permissions == (perms.BILLING,) assert p.legacy_role == "staff" def test_reordered_semantic_state_has_zero_drift(): desired = only(project(member("a@x.test", (ADMIN, WS), (STAFF, LOCAL), cloud="usr_1"))) reordered = replace( desired, project_roles=tuple(reversed(desired.project_roles)), project_permissions=tuple(reversed(desired.project_permissions)), ) assert reordered.semantic_key() == desired.semantic_key() assert reordered.digest() == desired.digest() assert plan_projection_writes([desired], {"a@x.test": reordered})[0].kind == "unchanged" def test_version_only_change_refreshes_without_audit(): before = only(project(member("a@x.test", (ADMIN, WS), version=3))) after = replace(before, authorization_version=4) [write] = plan_projection_writes([after], {"a@x.test": before}) assert write.kind == "upsert" and write.audit is False def test_effective_change_upserts_and_audits(): before = only(project(member("a@x.test", (ADMIN, WS)))) after = only(project(member("a@x.test", (STAFF, WS), version=2))) [write] = plan_projection_writes([after], {"a@x.test": before}) assert write.kind == "upsert" and write.audit is True def test_new_membership_creates_and_is_audited(): [write] = plan_projection_writes(project(member("new@x.test", (ADMIN, WS))), {}) assert write.kind == "upsert" and write.audit is True def test_removed_membership_retires_and_audits_only_if_it_was_active(): active = only(project(member("gone@x.test", (ADMIN, WS)))) held = only(project(member("gone@x.test", (ADMIN, WS), holds=["workspace"]))) [w_active] = plan_projection_writes([], {"gone@x.test": active}) [w_held] = plan_projection_writes([], {"gone@x.test": held}) assert w_active.kind == "retire" and w_active.audit is True assert w_held.kind == "retire" and w_held.audit is False def test_cloud_payload_roundtrips_to_the_same_semantic_state(): p = only(project(member("a@x.test", (ADMIN, WS), (MEMBER, LOCAL), cloud="usr_9", version=5))) payload = to_cloud_payload("prj_1", p) assert payload["project_id"] == "prj_1" and payload["schema_version"] == 1 parsed = projection_from_cloud_payload(payload) assert parsed.semantic_key() == p.semantic_key() assert parsed.authorization_version == 5 def test_unsupported_payload_schema_is_refused(): p = only(project(member("a@x.test", (ADMIN, WS)))) payload = to_cloud_payload("prj_1", p) payload["schema_version"] = 99 try: projection_from_cloud_payload(payload) except ValueError as exc: assert str(exc) == "unsupported_projection_schema" else: raise AssertionError("expected refusal") def test_projections_are_built_in_stable_email_order(): ps = project(member("b@x.test", (ADMIN, WS)), member("a@x.test", (STAFF, WS))) assert [p.email for p in ps] == ["a@x.test", "b@x.test"] def test_digest_depends_on_version_but_semantic_key_does_not(): p = only(project(member("a@x.test", (ADMIN, WS), version=1))) q = replace(p, authorization_version=2) assert p.semantic_key() == q.semantic_key() assert p.digest() != q.digest() assert isinstance(p, MembershipProjection)