Add initial polling backend logic

tests/test_manifest.py

@@ -16,6 +16,8 @@ class PollManifestTests(unittest.TestCase):
         self.assertIn("access", manifest.optional_dependencies)
         self.assertFalse(manifest.required_capabilities)
         self.assertIn("auth.principalResolver", manifest.optional_capabilities)
+        self.assertIsNotNone(manifest.route_factory)
+        self.assertIsNotNone(manifest.migration_spec)
         self.assertIn("poll.availability_matrix", {interface.name for interface in manifest.provides_interfaces})
         self.assertIn("poll:response:write", {permission.scope for permission in manifest.permissions})
 

tests/test_service.py

@@ -0,0 +1,181 @@
+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, PollOption, PollResponse
+from govoplan_poll.backend.schemas import PollAnswerInput, PollCreateRequest, PollOptionInput, PollSubmitResponseRequest
+from govoplan_poll.backend.service import PollError, create_poll, open_poll, poll_result_summary_by_id, submit_poll_response
+
+
+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__])
+        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 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_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})
+
+
+if __name__ == "__main__":
+    unittest.main()