diff --git a/src/govoplan_poll/backend/capabilities.py b/src/govoplan_poll/backend/capabilities.py new file mode 100644 index 0000000..90d832c --- /dev/null +++ b/src/govoplan_poll/backend/capabilities.py @@ -0,0 +1,263 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence + +from govoplan_core.core.poll import ( + PollCapabilityError, + PollCreateCommand, + PollInvitationCommand, + PollInvitationRef, + PollOptionRef, + PollRef, + PollResponseRef, + PollSchedulingProvider, + PollSubmitResponseCommand, + PollUpdateCommand, +) +from govoplan_poll.backend.schemas import ( + PollAnswerInput, + PollCreateRequest, + PollDecisionRequest, + PollInvitationCreateRequest, + PollOptionInput, + PollSubmitResponseRequest, +) +from govoplan_poll.backend.service import ( + PollError, + close_poll, + create_poll, + create_poll_invitation, + decide_poll, + get_poll, + list_poll_responses, + open_poll, + poll_result_summary_by_id, + submit_poll_response, +) + + +def _poll_ref(poll: object) -> PollRef: + return PollRef( + id=poll.id, + status=poll.status, + options=tuple(PollOptionRef(id=option.id, position=option.position) for option in poll.options), + ) + + +def _response_invitation_id(response: object) -> str | None: + value = (response.metadata_ or {}).get("invitation_id") + return value if isinstance(value, str) else None + + +class SqlPollSchedulingProvider(PollSchedulingProvider): + def create_poll( + self, + session: object, + *, + tenant_id: str, + user_id: str | None, + command: PollCreateCommand, + ) -> PollRef: + payload = PollCreateRequest( + title=command.title, + description=command.description, + kind=command.kind, + status=command.status, + visibility=command.visibility, + result_visibility=command.result_visibility, + context_module=command.context_module, + context_resource_type=command.context_resource_type, + context_resource_id=command.context_resource_id, + workflow_state=command.workflow_state, + workflow_steps=[dict(step) for step in command.workflow_steps], + allow_anonymous=command.allow_anonymous, + allow_response_update=command.allow_response_update, + min_choices=command.min_choices, + max_choices=command.max_choices, + opens_at=command.opens_at, + closes_at=command.closes_at, + options=[ + PollOptionInput( + key=option.key, + label=option.label, + description=option.description, + value=dict(option.value) if option.value is not None else None, + metadata=dict(option.metadata), + ) + for option in command.options + ], + metadata=dict(command.metadata), + ) + try: + poll = create_poll(session, tenant_id=tenant_id, user_id=user_id, payload=payload) + except PollError as exc: + raise PollCapabilityError(str(exc)) from exc + return _poll_ref(poll) + + def create_invitation( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + command: PollInvitationCommand, + ) -> PollInvitationRef: + payload = PollInvitationCreateRequest( + respondent_id=command.respondent_id, + respondent_label=command.respondent_label, + email=command.email, + expires_at=command.expires_at, + metadata=dict(command.metadata), + ) + try: + invitation, token = create_poll_invitation( + session, + tenant_id=tenant_id, + poll_id=poll_id, + payload=payload, + ) + except PollError as exc: + raise PollCapabilityError(str(exc)) from exc + return PollInvitationRef(id=invitation.id, token=token) + + def submit_response( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + command: PollSubmitResponseCommand, + ) -> PollResponseRef: + payload = PollSubmitResponseRequest( + respondent_id=command.respondent_id, + respondent_label=command.respondent_label, + answers=[ + PollAnswerInput( + option_id=answer.option_id, + option_key=answer.option_key, + value=answer.value, + rank=answer.rank, + ) + for answer in command.answers + ], + metadata=dict(command.metadata), + ) + try: + response = submit_poll_response( + session, + tenant_id=tenant_id, + poll_id=poll_id, + payload=payload, + ) + except PollError as exc: + raise PollCapabilityError(str(exc)) from exc + return PollResponseRef( + invitation_id=_response_invitation_id(response), + submitted_at=response.submitted_at, + respondent_id=response.respondent_id, + ) + + def update_poll( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + command: PollUpdateCommand, + ) -> PollRef: + try: + poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id) + except PollError as exc: + raise PollCapabilityError(str(exc)) from exc + poll.title = command.title + poll.description = command.description + poll.visibility = command.visibility + poll.result_visibility = command.result_visibility + poll.allow_anonymous = command.allow_anonymous + poll.allow_response_update = command.allow_response_update + poll.closes_at = command.closes_at + session.flush() + return _poll_ref(poll) + + def open_poll(self, session: object, *, tenant_id: str, poll_id: str) -> PollRef: + return self._transition(open_poll, session, tenant_id=tenant_id, poll_id=poll_id) + + def close_poll(self, session: object, *, tenant_id: str, poll_id: str) -> PollRef: + return self._transition(close_poll, session, tenant_id=tenant_id, poll_id=poll_id) + + def decide_poll( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + option_id: str | None, + ) -> PollRef: + try: + poll = decide_poll( + session, + tenant_id=tenant_id, + poll_id=poll_id, + payload=PollDecisionRequest(option_id=option_id), + ) + except PollError as exc: + raise PollCapabilityError(str(exc)) from exc + return _poll_ref(poll) + + def get_poll(self, session: object, *, tenant_id: str, poll_id: str) -> PollRef: + try: + return _poll_ref(get_poll(session, tenant_id=tenant_id, poll_id=poll_id)) + except PollError as exc: + raise PollCapabilityError(str(exc)) from exc + + def set_workflow_context( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + workflow_state: str, + workflow_steps: Sequence[Mapping[str, object]], + context_module: str, + context_resource_type: str, + context_resource_id: str, + ) -> PollRef: + try: + poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id) + except PollError as exc: + raise PollCapabilityError(str(exc)) from exc + poll.workflow_state = workflow_state + poll.workflow_steps = [dict(step) for step in workflow_steps] + poll.context_module = context_module + poll.context_resource_type = context_resource_type + poll.context_resource_id = context_resource_id + session.flush() + return _poll_ref(poll) + + def result_summary(self, session: object, *, tenant_id: str, poll_id: str) -> Mapping[str, object]: + try: + return poll_result_summary_by_id(session, tenant_id=tenant_id, poll_id=poll_id) + except PollError as exc: + raise PollCapabilityError(str(exc)) from exc + + def list_responses(self, session: object, *, tenant_id: str, poll_id: str) -> tuple[PollResponseRef, ...]: + try: + responses = list_poll_responses(session, tenant_id=tenant_id, poll_id=poll_id) + except PollError as exc: + raise PollCapabilityError(str(exc)) from exc + return tuple( + PollResponseRef( + invitation_id=_response_invitation_id(response), + submitted_at=response.submitted_at, + respondent_id=response.respondent_id, + ) + for response in responses + ) + + @staticmethod + def _transition(callback, session: object, *, tenant_id: str, poll_id: str) -> PollRef: + try: + poll = callback(session, tenant_id=tenant_id, poll_id=poll_id) + except PollError as exc: + raise PollCapabilityError(str(exc)) from exc + return _poll_ref(poll) diff --git a/src/govoplan_poll/backend/manifest.py b/src/govoplan_poll/backend/manifest.py index 1010625..bfdd867 100644 --- a/src/govoplan_poll/backend/manifest.py +++ b/src/govoplan_poll/backend/manifest.py @@ -14,6 +14,7 @@ from govoplan_core.core.modules import ( RoleTemplate, ) from govoplan_core.db.base import Base +from govoplan_core.core.poll import CAPABILITY_POLL_SCHEDULING from govoplan_poll.backend.db import models as poll_models # noqa: F401 - populate Poll ORM metadata MODULE_ID = "poll" @@ -99,6 +100,13 @@ def _poll_router(_context: ModuleContext): return router +def _poll_scheduling_provider(context: ModuleContext) -> object: + del context + from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider + + return SqlPollSchedulingProvider() + + manifest = ModuleManifest( id=MODULE_ID, name=MODULE_NAME, @@ -117,6 +125,7 @@ manifest = ModuleManifest( role_templates=ROLE_TEMPLATES, route_factory=_poll_router, tenant_summary_providers=(_tenant_summary,), + capability_factories={CAPABILITY_POLL_SCHEDULING: _poll_scheduling_provider}, migration_spec=MigrationSpec( module_id=MODULE_ID, metadata=Base.metadata, diff --git a/tests/test_service.py b/tests/test_service.py index 27839e2..b16d8f1 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -1,11 +1,14 @@ from __future__ import annotations import unittest +from datetime import datetime, timezone from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from govoplan_core.db.base import Base +from govoplan_core.core.poll import PollResponseRef, PollResponseSubmissionProvider, PollSchedulingProvider +from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider from govoplan_poll.backend.db.models import Poll, PollInvitation, PollOption, PollResponse from govoplan_poll.backend.schemas import ( PollAnswerInput, @@ -43,6 +46,15 @@ class PollServiceTests(unittest.TestCase): ) self.engine.dispose() + def test_response_projection_and_submission_extension_are_backward_compatible(self) -> None: + submitted_at = datetime(2026, 7, 20, tzinfo=timezone.utc) + legacy_ref = PollResponseRef(None, submitted_at) + provider = SqlPollSchedulingProvider() + + self.assertIsNone(legacy_ref.respondent_id) + self.assertIsInstance(provider, PollSchedulingProvider) + self.assertIsInstance(provider, PollResponseSubmissionProvider) + def test_single_choice_response_can_update_existing_respondent(self) -> None: poll = create_poll( self.session, @@ -66,13 +78,24 @@ class PollServiceTests(unittest.TestCase): self.session, tenant_id="tenant-1", poll_id=poll.id, - payload=PollSubmitResponseRequest(respondent_id="person-1", answers=[PollAnswerInput(option_key="b")]), + payload=PollSubmitResponseRequest( + respondent_id="person-1", + answers=[PollAnswerInput(option_key="b")], + metadata={"invitation_id": 42}, + ), ) 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"]]) + response_ref = SqlPollSchedulingProvider().list_responses( + self.session, + tenant_id="tenant-1", + poll_id=poll.id, + )[0] + self.assertIsNone(response_ref.invitation_id) + self.assertEqual(response_ref.respondent_id, "person-1") def test_anonymous_response_requires_poll_policy(self) -> None: poll = create_poll( @@ -267,13 +290,24 @@ class PollServiceTests(unittest.TestCase): response = submit_poll_response_with_token( self.session, token=token, - payload=PollSubmitResponseRequest(answers=[PollAnswerInput(option_key="yes")]), + payload=PollSubmitResponseRequest( + answers=[PollAnswerInput(option_key="yes")], + metadata={"invitation_id": "forged-invitation", "client_note": "kept"}, + ), ) 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.assertEqual(response.metadata_["client_note"], "kept") self.assertIsNotNone(invitation.last_used_at) + response_ref = SqlPollSchedulingProvider().list_responses( + self.session, + tenant_id="tenant-1", + poll_id=poll.id, + )[0] + self.assertEqual(response_ref.invitation_id, invitation.id) + self.assertEqual(response_ref.respondent_id, f"invitation:{invitation.id}") if __name__ == "__main__":