from __future__ import annotations import unittest from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from govoplan_core.db.base import Base from govoplan_poll.backend.db.models import Poll, PollInvitation, PollOption, PollResponse from govoplan_poll.backend.schemas import ( PollAnswerInput, PollCreateRequest, PollInvitationCreateRequest, PollOptionInput, PollSubmitResponseRequest, ) from govoplan_poll.backend.service import ( PollError, create_poll, create_poll_invitation, open_poll, poll_result_summary_by_id, submit_poll_response, submit_poll_response_with_token, ) class PollServiceTests(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() def test_single_choice_response_can_update_existing_respondent(self) -> None: poll = create_poll( self.session, tenant_id="tenant-1", user_id="user-1", payload=PollCreateRequest( title="Lunch", kind="single_choice", options=[PollOptionInput(key="a", label="A"), PollOptionInput(key="b", label="B")], ), ) open_poll(self.session, tenant_id="tenant-1", poll_id=poll.id) first = submit_poll_response( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollSubmitResponseRequest(respondent_id="person-1", answers=[PollAnswerInput(option_key="a")]), ) second = submit_poll_response( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollSubmitResponseRequest(respondent_id="person-1", answers=[PollAnswerInput(option_key="b")]), ) summary = poll_result_summary_by_id(self.session, tenant_id="tenant-1", poll_id=poll.id) self.assertEqual(first.id, second.id) self.assertEqual(summary["response_count"], 1) self.assertEqual(summary["leading_option_ids"], [second.answers[0]["option_id"]]) def test_anonymous_response_requires_poll_policy(self) -> None: poll = create_poll( self.session, tenant_id="tenant-1", user_id=None, payload=PollCreateRequest( title="Decision", kind="yes_no", ), ) open_poll(self.session, tenant_id="tenant-1", poll_id=poll.id) with self.assertRaisesRegex(PollError, "Anonymous responses are not allowed"): submit_poll_response( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollSubmitResponseRequest(answers=[PollAnswerInput(option_key="yes")]), ) def test_yes_no_maybe_uses_default_options(self) -> None: poll = create_poll( self.session, tenant_id="tenant-1", user_id=None, payload=PollCreateRequest( title="Tentative decision", kind="yes_no_maybe", allow_anonymous=True, ), ) open_poll(self.session, tenant_id="tenant-1", poll_id=poll.id) response = submit_poll_response( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollSubmitResponseRequest(answers=[PollAnswerInput(option_key="maybe")]), ) summary = poll_result_summary_by_id(self.session, tenant_id="tenant-1", poll_id=poll.id) self.assertEqual([option.key for option in poll.options], ["yes", "no", "maybe"]) self.assertEqual(response.answers[0]["option_key"], "maybe") self.assertEqual(summary["response_count"], 1) def test_poll_can_store_scheduling_workflow_context(self) -> None: poll = create_poll( self.session, tenant_id="tenant-1", user_id="user-1", payload=PollCreateRequest( title="Find a steering group slot", kind="availability", context_module="scheduling", context_resource_type="scheduling_request", context_resource_id="request-1", workflow_state="collecting_availability", workflow_steps=[ {"id": "collect", "label": "Collect availability", "status": "active"}, {"id": "decide", "label": "Choose slot", "status": "pending"}, ], options=[PollOptionInput(key="slot-1", label="Monday")], ), ) self.assertEqual(poll.context_module, "scheduling") self.assertEqual(poll.context_resource_id, "request-1") self.assertEqual(poll.workflow_state, "collecting_availability") self.assertEqual(poll.workflow_steps[0]["id"], "collect") def test_ranked_choice_summary_uses_borda_score(self) -> None: poll = create_poll( self.session, tenant_id="tenant-1", user_id="user-1", payload=PollCreateRequest( title="Priority", kind="ranked_choice", options=[ PollOptionInput(key="alpha", label="Alpha"), PollOptionInput(key="beta", label="Beta"), PollOptionInput(key="gamma", label="Gamma"), ], ), ) open_poll(self.session, tenant_id="tenant-1", poll_id=poll.id) submit_poll_response( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollSubmitResponseRequest( respondent_id="one", answers=[ PollAnswerInput(option_key="alpha"), PollAnswerInput(option_key="beta"), PollAnswerInput(option_key="gamma"), ], ), ) submit_poll_response( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollSubmitResponseRequest( respondent_id="two", answers=[ PollAnswerInput(option_key="beta"), PollAnswerInput(option_key="alpha"), PollAnswerInput(option_key="gamma"), ], ), ) summary = poll_result_summary_by_id(self.session, tenant_id="tenant-1", poll_id=poll.id) scores = {result["option_key"]: result["score"] for result in summary["option_results"]} self.assertEqual(scores["alpha"], 5) self.assertEqual(scores["beta"], 5) self.assertEqual(scores["gamma"], 2) expected_leaders = { result["option_id"] for result in summary["option_results"] if result["option_key"] in {"alpha", "beta"} } self.assertEqual(set(summary["leading_option_ids"]), expected_leaders) def test_availability_summary_counts_available_and_maybe(self) -> None: poll = create_poll( self.session, tenant_id="tenant-1", user_id="user-1", payload=PollCreateRequest( title="Find a slot", kind="availability", allow_anonymous=True, options=[ PollOptionInput(key="slot-1", label="Monday", value={"start": "2026-07-13T09:00:00Z"}), PollOptionInput(key="slot-2", label="Tuesday", value={"start": "2026-07-14T09:00:00Z"}), ], ), ) open_poll(self.session, tenant_id="tenant-1", poll_id=poll.id) submit_poll_response( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollSubmitResponseRequest( answers=[ PollAnswerInput(option_key="slot-1", value="available"), PollAnswerInput(option_key="slot-2", value="maybe"), ] ), ) submit_poll_response( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollSubmitResponseRequest( answers=[ PollAnswerInput(option_key="slot-1", value="unavailable"), PollAnswerInput(option_key="slot-2", value="available"), ] ), ) summary = poll_result_summary_by_id(self.session, tenant_id="tenant-1", poll_id=poll.id) values = {result["option_key"]: result["values"] for result in summary["option_results"]} self.assertEqual(summary["response_count"], 2) self.assertEqual(values["slot-1"], {"available": 1, "unavailable": 1}) self.assertEqual(values["slot-2"], {"maybe": 1, "available": 1}) def test_signed_invitation_can_submit_without_anonymous_policy(self) -> None: poll = create_poll( self.session, tenant_id="tenant-1", user_id="user-1", payload=PollCreateRequest( title="External vote", kind="yes_no_maybe", ), ) open_poll(self.session, tenant_id="tenant-1", poll_id=poll.id) invitation, token = create_poll_invitation( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollInvitationCreateRequest(respondent_label="External participant"), ) response = submit_poll_response_with_token( self.session, token=token, payload=PollSubmitResponseRequest(answers=[PollAnswerInput(option_key="yes")]), ) self.assertEqual(response.respondent_id, f"invitation:{invitation.id}") self.assertEqual(response.respondent_label, "External participant") self.assertEqual(response.metadata_["invitation_id"], invitation.id) self.assertIsNotNone(invitation.last_used_at) if __name__ == "__main__": unittest.main()