from __future__ import annotations import unittest from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from govoplan_core.core.poll import ( PollCapabilityError, PollOptionOrderCommand, 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 from govoplan_poll.backend.schemas import PollAnswerInput, PollCreateRequest, PollOptionInput, PollSubmitResponseRequest from govoplan_poll.backend.service import create_poll, submit_poll_response class PollResponseEditingTests(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__], ) 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=[PollResponse.__table__, PollOption.__table__, Poll.__table__], ) self.engine.dispose() def _poll(self, *, allow_response_update: bool = True) -> Poll: return create_poll( self.session, tenant_id="tenant-1", user_id="organizer-1", payload=PollCreateRequest( title="Availability", kind="availability", status="open", min_choices=1, max_choices=2, allow_response_update=allow_response_update, options=[ PollOptionInput( key="slot-1", label="Monday", value={"slot_id": "slot-1"}, metadata={"source": "scheduling"}, ), PollOptionInput( key="slot-2", label="Tuesday", value={"slot_id": "slot-2"}, ), ], ), ) def _submit_both(self, poll: Poll, respondent_id: str) -> PollResponse: return submit_poll_response( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollSubmitResponseRequest( respondent_id=respondent_id, answers=[ PollAnswerInput(option_id=poll.options[0].id, value="available"), PollAnswerInput(option_id=poll.options[1].id, value="maybe"), ], ), ) def test_provider_projects_answers_and_option_change_invalidates_only_that_option(self) -> None: poll = self._poll() first = self._submit_both(poll, "person-1") second = self._submit_both(poll, "person-2") provider = SqlPollSchedulingProvider() self.assertIsInstance(provider, PollSchedulingProvider) projected = provider.list_responses( self.session, tenant_id="tenant-1", 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, 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(option_ref.id, poll.options[0].id) self.assertEqual(first.answers, [{"option_id": poll.options[1].id, "option_key": "slot-2", "value": "maybe"}]) self.assertEqual(second.answers, [{"option_id": poll.options[1].id, "option_key": "slot-2", "value": "maybe"}]) # Re-answer, then replay the exact option snapshot. No response data is # invalidated because the option did not actually change. first = self._submit_both(poll, "person-1") provider.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(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) def test_reorder_preserves_option_identity_and_existing_answers(self) -> None: poll = self._poll(allow_response_update=False) response = self._submit_both(poll, "person-1") first_id, second_id = (option.id for option in poll.options) original_answers = [dict(answer) for answer in response.answers] provider = SqlPollSchedulingProvider() reordered = provider.reorder_options( self.session, tenant_id="tenant-1", poll_id=poll.id, command=PollOptionOrderCommand(option_ids=(second_id, first_id)), ) self.assertEqual( [(option.id, option.position) for option in reordered.options], [(second_id, 0), (first_id, 1)], ) self.assertEqual(response.answers, original_answers) self.assertIsNone(response.deleted_at) replayed = provider.reorder_options( self.session, tenant_id="tenant-1", poll_id=poll.id, command=PollOptionOrderCommand(option_ids=(second_id, first_id)), ) self.assertEqual(replayed.options, reordered.options) self.assertEqual(response.answers, original_answers) def test_reorder_rejects_partial_or_duplicate_active_option_order(self) -> None: poll = self._poll() first_id, second_id = (option.id for option in poll.options) provider = SqlPollSchedulingProvider() with self.assertRaisesRegex(PollCapabilityError, "every active option"): provider.reorder_options( self.session, tenant_id="tenant-1", poll_id=poll.id, command=PollOptionOrderCommand(option_ids=(first_id,)), ) with self.assertRaisesRegex(PollCapabilityError, "duplicate"): provider.reorder_options( self.session, tenant_id="tenant-1", poll_id=poll.id, command=PollOptionOrderCommand(option_ids=(first_id, first_id)), ) self.assertEqual( [(option.id, option.position) for option in poll.options], [(first_id, 0), (second_id, 1)], ) if __name__ == "__main__": unittest.main()