GovOPlaN/govoplan-poll
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(responses): invalidate changed options safely

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-20 18:30:04 +02:00
parent e0de54b756
commit 12b3175b07
3 changed files with 122 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
src/govoplan_poll/backend/capabilities.py
View File

@@ -31,6 +31,7 @@ from govoplan_poll.backend.service import (
create_poll_invitation,
decide_poll,
get_poll,
get_poll_response_for_respondents,
list_poll_responses,
open_poll,
poll_result_summary_by_id,
@@ -291,6 +292,27 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
raise PollCapabilityError(str(exc)) from exc
return tuple(_response_ref(response) for response in responses)
def get_response(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
respondent_ids: Sequence[str],
invitation_id: str | None = None,
) -> PollResponseRef | None:
try:
response = get_poll_response_for_respondents(
session,
tenant_id=tenant_id,
poll_id=poll_id,
respondent_ids=tuple(respondent_ids),
invitation_id=invitation_id,
)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
return _response_ref(response) if response is not None else None
@staticmethod
def _transition(callback, session: object, *, tenant_id: str, poll_id: str) -> PollRef:
try:

42
src/govoplan_poll/backend/service.py
View File

@@ -732,6 +732,44 @@ def _lock_poll_for_response(session: Session, *, tenant_id: str, poll_id: str) -
return poll
def get_poll_response_for_respondents(
session: Session,
*,
tenant_id: str,
poll_id: str,
respondent_ids: tuple[str, ...],
invitation_id: str | None = None,
) -> PollResponse | None:
"""Resolve a response from server-trusted respondent identities."""
get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
ids = tuple(dict.fromkeys(value for value in respondent_ids if value))
if ids:
response = (
session.query(PollResponse)
.filter(
PollResponse.tenant_id == tenant_id,
PollResponse.poll_id == poll_id,
PollResponse.respondent_id.in_(ids),
PollResponse.deleted_at.is_(None),
)
.order_by(PollResponse.submitted_at.desc(), PollResponse.id.desc())
.first()
)
if response is not None:
return response
if invitation_id is None:
return None
return next(
(
response
for response in reversed(list_poll_responses(session, tenant_id=tenant_id, poll_id=poll_id))
if (response.metadata_ or {}).get("invitation_id") == invitation_id
),
None,
)
def update_poll_option(
session: Session,
*,
@@ -793,6 +831,8 @@ def update_poll_option(
.with_for_update()
.all()
)
if responses and not poll.allow_response_update:
raise PollError("Poll options cannot be edited after responses when response updates are disabled")
option.label = label
option.description = description
option.value = normalized_value
@@ -805,6 +845,8 @@ def update_poll_option(
]
if retained_answers != (response.answers or []):
response.answers = retained_answers
if not retained_answers:
response.deleted_at = _now()
session.flush()
return option

60
tests/test_response_editing.py
View File

@@ -5,7 +5,7 @@ import unittest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from govoplan_core.core.poll import PollOptionUpdateCommand, PollSchedulingProvider
from govoplan_core.core.poll import PollCapabilityError, PollOptionUpdateCommand, PollSchedulingProvider
from govoplan_core.db.base import Base
from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
from govoplan_poll.backend.db.models import Poll, PollOption, PollResponse
@@ -31,7 +31,7 @@ class PollResponseEditingTests(unittest.TestCase):
)
self.engine.dispose()
def _poll(self) -> Poll:
def _poll(self, *, allow_response_update: bool = True) -> Poll:
return create_poll(
self.session,
tenant_id="tenant-1",
@@ -42,6 +42,7 @@ class PollResponseEditingTests(unittest.TestCase):
status="open",
min_choices=1,
max_choices=2,
allow_response_update=allow_response_update,
options=[
PollOptionInput(
key="slot-1",
@@ -85,6 +86,14 @@ class PollResponseEditingTests(unittest.TestCase):
poll_id=poll.id,
)
self.assertEqual([answer.value for answer in projected[0].answers], ["available", "maybe"])
current = provider.get_response(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
respondent_ids=("person-1",),
)
self.assertIsNotNone(current)
self.assertEqual([answer.value for answer in current.answers], ["available", "maybe"])
option_ref = provider.update_option(
self.session,
@@ -118,6 +127,53 @@ class PollResponseEditingTests(unittest.TestCase):
)
self.assertEqual(len(first.answers), 2)
def test_fully_invalidated_response_is_no_longer_active_or_counted(self) -> None:
poll = self._poll()
response = submit_poll_response(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
payload=PollSubmitResponseRequest(
respondent_id="person-1",
answers=[PollAnswerInput(option_id=poll.options[0].id, value="available")],
),
)
SqlPollSchedulingProvider().update_option(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
option_id=poll.options[0].id,
command=PollOptionUpdateCommand(label="Monday, revised", value={"slot_id": "slot-1"}, metadata={"source": "scheduling"}),
)
self.assertEqual(response.answers, [])
self.assertIsNotNone(response.deleted_at)
self.assertEqual(
SqlPollSchedulingProvider().list_responses(self.session, tenant_id="tenant-1", poll_id=poll.id),
(),
)
def test_option_change_is_rejected_when_responses_cannot_be_updated(self) -> None:
poll = self._poll(allow_response_update=False)
response = self._submit_both(poll, "person-1")
with self.assertRaisesRegex(PollCapabilityError, "response updates are disabled"):
SqlPollSchedulingProvider().update_option(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
option_id=poll.options[0].id,
command=PollOptionUpdateCommand(
label="Monday, revised",
value={"slot_id": "slot-1"},
metadata={"source": "scheduling"},
),
)
self.assertEqual(poll.options[0].label, "Monday")
self.assertEqual(len(response.answers), 2)
if __name__ == "__main__":
unittest.main()