Vote storage through adapter

Author: pilz

news/vote-storage.feature

@@ -0,0 +1 @@
+Add an annotation-backed `IVoteStorage` adapter for `Poll` with a Python API to cast, read, remove, and tally votes.

src/experimental/doodle/adapters/__init__.py

@@ -0,0 +1,8 @@
+<configure
+    xmlns="http://namespaces.zope.org/zope"
+    i18n_domain="experimental.doodle"
+    >
+
+  <adapter factory=".vote_storage.VoteStorage" />
+
+</configure>

src/experimental/doodle/adapters/configure.zcml

@@ -0,0 +1,83 @@
+"""Annotation-backed vote storage adapter for ``Poll`` objects."""
+
+from collections.abc import Sequence
+from experimental.doodle.content.poll import IPoll
+from experimental.doodle.interfaces import IVoteStorage
+from persistent.list import PersistentList
+from persistent.mapping import PersistentMapping
+from zope.annotation.interfaces import IAnnotations
+from zope.component import adapter
+from zope.interface import implementer
+
+
+ANNOTATION_KEY = "experimental.doodle.votes"
+
+
+@implementer(IVoteStorage)
+@adapter(IPoll)
+class VoteStorage:
+    """Store and aggregate votes as an annotation on a Poll."""
+
+    def __init__(self, context):
+        self.context = context
+        annotations = IAnnotations(context)
+        if ANNOTATION_KEY not in annotations:
+            annotations[ANNOTATION_KEY] = PersistentMapping()
+        self._votes: PersistentMapping = annotations[ANNOTATION_KEY]
+
+    # ----- helpers -----------------------------------------------------
+
+    @property
+    def _option_count(self) -> int:
+        return len(self.context.options or [])
+
+    @staticmethod
+    def _clean_name(name: str) -> str:
+        if not isinstance(name, str):
+            raise ValueError("Participant name must be a string.")
+        cleaned = name.strip()
+        if not cleaned:
+            raise ValueError("Participant name must not be empty.")
+        return cleaned
+
+    # ----- public API --------------------------------------------------
+
+    def cast_vote(self, name: str, votes: Sequence[bool]) -> None:
+        cleaned = self._clean_name(name)
+        votes_list = list(votes)
+        expected = self._option_count
+        if len(votes_list) != expected:
+            raise ValueError(
+                f"Expected {expected} votes (one per option), got {len(votes_list)}."
+            )
+        self._votes[cleaned] = PersistentList(bool(v) for v in votes_list)
+
+    def get_vote(self, name: str) -> list[bool] | None:
+        cleaned = self._clean_name(name)
+        stored = self._votes.get(cleaned)
+        if stored is None:
+            return None
+        return list(stored)
+
+    def get_votes(self) -> dict[str, list[bool]]:
+        return {name: list(votes) for name, votes in self._votes.items()}
+
+    def remove_vote(self, name: str) -> bool:
+        cleaned = self._clean_name(name)
+        if cleaned in self._votes:
+            del self._votes[cleaned]
+            return True
+        return False
+
+    def participants(self) -> list[str]:
+        return sorted(self._votes.keys())
+
+    def tally(self) -> list[int]:
+        counts = [0] * self._option_count
+        for votes in self._votes.values():
+            for index, choice in enumerate(votes):
+                if index >= len(counts):
+                    break
+                if choice:
+                    counts[index] += 1
+        return counts

src/experimental/doodle/adapters/vote_storage.py

@@ -17,6 +17,7 @@
   <include file="permissions.zcml" />
 
   <include package=".content" />
+  <include package=".adapters" />
   <include package=".controlpanels" />
   <include package=".indexers" />
   <include package=".vocabularies" />

src/experimental/doodle/configure.zcml

@@ -1,7 +1,50 @@
 """Module where all interfaces, events and exceptions live."""
 
+from zope.interface import Interface
 from zope.publisher.interfaces.browser import IDefaultBrowserLayer
 
 
 class IBrowserLayer(IDefaultBrowserLayer):
     """Marker interface that defines a browser layer."""
+
+
+class IVoteStorage(Interface):
+    """Storage adapter for votes on a Poll.
+
+    Votes are persisted as an annotation on the adapted Poll. Each entry
+    maps a participant's display name to a list of booleans whose length
+    equals ``len(poll.options)`` at write time.
+    """
+
+    def cast_vote(name, votes):
+        """Record ``votes`` for the participant ``name``.
+
+        ``name`` is stripped of surrounding whitespace; case is preserved.
+        Casting again with the same name overwrites the previous vote.
+
+        Raises ``ValueError`` if ``name`` is empty after stripping, or if
+        ``votes`` has a different length than the poll's options.
+        """
+
+    def get_vote(name):
+        """Return the votes cast by ``name`` as a plain ``list[bool]``,
+        or ``None`` if the participant has not voted yet."""
+
+    def get_votes():
+        """Return a plain ``dict[str, list[bool]]`` snapshot of all votes.
+
+        Mutating the returned mapping or its lists does not affect the
+        stored state.
+        """
+
+    def remove_vote(name):
+        """Remove ``name``'s vote. Return ``True`` if a vote was removed,
+        ``False`` if the participant had not voted."""
+
+    def participants():
+        """Return participant names that have cast a vote, sorted
+        alphabetically."""
+
+    def tally():
+        """Return a ``list[int]`` of ``yes`` counts, parallel to
+        ``poll.options``."""

src/experimental/doodle/interfaces.py

@@ -0,0 +1,167 @@
+"""Tests for the VoteStorage adapter."""
+
+from datetime import datetime
+from datetime import timedelta
+from datetime import timezone
+from experimental.doodle.adapters.vote_storage import ANNOTATION_KEY
+from experimental.doodle.adapters.vote_storage import VoteStorage
+from experimental.doodle.interfaces import IVoteStorage
+from plone import api
+from plone.app.testing import setRoles
+from plone.app.testing import TEST_USER_ID
+from zope.annotation.interfaces import IAnnotations
+
+import pytest
+
+
+@pytest.fixture
+def three_slots():
+    """Three timezone-aware datetimes in the future."""
+    base = datetime.now(tz=timezone.utc) + timedelta(days=1)
+    return [base, base + timedelta(hours=1), base + timedelta(hours=2)]
+
+
+@pytest.fixture
+def poll(integration, three_slots):
+    """A Poll with three options inside the test portal."""
+    portal = integration["portal"]
+    setRoles(portal, TEST_USER_ID, ["Manager"])
+    return api.content.create(
+        container=portal,
+        type="Poll",
+        title="Team lunch",
+        options=three_slots,
+    )
+
+
+class TestAdapterLookup:
+    """``IVoteStorage`` resolves to ``VoteStorage`` for a Poll."""
+
+    def test_adapter_returns_vote_storage(self, poll):
+        storage = IVoteStorage(poll)
+        assert isinstance(storage, VoteStorage)
+
+    def test_adapter_uses_annotation_key(self, poll):
+        IVoteStorage(poll)  # triggers annotation initialization
+        annotations = IAnnotations(poll)
+        assert ANNOTATION_KEY in annotations
+
+
+class TestCastVote:
+    """Recording votes."""
+
+    def test_cast_vote_stores_choices(self, poll):
+        storage = IVoteStorage(poll)
+        storage.cast_vote("Alice", [True, False, True])
+        assert storage.get_vote("Alice") == [True, False, True]
+
+    def test_cast_vote_is_idempotent_overwrite(self, poll):
+        storage = IVoteStorage(poll)
+        storage.cast_vote("Alice", [True, False, True])
+        storage.cast_vote("Alice", [False, False, False])
+        assert storage.get_vote("Alice") == [False, False, False]
+        assert storage.participants() == ["Alice"]
+
+    def test_cast_vote_strips_whitespace_preserves_case(self, poll):
+        storage = IVoteStorage(poll)
+        storage.cast_vote("  Alice  ", [True, True, True])
+        assert storage.get_vote("Alice") == [True, True, True]
+        assert "Alice" in storage.participants()
+
+    def test_cast_vote_coerces_to_bool(self, poll):
+        storage = IVoteStorage(poll)
+        storage.cast_vote("Alice", [1, 0, "yes"])
+        assert storage.get_vote("Alice") == [True, False, True]
+
+
+class TestCastVoteErrors:
+    """``cast_vote`` rejects invalid input."""
+
+    @pytest.mark.parametrize("bad_name", ["", "   ", "\t\n"])
+    def test_empty_name_raises(self, poll, bad_name):
+        storage = IVoteStorage(poll)
+        with pytest.raises(ValueError):
+            storage.cast_vote(bad_name, [True, False, True])
+
+    def test_non_string_name_raises(self, poll):
+        storage = IVoteStorage(poll)
+        with pytest.raises(ValueError):
+            storage.cast_vote(None, [True, False, True])
+
+    def test_wrong_length_raises(self, poll):
+        storage = IVoteStorage(poll)
+        with pytest.raises(ValueError):
+            storage.cast_vote("Alice", [True, False])  # poll has 3 options
+
+
+class TestReading:
+    """Reading back stored votes."""
+
+    def test_get_vote_returns_none_for_unknown(self, poll):
+        storage = IVoteStorage(poll)
+        assert storage.get_vote("Nobody") is None
+
+    def test_get_votes_returns_plain_dict_snapshot(self, poll):
+        storage = IVoteStorage(poll)
+        storage.cast_vote("Alice", [True, False, True])
+
+        snapshot = storage.get_votes()
+        assert snapshot == {"Alice": [True, False, True]}
+
+        # Mutating the snapshot must not affect the stored state.
+        snapshot["Alice"][0] = False
+        snapshot["Mallory"] = [True, True, True]
+        assert storage.get_vote("Alice") == [True, False, True]
+        assert "Mallory" not in storage.participants()
+
+    def test_participants_sorted(self, poll):
+        storage = IVoteStorage(poll)
+        storage.cast_vote("Charlie", [True, True, True])
+        storage.cast_vote("Alice", [False, False, False])
+        storage.cast_vote("Bob", [True, False, True])
+
+        assert storage.participants() == ["Alice", "Bob", "Charlie"]
+
+
+class TestRemoveVote:
+    """Removing votes."""
+
+    def test_remove_vote_existing_returns_true(self, poll):
+        storage = IVoteStorage(poll)
+        storage.cast_vote("Alice", [True, False, True])
+
+        assert storage.remove_vote("Alice") is True
+        assert storage.get_vote("Alice") is None
+
+    def test_remove_vote_unknown_returns_false(self, poll):
+        storage = IVoteStorage(poll)
+        assert storage.remove_vote("Nobody") is False
+
+
+class TestTally:
+    """Aggregating yes-counts per option."""
+
+    def test_tally_empty_poll(self, poll):
+        storage = IVoteStorage(poll)
+        assert storage.tally() == [0, 0, 0]
+
+    def test_tally_counts_yes_per_option(self, poll):
+        storage = IVoteStorage(poll)
+        storage.cast_vote("Alice", [True, False, True])
+        storage.cast_vote("Bob", [True, True, False])
+        storage.cast_vote("Charlie", [False, True, True])
+
+        # Option 0: Alice + Bob = 2 yes
+        # Option 1: Bob + Charlie = 2 yes
+        # Option 2: Alice + Charlie = 2 yes
+        assert storage.tally() == [2, 2, 2]
+
+    def test_tally_ignores_extra_entries_when_options_shrink(self, poll):
+        """If the Poll loses an option after a vote, tally truncates."""
+        storage = IVoteStorage(poll)
+        storage.cast_vote("Alice", [True, True, True])
+
+        # Simulate the Poll losing one option.
+        poll.options = poll.options[:2]
+
+        assert storage.tally() == [1, 1]