281 lines
10 KiB
Python
281 lines
10 KiB
Python
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()
|