Add explicit acting-in-place context

tests/test_auth_dependencies.py

@@ -11,6 +11,7 @@ from starlette.requests import Request
 
 from govoplan_access.backend.auth.dependencies import (
     _extract_token,
+    _principal_idm_context,
     _requires_csrf,
     _resolve_legacy_principal_context,
     _resolve_legacy_principal_ref,
@@ -21,6 +22,7 @@ from govoplan_access.backend.db.base import AccessBase
 from govoplan_access.backend.db.models import Account, AuthSession, Role, User, UserRoleAssignment
 from govoplan_core.core.change_sequence import ChangeSequenceEntry, ChangeSequenceRetentionFloor
 from govoplan_core.core.principal_cache import invalidate_auth_principals
+from govoplan_core.core.idm import OrganizationFunctionAssignmentRef
 from govoplan_core.db.base import Base
 from govoplan_core.security.time import utc_now
 from govoplan_core.settings import settings
@@ -156,6 +158,108 @@ class AuthDependencyTests(unittest.TestCase):
             )
             engine.dispose()
 
+    def test_acting_assignment_requires_exact_session_selection(self) -> None:
+        class Directory:
+            def organization_function_assignments_for_account(
+                self,
+                account_id: str,
+                *,
+                tenant_id: str | None = None,
+                effective_at=None,
+            ):
+                del account_id, effective_at
+                return (
+                    OrganizationFunctionAssignmentRef(
+                        id="direct-1",
+                        tenant_id=str(tenant_id),
+                        identity_id="identity-1",
+                        account_id="account-1",
+                        function_id="function-direct",
+                        organization_unit_id="unit-1",
+                    ),
+                    OrganizationFunctionAssignmentRef(
+                        id="acting-1",
+                        tenant_id=str(tenant_id),
+                        identity_id="identity-1",
+                        account_id="account-1",
+                        function_id="function-acting",
+                        organization_unit_id="unit-1",
+                        source="acting_for",
+                        acting_for_account_id="represented-1",
+                    ),
+                )
+
+        engine = create_engine("sqlite:///:memory:")
+        create_scope_tables(engine)
+        AccessBase.metadata.create_all(bind=engine)
+        SessionLocal = sessionmaker(bind=engine)
+        try:
+            with SessionLocal() as session:
+                tenant = Tenant(id="tenant-1", slug="tenant-1", name="Tenant 1")
+                account = Account(
+                    id="account-1",
+                    email="actor@example.test",
+                    normalized_email="actor@example.test",
+                )
+                user = User(
+                    id="user-1",
+                    tenant_id=tenant.id,
+                    account_id=account.id,
+                    email=account.email,
+                )
+                auth_session = AuthSession(
+                    id="session-1",
+                    tenant_id=tenant.id,
+                    user_id=user.id,
+                    account_id=account.id,
+                    token_hash="token-hash",
+                    expires_at=utc_now() + timedelta(hours=1),
+                )
+                session.add_all((tenant, account, user, auth_session))
+                session.flush()
+
+                ordinary, _ = _principal_idm_context(
+                    session,
+                    user=user,
+                    account=account,
+                    tenant_id=tenant.id,
+                    idm_directory=Directory(),  # type: ignore[arg-type]
+                    organization_directory=None,
+                )
+                self.assertEqual([item.id for item in ordinary], ["direct-1"])
+
+                auth_session.acting_assignment_id = "acting-1"
+                auth_session.acting_for_account_id = "represented-1"
+                selected, _ = _principal_idm_context(
+                    session,
+                    user=user,
+                    account=account,
+                    tenant_id=tenant.id,
+                    idm_directory=Directory(),  # type: ignore[arg-type]
+                    organization_directory=None,
+                    auth_session=auth_session,
+                )
+                self.assertEqual(
+                    [item.id for item in selected],
+                    ["direct-1", "acting-1"],
+                )
+
+                auth_session.acting_for_account_id = "wrong-account"
+                mismatched, _ = _principal_idm_context(
+                    session,
+                    user=user,
+                    account=account,
+                    tenant_id=tenant.id,
+                    idm_directory=Directory(),  # type: ignore[arg-type]
+                    organization_directory=None,
+                    auth_session=auth_session,
+                )
+                self.assertEqual([item.id for item in mismatched], ["direct-1"])
+        finally:
+            AccessBase.metadata.drop_all(bind=engine)
+            scope_registry.metadata.drop_all(bind=engine)
+            engine.dispose()
+
 
 if __name__ == "__main__":
     unittest.main()