govoplan-voting/tests/test_voting.py

from __future__ import annotations

from dataclasses import dataclass, replace
import unittest

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from govoplan_core.core.voting import (
    VotingBallotCreateCommand,
    VotingCastCommand,
    VotingElector,
    VotingOption,
)
from govoplan_core.db.base import Base
from govoplan_voting.backend.db.models import VotingCastRecord
from govoplan_voting.backend.service import SqlVotingBallots, VotingStoreError


@dataclass
class Principal:
    tenant_id: str
    account_id: str


def command(*, assurance: str = "recorded", provider_id: str | None = None):
    return VotingBallotCreateCommand(
        title="Budget choice",
        method="single_choice",
        assurance_profile=assurance,
        options=(VotingOption("a", "Option A"), VotingOption("b", "Option B")),
        electorate=(
            VotingElector("alice", "Alice", 2, {"source": "mandate:1"}),
            VotingElector("bob", "Bob", 1, {"source": "mandate:2"}),
        ),
        quorum_weight=2,
        threshold_numerator=1,
        threshold_denominator=2,
        provider_id=provider_id,
        provider_ballot_ref="provider-ballot-1" if provider_id else None,
    )


class VotingTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite+pysqlite:///:memory:")
        Base.metadata.create_all(self.engine)
        self.Session = sessionmaker(bind=self.engine)
        self.service = SqlVotingBallots()
        self.manager = Principal("tenant-1", "manager")

    def tearDown(self) -> None:
        Base.metadata.drop_all(self.engine)
        self.engine.dispose()

    def create_open(self, session):
        created = self.service.create_ballot(
            session,
            self.manager,
            command=command(),
            idempotency_key="create-1",
        )
        opened = self.service.open_ballot(
            session,
            self.manager,
            ballot_id=created.id,
            expected_revision=created.revision,
            idempotency_key="open-1",
        )
        return opened

    def test_recorded_ballot_freezes_replaces_tallies_and_certifies(self) -> None:
        with self.Session() as session:
            opened = self.create_open(session)
            detail = self.service.get_ballot(session, self.manager, ballot_id=opened.id)
            self.assertEqual(64, len(str(detail["definition_sha256"])))
            self.assertEqual(64, len(str(detail["electorate_sha256"])))

            first = self.service.cast_ballot(
                session,
                Principal("tenant-1", "alice"),
                ballot_id=opened.id,
                command=VotingCastCommand(("a",), "cast-alice-1"),
            )
            replay = self.service.cast_ballot(
                session,
                Principal("tenant-1", "alice"),
                ballot_id=opened.id,
                command=VotingCastCommand(("a",), "cast-alice-1"),
            )
            self.assertEqual(first.receipt_sha256, replay.receipt_sha256)
            self.assertTrue(replay.replayed)

            replaced = self.service.cast_ballot(
                session,
                Principal("tenant-1", "alice"),
                ballot_id=opened.id,
                command=VotingCastCommand(("b",), "cast-alice-2"),
            )
            self.assertTrue(replaced.replaced_previous)
            self.service.cast_ballot(
                session,
                Principal("tenant-1", "bob"),
                ballot_id=opened.id,
                command=VotingCastCommand(("b",), "cast-bob-1"),
            )
            result = self.service.close_ballot(
                session,
                self.manager,
                ballot_id=opened.id,
                expected_revision=opened.revision,
                idempotency_key="close-1",
            )
            self.assertEqual({"a": 0, "b": 2}, dict(result.counts))
            self.assertEqual({"a": 0, "b": 3}, dict(result.weighted_counts))
            self.assertTrue(result.quorum_met)
            self.assertTrue(result.threshold_met)
            certified = self.service.certify_ballot(
                session,
                self.manager,
                ballot_id=opened.id,
                expected_revision=result.revision,
                evidence=(),
                idempotency_key="certify-1",
            )
            self.assertEqual("certified", certified.state)
            events = self.service.history(session, self.manager, ballot_id=opened.id)
            self.assertEqual(
                [
                    "ballot.created",
                    "ballot.opened",
                    "ballot.vote_cast",
                    "ballot.vote_cast",
                    "ballot.vote_cast",
                    "ballot.closed",
                    "ballot.certified",
                ],
                [item["event_type"] for item in events],
            )
            active = (
                session.query(VotingCastRecord)
                .filter(
                    VotingCastRecord.ballot_id == opened.id,
                    VotingCastRecord.superseded_at.is_(None),
                )
                .count()
            )
            self.assertEqual(2, active)

    def test_eligibility_tenant_and_revision_are_enforced(self) -> None:
        with self.Session() as session:
            opened = self.create_open(session)
            with self.assertRaisesRegex(VotingStoreError, "frozen electorate"):
                self.service.cast_ballot(
                    session,
                    Principal("tenant-1", "mallory"),
                    ballot_id=opened.id,
                    command=VotingCastCommand(("a",), "cast-mallory"),
                )
            self.assertIsNone(
                self.service.get_ballot(
                    session,
                    Principal("tenant-2", "manager"),
                    ballot_id=opened.id,
                )
            )
            with self.assertRaisesRegex(VotingStoreError, "revision conflict"):
                self.service.close_ballot(
                    session,
                    self.manager,
                    ballot_id=opened.id,
                    expected_revision=1,
                    idempotency_key="close-stale",
                )

    def test_secret_profile_fails_closed_without_provider(self) -> None:
        with self.Session() as session:
            created = self.service.create_ballot(
                session,
                self.manager,
                command=command(assurance="secret", provider_id="certified"),
                idempotency_key="create-secret",
            )
            with self.assertRaisesRegex(
                VotingStoreError, "requires an available external provider"
            ):
                self.service.open_ballot(
                    session,
                    self.manager,
                    ballot_id=created.id,
                    expected_revision=created.revision,
                    idempotency_key="open-secret",
                )

    def test_idempotency_key_cannot_change_command(self) -> None:
        with self.Session() as session:
            self.service.create_ballot(
                session,
                self.manager,
                command=command(),
                idempotency_key="create-replay",
            )
            changed = replace(command(), title="Different")
            with self.assertRaisesRegex(VotingStoreError, "idempotency key"):
                self.service.create_ballot(
                    session,
                    self.manager,
                    command=changed,
                    idempotency_key="create-replay",
                )


if __name__ == "__main__":
    unittest.main()