295 lines
12 KiB
Python
295 lines
12 KiB
Python
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, 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__],
|
|
)
|
|
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=[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()
|