from __future__ import annotations import unittest from types import SimpleNamespace from fastapi import HTTPException from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef from govoplan_core.db.base import Base from govoplan_poll.backend.db.models import Poll, PollInvitation, PollLifecycleTransition, PollOption, PollResponse from govoplan_poll.backend.router import ( api_create_poll, api_get_poll, api_list_poll_invitations, api_list_poll_responses, api_list_polls, api_poll_summary, api_submit_poll_response, ) from govoplan_poll.backend.schemas import ( PollAnswerInput, PollCreateRequest, PollInvitationCreateRequest, PollSubmitResponseRequest, ) from govoplan_poll.backend.service import close_poll, create_poll, create_poll_invitation, open_poll READ_SCOPE = "poll:poll:read" WRITE_SCOPE = "poll:poll:write" RESPOND_SCOPE = "poll:response:write" class PollAuthorizationTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:") Base.metadata.create_all( self.engine, tables=[ Poll.__table__, PollOption.__table__, PollResponse.__table__, PollInvitation.__table__, PollLifecycleTransition.__table__, ], ) self.Session = sessionmaker(bind=self.engine) self.session: Session = self.Session() def tearDown(self) -> None: self.session.close() Base.metadata.drop_all( self.engine, tables=[ PollLifecycleTransition.__table__, PollInvitation.__table__, PollResponse.__table__, PollOption.__table__, Poll.__table__, ], ) self.engine.dispose() @staticmethod def _principal( account_id: str, *, tenant_id: str = "tenant-1", user_id: str | None = None, scopes: set[str] | None = None, ) -> ApiPrincipal: membership_id = user_id or f"membership-{account_id}" display_name = f"Actor {account_id}" return ApiPrincipal( principal=PrincipalRef( account_id=account_id, membership_id=membership_id, tenant_id=tenant_id, scopes=frozenset(scopes or {READ_SCOPE, RESPOND_SCOPE}), display_name=display_name, ), account=SimpleNamespace(id=account_id), user=SimpleNamespace(id=membership_id), ) def _poll( self, *, owner_id: str = "owner", visibility: str = "tenant", result_visibility: str = "after_close", title: str = "Decision", ) -> Poll: poll = create_poll( self.session, tenant_id="tenant-1", user_id=owner_id, payload=PollCreateRequest( title=title, kind="yes_no", visibility=visibility, result_visibility=result_visibility, ), ) open_poll(self.session, tenant_id="tenant-1", poll_id=poll.id) return poll @staticmethod def _answer(*, claimed_respondent: str = "victim") -> PollSubmitResponseRequest: return PollSubmitResponseRequest( respondent_id=claimed_respondent, respondent_label="Claimed victim", answers=[PollAnswerInput(option_key="yes")], ) def test_authenticated_response_identity_is_bound_to_actor(self) -> None: poll = self._poll(result_visibility="public") first_actor = self._principal("account-1") second_actor = self._principal("account-2") first = api_submit_poll_response( poll.id, self._answer(), session=self.session, principal=first_actor, ) second = api_submit_poll_response( poll.id, self._answer(), session=self.session, principal=second_actor, ) updated_first = api_submit_poll_response( poll.id, PollSubmitResponseRequest( respondent_id="account-2", answers=[PollAnswerInput(option_key="no")], ), session=self.session, principal=first_actor, ) self.assertEqual(first.respondent_id, "account-1") self.assertEqual(first.respondent_label, "Actor account-1") self.assertEqual(second.respondent_id, "account-2") self.assertNotEqual(first.id, second.id) self.assertEqual(updated_first.id, first.id) self.assertEqual(updated_first.answers[0]["option_key"], "no") cross_tenant_actor = self._principal("account-1", tenant_id="tenant-2") with self.assertRaises(HTTPException) as cross_tenant: api_submit_poll_response( poll.id, self._answer(), session=self.session, principal=cross_tenant_actor, ) self.assertEqual(cross_tenant.exception.status_code, 404) def test_create_route_persists_after_request_session_closes(self) -> None: created = api_create_poll( PollCreateRequest(title="Persisted poll", kind="yes_no", visibility="tenant"), session=self.session, principal=self._principal("owner", scopes={WRITE_SCOPE}), ) self.session.close() self.session = self.Session() persisted = self.session.query(Poll).filter(Poll.id == created.id).one_or_none() self.assertIsNotNone(persisted) self.assertEqual(persisted.title, "Persisted poll") def test_authenticated_response_cannot_claim_an_invitation(self) -> None: poll = self._poll(result_visibility="public") response = api_submit_poll_response( poll.id, self._answer().model_copy( update={"metadata": {"invitation_id": "forged-invitation", "client_note": "kept"}} ), session=self.session, principal=self._principal("account-1"), ) self.assertEqual(response.metadata, {"client_note": "kept"}) def test_private_poll_requires_ownership_management_or_active_assignment(self) -> None: poll = self._poll(visibility="private") reader = self._principal("reader") listed = api_list_polls( status_filter=None, kind=None, session=self.session, principal=reader, ) self.assertEqual(listed.polls, []) with self.assertRaises(HTTPException) as hidden: api_get_poll(poll.id, session=self.session, principal=reader) self.assertEqual(hidden.exception.status_code, 404) invitation, _token = create_poll_invitation( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollInvitationCreateRequest(respondent_id=reader.membership_id), ) visible = api_get_poll(poll.id, session=self.session, principal=reader) self.assertEqual(visible.id, poll.id) invitation.revoked_at = invitation.created_at self.session.flush() with self.assertRaises(HTTPException) as revoked: api_get_poll(poll.id, session=self.session, principal=reader) self.assertEqual(revoked.exception.status_code, 404) manager = self._principal("manager", scopes={READ_SCOPE, WRITE_SCOPE}) self.assertEqual(api_get_poll(poll.id, session=self.session, principal=manager).id, poll.id) other_tenant = self._principal("owner", tenant_id="tenant-2", scopes={READ_SCOPE, WRITE_SCOPE}) with self.assertRaises(HTTPException) as cross_tenant: api_get_poll(poll.id, session=self.session, principal=other_tenant) self.assertEqual(cross_tenant.exception.status_code, 404) def test_after_response_results_do_not_leak_to_other_actors(self) -> None: poll = self._poll(result_visibility="after_response") respondent = self._principal("respondent") bystander = self._principal("bystander") with self.assertRaises(HTTPException) as before_response: api_poll_summary(poll.id, session=self.session, principal=respondent) self.assertEqual(before_response.exception.status_code, 403) api_submit_poll_response(poll.id, self._answer(), session=self.session, principal=respondent) self.assertEqual(api_poll_summary(poll.id, session=self.session, principal=respondent).response_count, 1) with self.assertRaises(HTTPException) as own_raw_responses: api_list_poll_responses(poll.id, session=self.session, principal=respondent) self.assertEqual(own_raw_responses.exception.status_code, 403) with self.assertRaises(HTTPException) as other_actor_summary: api_poll_summary(poll.id, session=self.session, principal=bystander) self.assertEqual(other_actor_summary.exception.status_code, 403) with self.assertRaises(HTTPException) as other_actor_responses: api_list_poll_responses(poll.id, session=self.session, principal=bystander) self.assertEqual(other_actor_responses.exception.status_code, 403) def test_participants_can_see_aggregate_but_not_raw_responses_or_invitation_roster(self) -> None: poll = self._poll(result_visibility="public") first_actor = self._principal("account-1") second_actor = self._principal("account-2") create_poll_invitation( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollInvitationCreateRequest(respondent_id=first_actor.membership_id, email="one@example.test"), ) create_poll_invitation( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollInvitationCreateRequest(respondent_id=second_actor.membership_id, email="two@example.test"), ) api_submit_poll_response(poll.id, self._answer(), session=self.session, principal=first_actor) api_submit_poll_response(poll.id, self._answer(), session=self.session, principal=second_actor) self.assertEqual(api_poll_summary(poll.id, session=self.session, principal=first_actor).response_count, 2) with self.assertRaises(HTTPException) as raw_responses: api_list_poll_responses(poll.id, session=self.session, principal=first_actor) self.assertEqual(raw_responses.exception.status_code, 403) with self.assertRaises(HTTPException) as invitation_roster: api_list_poll_invitations(poll.id, session=self.session, principal=first_actor) self.assertEqual(invitation_roster.exception.status_code, 403) organizer = self._principal("owner", scopes={READ_SCOPE}) self.assertEqual(len(api_list_poll_responses(poll.id, session=self.session, principal=organizer).responses), 2) self.assertEqual(len(api_list_poll_invitations(poll.id, session=self.session, principal=organizer).invitations), 2) manager = self._principal("manager", scopes={WRITE_SCOPE}) self.assertEqual(len(api_list_poll_responses(poll.id, session=self.session, principal=manager).responses), 2) self.assertEqual(len(api_list_poll_invitations(poll.id, session=self.session, principal=manager).invitations), 2) def test_after_close_and_organizer_result_visibility_are_enforced(self) -> None: after_close = self._poll(result_visibility="after_close", title="After close") reader = self._principal("reader") with self.assertRaises(HTTPException) as still_open: api_poll_summary(after_close.id, session=self.session, principal=reader) self.assertEqual(still_open.exception.status_code, 403) close_poll(self.session, tenant_id="tenant-1", poll_id=after_close.id) self.assertEqual(api_poll_summary(after_close.id, session=self.session, principal=reader).response_count, 0) organizer_only = self._poll(result_visibility="organizer", title="Organizer only") api_submit_poll_response(organizer_only.id, self._answer(), session=self.session, principal=reader) with self.assertRaises(HTTPException) as participant_denied: api_poll_summary(organizer_only.id, session=self.session, principal=reader) self.assertEqual(participant_denied.exception.status_code, 403) owner = self._principal("owner", scopes={READ_SCOPE}) self.assertEqual(api_poll_summary(organizer_only.id, session=self.session, principal=owner).response_count, 1) if __name__ == "__main__": unittest.main()