diff --git a/README.md b/README.md index abf4f79..e6a6f24 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,17 @@ resolution, permission checks, and role templates. Without Access, Poll remains usable for reduced flows such as anonymous participation, signed public links, or participant lists supplied by another adapter. +## Current Backend Slice + +The first backend implementation adds poll storage, option storage, response +storage, validation, result aggregation, tenant summaries, module migrations, +and authenticated API routes for: + +- single-choice, multiple-choice, yes/no, ranked-choice, and availability polls +- opening, closing, and deciding polls +- submitting and updating respondent answers +- listing responses and reading result summaries + ## Development Install ```bash diff --git a/src/govoplan_poll/backend/db/__init__.py b/src/govoplan_poll/backend/db/__init__.py new file mode 100644 index 0000000..1bed0d3 --- /dev/null +++ b/src/govoplan_poll/backend/db/__init__.py @@ -0,0 +1,5 @@ +from __future__ import annotations + +from govoplan_poll.backend.db.models import Poll, PollOption, PollResponse + +__all__ = ["Poll", "PollOption", "PollResponse"] diff --git a/src/govoplan_poll/backend/db/models.py b/src/govoplan_poll/backend/db/models.py new file mode 100644 index 0000000..77e9f92 --- /dev/null +++ b/src/govoplan_poll/backend/db/models.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +import uuid +from datetime import datetime +from typing import Any + +from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, JSON, String, Text, UniqueConstraint, text +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from govoplan_core.db.base import Base, TimestampMixin + + +def new_uuid() -> str: + return str(uuid.uuid4()) + + +class Poll(Base, TimestampMixin): + __tablename__ = "poll_polls" + __table_args__ = ( + Index( + "uq_poll_polls_active_slug", + "tenant_id", + "slug", + unique=True, + sqlite_where=text("deleted_at IS NULL"), + postgresql_where=text("deleted_at IS NULL"), + ), + Index("ix_poll_polls_tenant_status", "tenant_id", "status"), + Index("ix_poll_polls_tenant_kind", "tenant_id", "kind"), + Index("ix_poll_polls_window", "tenant_id", "opens_at", "closes_at"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + slug: Mapped[str] = mapped_column(String(120), nullable=False) + title: Mapped[str] = mapped_column(String(500), nullable=False) + description: Mapped[str | None] = mapped_column(Text) + kind: Mapped[str] = mapped_column(String(40), nullable=False, index=True) + status: Mapped[str] = mapped_column(String(30), default="draft", nullable=False, index=True) + visibility: Mapped[str] = mapped_column(String(30), default="private", nullable=False, index=True) + result_visibility: Mapped[str] = mapped_column(String(30), default="after_close", nullable=False) + allow_anonymous: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) + allow_response_update: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False) + min_choices: Mapped[int] = mapped_column(Integer, default=1, nullable=False) + max_choices: Mapped[int | None] = mapped_column(Integer) + opens_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + closes_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + closed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + decided_option_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True) + created_by_user_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True) + deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + metadata_: Mapped[dict[str, Any] | None] = mapped_column("metadata", JSON, nullable=True) + + options: Mapped[list["PollOption"]] = relationship(back_populates="poll", cascade="all, delete-orphan", order_by="PollOption.position") + responses: Mapped[list["PollResponse"]] = relationship(back_populates="poll", cascade="all, delete-orphan") + + +class PollOption(Base, TimestampMixin): + __tablename__ = "poll_options" + __table_args__ = ( + UniqueConstraint("poll_id", "key", name="uq_poll_options_poll_key"), + Index("ix_poll_options_poll_position", "poll_id", "position"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + poll_id: Mapped[str] = mapped_column(ForeignKey("poll_polls.id", ondelete="CASCADE"), nullable=False, index=True) + key: Mapped[str] = mapped_column(String(120), nullable=False) + label: Mapped[str] = mapped_column(String(500), nullable=False) + description: Mapped[str | None] = mapped_column(Text) + position: Mapped[int] = mapped_column(Integer, default=0, nullable=False) + value: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + metadata_: Mapped[dict[str, Any] | None] = mapped_column("metadata", JSON, nullable=True) + deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + + poll: Mapped[Poll] = relationship(back_populates="options") + + +class PollResponse(Base, TimestampMixin): + __tablename__ = "poll_responses" + __table_args__ = ( + Index("ix_poll_responses_poll_submitted", "poll_id", "submitted_at"), + Index("ix_poll_responses_poll_respondent", "poll_id", "respondent_id"), + Index("ix_poll_responses_tenant_poll", "tenant_id", "poll_id"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + poll_id: Mapped[str] = mapped_column(ForeignKey("poll_polls.id", ondelete="CASCADE"), nullable=False, index=True) + respondent_id: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True) + respondent_label: Mapped[str | None] = mapped_column(String(500), nullable=True) + answers: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False) + submitted_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True) + deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + metadata_: Mapped[dict[str, Any] | None] = mapped_column("metadata", JSON, nullable=True) + + poll: Mapped[Poll] = relationship(back_populates="responses") + + +__all__ = ["Poll", "PollOption", "PollResponse", "new_uuid"] diff --git a/src/govoplan_poll/backend/manifest.py b/src/govoplan_poll/backend/manifest.py index ca28262..50889d0 100644 --- a/src/govoplan_poll/backend/manifest.py +++ b/src/govoplan_poll/backend/manifest.py @@ -1,13 +1,20 @@ from __future__ import annotations +from pathlib import Path + from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER +from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard from govoplan_core.core.modules import ( DocumentationTopic, + MigrationSpec, + ModuleContext, ModuleInterfaceProvider, ModuleManifest, PermissionDefinition, RoleTemplate, ) +from govoplan_core.db.base import Base +from govoplan_poll.backend.db import models as poll_models # noqa: F401 - populate Poll ORM metadata MODULE_ID = "poll" MODULE_NAME = "Poll" @@ -75,6 +82,22 @@ DOCUMENTATION = ( ), ) + +def _tenant_summary(session, tenant_id: str) -> dict[str, int]: + from govoplan_poll.backend.db.models import Poll, PollResponse + + return { + "polls": session.query(Poll).filter(Poll.tenant_id == tenant_id, Poll.deleted_at.is_(None)).count(), + "poll_responses": session.query(PollResponse).filter(PollResponse.tenant_id == tenant_id, PollResponse.deleted_at.is_(None)).count(), + } + + +def _poll_router(_context: ModuleContext): + from govoplan_poll.backend.router import router + + return router + + manifest = ModuleManifest( id=MODULE_ID, name=MODULE_NAME, @@ -89,6 +112,29 @@ manifest = ModuleManifest( ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, + route_factory=_poll_router, + tenant_summary_providers=(_tenant_summary,), + migration_spec=MigrationSpec( + module_id=MODULE_ID, + metadata=Base.metadata, + script_location=str(Path(__file__).with_name("migrations") / "versions"), + retirement_supported=True, + retirement_provider=drop_table_retirement_provider( + poll_models.Poll, + poll_models.PollOption, + poll_models.PollResponse, + label="Poll", + ), + retirement_notes="Destructive retirement drops poll-owned database tables after the installer captures a database snapshot.", + ), + uninstall_guard_providers=( + persistent_table_uninstall_guard( + poll_models.Poll, + poll_models.PollOption, + poll_models.PollResponse, + label="Poll", + ), + ), documentation=DOCUMENTATION, ) diff --git a/src/govoplan_poll/backend/migrations/__init__.py b/src/govoplan_poll/backend/migrations/__init__.py new file mode 100644 index 0000000..9d48db4 --- /dev/null +++ b/src/govoplan_poll/backend/migrations/__init__.py @@ -0,0 +1 @@ +from __future__ import annotations diff --git a/src/govoplan_poll/backend/migrations/versions/1f2e3d4c5b6a_v018_poll_baseline.py b/src/govoplan_poll/backend/migrations/versions/1f2e3d4c5b6a_v018_poll_baseline.py new file mode 100644 index 0000000..29a2fad --- /dev/null +++ b/src/govoplan_poll/backend/migrations/versions/1f2e3d4c5b6a_v018_poll_baseline.py @@ -0,0 +1,120 @@ +"""v0.1.8 poll baseline + +Revision ID: 1f2e3d4c5b6a +Revises: None +Create Date: 2026-07-12 00:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "1f2e3d4c5b6a" +down_revision = None +branch_labels = None +depends_on = "4f2a9c8e7b6d" + + +def upgrade() -> None: + op.create_table( + "poll_polls", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("slug", sa.String(length=120), nullable=False), + sa.Column("title", sa.String(length=500), nullable=False), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("kind", sa.String(length=40), nullable=False), + sa.Column("status", sa.String(length=30), nullable=False), + sa.Column("visibility", sa.String(length=30), nullable=False), + sa.Column("result_visibility", sa.String(length=30), nullable=False), + sa.Column("allow_anonymous", sa.Boolean(), nullable=False), + sa.Column("allow_response_update", sa.Boolean(), nullable=False), + sa.Column("min_choices", sa.Integer(), nullable=False), + sa.Column("max_choices", sa.Integer(), nullable=True), + sa.Column("opens_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("closes_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("closed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("decided_option_id", sa.String(length=36), nullable=True), + sa.Column("created_by_user_id", sa.String(length=36), nullable=True), + sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("metadata", sa.JSON(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id", name=op.f("pk_poll_polls")), + ) + op.create_index(op.f("ix_poll_polls_closes_at"), "poll_polls", ["closes_at"], unique=False) + op.create_index(op.f("ix_poll_polls_created_by_user_id"), "poll_polls", ["created_by_user_id"], unique=False) + op.create_index(op.f("ix_poll_polls_decided_option_id"), "poll_polls", ["decided_option_id"], unique=False) + op.create_index(op.f("ix_poll_polls_deleted_at"), "poll_polls", ["deleted_at"], unique=False) + op.create_index(op.f("ix_poll_polls_kind"), "poll_polls", ["kind"], unique=False) + op.create_index(op.f("ix_poll_polls_opens_at"), "poll_polls", ["opens_at"], unique=False) + op.create_index(op.f("ix_poll_polls_status"), "poll_polls", ["status"], unique=False) + op.create_index(op.f("ix_poll_polls_tenant_id"), "poll_polls", ["tenant_id"], unique=False) + op.create_index(op.f("ix_poll_polls_visibility"), "poll_polls", ["visibility"], unique=False) + op.create_index("ix_poll_polls_tenant_kind", "poll_polls", ["tenant_id", "kind"], unique=False) + op.create_index("ix_poll_polls_tenant_status", "poll_polls", ["tenant_id", "status"], unique=False) + op.create_index("ix_poll_polls_window", "poll_polls", ["tenant_id", "opens_at", "closes_at"], unique=False) + op.create_index( + "uq_poll_polls_active_slug", + "poll_polls", + ["tenant_id", "slug"], + unique=True, + sqlite_where=sa.text("deleted_at IS NULL"), + postgresql_where=sa.text("deleted_at IS NULL"), + ) + + op.create_table( + "poll_options", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("poll_id", sa.String(length=36), nullable=False), + sa.Column("key", sa.String(length=120), nullable=False), + sa.Column("label", sa.String(length=500), nullable=False), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("position", sa.Integer(), nullable=False), + sa.Column("value", sa.JSON(), nullable=True), + sa.Column("metadata", sa.JSON(), nullable=True), + sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["poll_id"], ["poll_polls.id"], name=op.f("fk_poll_options_poll_id_poll_polls"), ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id", name=op.f("pk_poll_options")), + sa.UniqueConstraint("poll_id", "key", name="uq_poll_options_poll_key"), + ) + op.create_index(op.f("ix_poll_options_deleted_at"), "poll_options", ["deleted_at"], unique=False) + op.create_index(op.f("ix_poll_options_poll_id"), "poll_options", ["poll_id"], unique=False) + op.create_index("ix_poll_options_poll_position", "poll_options", ["poll_id", "position"], unique=False) + op.create_index(op.f("ix_poll_options_tenant_id"), "poll_options", ["tenant_id"], unique=False) + + op.create_table( + "poll_responses", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("poll_id", sa.String(length=36), nullable=False), + sa.Column("respondent_id", sa.String(length=255), nullable=True), + sa.Column("respondent_label", sa.String(length=500), nullable=True), + sa.Column("answers", sa.JSON(), nullable=False), + sa.Column("submitted_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("metadata", sa.JSON(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["poll_id"], ["poll_polls.id"], name=op.f("fk_poll_responses_poll_id_poll_polls"), ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id", name=op.f("pk_poll_responses")), + ) + op.create_index(op.f("ix_poll_responses_deleted_at"), "poll_responses", ["deleted_at"], unique=False) + op.create_index(op.f("ix_poll_responses_poll_id"), "poll_responses", ["poll_id"], unique=False) + op.create_index("ix_poll_responses_poll_respondent", "poll_responses", ["poll_id", "respondent_id"], unique=False) + op.create_index("ix_poll_responses_poll_submitted", "poll_responses", ["poll_id", "submitted_at"], unique=False) + op.create_index(op.f("ix_poll_responses_respondent_id"), "poll_responses", ["respondent_id"], unique=False) + op.create_index(op.f("ix_poll_responses_submitted_at"), "poll_responses", ["submitted_at"], unique=False) + op.create_index(op.f("ix_poll_responses_tenant_id"), "poll_responses", ["tenant_id"], unique=False) + op.create_index("ix_poll_responses_tenant_poll", "poll_responses", ["tenant_id", "poll_id"], unique=False) + + +def downgrade() -> None: + op.drop_table("poll_responses") + op.drop_table("poll_options") + op.drop_table("poll_polls") diff --git a/src/govoplan_poll/backend/migrations/versions/__init__.py b/src/govoplan_poll/backend/migrations/versions/__init__.py new file mode 100644 index 0000000..9d48db4 --- /dev/null +++ b/src/govoplan_poll/backend/migrations/versions/__init__.py @@ -0,0 +1 @@ +from __future__ import annotations diff --git a/src/govoplan_poll/backend/router.py b/src/govoplan_poll/backend/router.py new file mode 100644 index 0000000..074f454 --- /dev/null +++ b/src/govoplan_poll/backend/router.py @@ -0,0 +1,191 @@ +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from sqlalchemy.orm import Session + +from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope +from govoplan_core.db.session import get_session +from govoplan_poll.backend.manifest import READ_SCOPE, RESPOND_SCOPE, WRITE_SCOPE +from govoplan_poll.backend.schemas import ( + PollCreateRequest, + PollDecisionRequest, + PollListResponse, + PollResponse, + PollResponseItem, + PollResponseListResponse, + PollResultSummaryResponse, + PollStatusResponse, + PollSubmitResponseRequest, + PollUpdateRequest, +) +from govoplan_poll.backend.service import ( + PollError, + close_poll, + create_poll, + decide_poll, + get_poll, + list_poll_responses, + list_polls, + open_poll, + poll_response, + poll_response_item, + poll_result_summary_by_id, + submit_poll_response, + update_poll, +) + + +router = APIRouter(prefix="/poll", tags=["poll"]) + + +def _require_scope(principal: ApiPrincipal, scope: str) -> None: + if not has_scope(principal, scope): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {scope}") + + +def _poll_http_error(exc: PollError) -> HTTPException: + if str(exc) == "Poll not found": + return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) + return HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) + + +def _poll_response(poll) -> PollResponse: + return PollResponse.model_validate(poll_response(poll)) + + +@router.get("/polls", response_model=PollListResponse) +def api_list_polls( + status_filter: str | None = Query(default=None, alias="status"), + kind: str | None = None, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PollListResponse: + _require_scope(principal, READ_SCOPE) + polls = list_polls(session, tenant_id=principal.tenant_id, status=status_filter, kind=kind) + return PollListResponse(polls=[_poll_response(poll) for poll in polls]) + + +@router.post("/polls", response_model=PollResponse, status_code=status.HTTP_201_CREATED) +def api_create_poll( + payload: PollCreateRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PollResponse: + _require_scope(principal, WRITE_SCOPE) + try: + poll = create_poll(session, tenant_id=principal.tenant_id, user_id=principal.account_id, payload=payload) + except PollError as exc: + raise _poll_http_error(exc) from exc + return _poll_response(poll) + + +@router.get("/polls/{poll_id}", response_model=PollResponse) +def api_get_poll( + poll_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PollResponse: + _require_scope(principal, READ_SCOPE) + try: + return _poll_response(get_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id)) + except PollError as exc: + raise _poll_http_error(exc) from exc + + +@router.patch("/polls/{poll_id}", response_model=PollResponse) +def api_update_poll( + poll_id: str, + payload: PollUpdateRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PollResponse: + _require_scope(principal, WRITE_SCOPE) + try: + return _poll_response(update_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload)) + except PollError as exc: + raise _poll_http_error(exc) from exc + + +@router.post("/polls/{poll_id}/open", response_model=PollStatusResponse) +def api_open_poll( + poll_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PollStatusResponse: + _require_scope(principal, WRITE_SCOPE) + try: + poll = open_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id) + except PollError as exc: + raise _poll_http_error(exc) from exc + return PollStatusResponse(poll=_poll_response(poll)) + + +@router.post("/polls/{poll_id}/close", response_model=PollStatusResponse) +def api_close_poll( + poll_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PollStatusResponse: + _require_scope(principal, WRITE_SCOPE) + try: + poll = close_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id) + except PollError as exc: + raise _poll_http_error(exc) from exc + return PollStatusResponse(poll=_poll_response(poll)) + + +@router.post("/polls/{poll_id}/decide", response_model=PollStatusResponse) +def api_decide_poll( + poll_id: str, + payload: PollDecisionRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PollStatusResponse: + _require_scope(principal, WRITE_SCOPE) + try: + poll = decide_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload) + except PollError as exc: + raise _poll_http_error(exc) from exc + return PollStatusResponse(poll=_poll_response(poll)) + + +@router.post("/polls/{poll_id}/responses", response_model=PollResponseItem, status_code=status.HTTP_201_CREATED) +def api_submit_poll_response( + poll_id: str, + payload: PollSubmitResponseRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PollResponseItem: + _require_scope(principal, RESPOND_SCOPE) + try: + response = submit_poll_response(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload) + except PollError as exc: + raise _poll_http_error(exc) from exc + return PollResponseItem.model_validate(poll_response_item(response)) + + +@router.get("/polls/{poll_id}/responses", response_model=PollResponseListResponse) +def api_list_poll_responses( + poll_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PollResponseListResponse: + _require_scope(principal, READ_SCOPE) + try: + responses = list_poll_responses(session, tenant_id=principal.tenant_id, poll_id=poll_id) + except PollError as exc: + raise _poll_http_error(exc) from exc + return PollResponseListResponse(responses=[PollResponseItem.model_validate(poll_response_item(response)) for response in responses]) + + +@router.get("/polls/{poll_id}/summary", response_model=PollResultSummaryResponse) +def api_poll_summary( + poll_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> PollResultSummaryResponse: + _require_scope(principal, READ_SCOPE) + try: + return PollResultSummaryResponse.model_validate(poll_result_summary_by_id(session, tenant_id=principal.tenant_id, poll_id=poll_id)) + except PollError as exc: + raise _poll_http_error(exc) from exc diff --git a/src/govoplan_poll/backend/schemas.py b/src/govoplan_poll/backend/schemas.py new file mode 100644 index 0000000..246ea14 --- /dev/null +++ b/src/govoplan_poll/backend/schemas.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + + +PollKind = Literal["single_choice", "multiple_choice", "yes_no", "ranked_choice", "availability"] +PollStatus = Literal["draft", "open", "closed", "decided", "archived"] +PollVisibility = Literal["private", "tenant", "public", "unlisted"] +PollResultVisibility = Literal["organizer", "after_response", "after_close", "public"] +AvailabilityValue = Literal["available", "maybe", "unavailable"] + + +class PollOptionInput(BaseModel): + model_config = ConfigDict(extra="forbid") + + key: str | None = Field(default=None, min_length=1, max_length=120) + label: str = Field(min_length=1, max_length=500) + description: str | None = None + value: dict[str, Any] | None = None + metadata: dict[str, Any] = Field(default_factory=dict) + + +class PollCreateRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + title: str = Field(min_length=1, max_length=500) + slug: str | None = Field(default=None, max_length=120) + description: str | None = None + kind: PollKind = "single_choice" + status: PollStatus = "draft" + visibility: PollVisibility = "private" + result_visibility: PollResultVisibility = "after_close" + allow_anonymous: bool = False + allow_response_update: bool = True + min_choices: int = Field(default=1, ge=0) + max_choices: int | None = Field(default=None, ge=1) + opens_at: datetime | None = None + closes_at: datetime | None = None + options: list[PollOptionInput] = Field(default_factory=list) + metadata: dict[str, Any] = Field(default_factory=dict) + + @model_validator(mode="after") + def validate_window(self) -> "PollCreateRequest": + if self.opens_at is not None and self.closes_at is not None and self.closes_at <= self.opens_at: + raise ValueError("closes_at must be after opens_at") + return self + + +class PollUpdateRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + title: str | None = Field(default=None, min_length=1, max_length=500) + description: str | None = None + visibility: PollVisibility | None = None + result_visibility: PollResultVisibility | None = None + allow_anonymous: bool | None = None + allow_response_update: bool | None = None + min_choices: int | None = Field(default=None, ge=0) + max_choices: int | None = Field(default=None, ge=1) + opens_at: datetime | None = None + closes_at: datetime | None = None + metadata: dict[str, Any] | None = None + + +class PollOptionResponse(BaseModel): + id: str + key: str + label: str + description: str | None = None + position: int + value: dict[str, Any] | None = None + metadata: dict[str, Any] = Field(default_factory=dict) + + +class PollResponseItem(BaseModel): + id: str + respondent_id: str | None = None + respondent_label: str | None = None + answers: list[dict[str, Any]] = Field(default_factory=list) + submitted_at: datetime + created_at: datetime + updated_at: datetime + metadata: dict[str, Any] = Field(default_factory=dict) + + +class PollResponse(BaseModel): + id: str + tenant_id: str + slug: str + title: str + description: str | None = None + kind: str + status: str + visibility: str + result_visibility: str + allow_anonymous: bool + allow_response_update: bool + min_choices: int + max_choices: int | None = None + opens_at: datetime | None = None + closes_at: datetime | None = None + closed_at: datetime | None = None + decided_at: datetime | None = None + decided_option_id: str | None = None + created_by_user_id: str | None = None + created_at: datetime + updated_at: datetime + metadata: dict[str, Any] = Field(default_factory=dict) + options: list[PollOptionResponse] = Field(default_factory=list) + + +class PollListResponse(BaseModel): + polls: list[PollResponse] = Field(default_factory=list) + + +class PollAnswerInput(BaseModel): + model_config = ConfigDict(extra="forbid") + + option_id: str | None = None + option_key: str | None = None + value: Any = None + rank: int | None = Field(default=None, ge=1) + + +class PollSubmitResponseRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + respondent_id: str | None = Field(default=None, max_length=255) + respondent_label: str | None = Field(default=None, max_length=500) + answers: list[PollAnswerInput] = Field(default_factory=list) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class PollOptionResultResponse(BaseModel): + option_id: str + option_key: str + label: str + count: int = 0 + score: int = 0 + values: dict[str, int] = Field(default_factory=dict) + ranks: dict[int, int] = Field(default_factory=dict) + + +class PollResultSummaryResponse(BaseModel): + poll_id: str + kind: str + status: str + response_count: int + option_results: list[PollOptionResultResponse] = Field(default_factory=list) + leading_option_ids: list[str] = Field(default_factory=list) + + +class PollDecisionRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + option_id: str | None = None + option_key: str | None = None + + +class PollStatusResponse(BaseModel): + poll: PollResponse + + +class PollResponseListResponse(BaseModel): + responses: list[PollResponseItem] = Field(default_factory=list) diff --git a/src/govoplan_poll/backend/service.py b/src/govoplan_poll/backend/service.py new file mode 100644 index 0000000..3d9ed66 --- /dev/null +++ b/src/govoplan_poll/backend/service.py @@ -0,0 +1,498 @@ +from __future__ import annotations + +import re +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy.orm import Session + +from govoplan_core.db.base import utcnow +from govoplan_poll.backend.db.models import Poll, PollOption, PollResponse +from govoplan_poll.backend.schemas import ( + PollCreateRequest, + PollDecisionRequest, + PollOptionInput, + PollSubmitResponseRequest, + PollUpdateRequest, +) + + +class PollError(ValueError): + pass + + +POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "ranked_choice", "availability"} +POLL_STATUSES = {"draft", "open", "closed", "decided", "archived"} +CHOICE_POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "ranked_choice"} +AVAILABILITY_VALUES = {"available", "maybe", "unavailable"} +YES_NO_OPTIONS = ( + PollOptionInput(key="yes", label="Yes"), + PollOptionInput(key="no", label="No"), +) + + +def slugify(value: str, *, fallback: str = "poll") -> str: + slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") + return slug or fallback + + +def response_datetime(value: datetime | None) -> datetime | None: + if value is None: + return None + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _now() -> datetime: + return utcnow() + + +def _active_options(poll: Poll) -> list[PollOption]: + return [option for option in poll.options if option.deleted_at is None] + + +def _active_responses(poll: Poll) -> list[PollResponse]: + return [response for response in poll.responses if response.deleted_at is None] + + +def _option_by_id_or_key(poll: Poll, *, option_id: str | None = None, option_key: str | None = None) -> PollOption: + for option in _active_options(poll): + if option_id is not None and option.id == option_id: + return option + if option_key is not None and option.key == option_key: + return option + raise PollError("Unknown poll option") + + +def _poll_slug_exists(session: Session, *, tenant_id: str, slug: str, exclude_poll_id: str | None = None) -> bool: + query = session.query(Poll).filter(Poll.tenant_id == tenant_id, Poll.slug == slug, Poll.deleted_at.is_(None)) + if exclude_poll_id is not None: + query = query.filter(Poll.id != exclude_poll_id) + return session.query(query.exists()).scalar() + + +def _unique_slug(session: Session, *, tenant_id: str, slug: str) -> str: + candidate = slug + suffix = 2 + while _poll_slug_exists(session, tenant_id=tenant_id, slug=candidate): + candidate = f"{slug}-{suffix}" + suffix += 1 + return candidate + + +def _normalize_options(kind: str, options: list[PollOptionInput]) -> list[PollOptionInput]: + normalized = list(options) + if kind == "yes_no" and not normalized: + normalized = list(YES_NO_OPTIONS) + if kind in CHOICE_POLL_KINDS and len(normalized) < 2: + raise PollError(f"{kind} polls require at least two options") + if kind == "availability" and not normalized: + raise PollError("Availability polls require at least one candidate slot option") + seen: set[str] = set() + result: list[PollOptionInput] = [] + for position, option in enumerate(normalized): + key = option.key or slugify(option.label, fallback=f"option-{position + 1}") + if key in seen: + raise PollError(f"Duplicate poll option key: {key}") + seen.add(key) + result.append( + PollOptionInput( + key=key, + label=option.label, + description=option.description, + value=option.value, + metadata=option.metadata, + ) + ) + return result + + +def _validate_choice_bounds(kind: str, min_choices: int, max_choices: int | None, option_count: int) -> tuple[int, int | None]: + if kind in {"single_choice", "yes_no"}: + return 1, 1 + if kind == "ranked_choice": + if min_choices < 1: + min_choices = 1 + if max_choices is None: + max_choices = option_count + if min_choices > option_count: + raise PollError("min_choices cannot be greater than the number of options") + if max_choices is not None: + if max_choices < min_choices: + raise PollError("max_choices cannot be smaller than min_choices") + if max_choices > option_count: + raise PollError("max_choices cannot be greater than the number of options") + return min_choices, max_choices + + +def _ensure_valid_poll_payload(payload: PollCreateRequest) -> tuple[list[PollOptionInput], int, int | None]: + if payload.kind not in POLL_KINDS: + raise PollError(f"Unsupported poll kind: {payload.kind}") + if payload.status not in POLL_STATUSES: + raise PollError(f"Unsupported poll status: {payload.status}") + if payload.opens_at is not None and payload.closes_at is not None and payload.closes_at <= payload.opens_at: + raise PollError("closes_at must be after opens_at") + options = _normalize_options(payload.kind, payload.options) + min_choices, max_choices = _validate_choice_bounds(payload.kind, payload.min_choices, payload.max_choices, len(options)) + return options, min_choices, max_choices + + +def create_poll(session: Session, *, tenant_id: str, user_id: str | None, payload: PollCreateRequest) -> Poll: + options, min_choices, max_choices = _ensure_valid_poll_payload(payload) + base_slug = slugify(payload.slug or payload.title) + poll = Poll( + tenant_id=tenant_id, + slug=_unique_slug(session, tenant_id=tenant_id, slug=base_slug), + title=payload.title, + description=payload.description, + kind=payload.kind, + status=payload.status, + visibility=payload.visibility, + result_visibility=payload.result_visibility, + allow_anonymous=payload.allow_anonymous, + allow_response_update=payload.allow_response_update, + min_choices=min_choices, + max_choices=max_choices, + opens_at=payload.opens_at, + closes_at=payload.closes_at, + created_by_user_id=user_id, + metadata_=payload.metadata, + ) + session.add(poll) + session.flush() + for position, option in enumerate(options): + session.add( + PollOption( + tenant_id=tenant_id, + poll_id=poll.id, + key=option.key or f"option-{position + 1}", + label=option.label, + description=option.description, + position=position, + value=option.value, + metadata_=option.metadata, + ) + ) + session.flush() + return poll + + +def list_polls(session: Session, *, tenant_id: str, status: str | None = None, kind: str | None = None) -> list[Poll]: + query = session.query(Poll).filter(Poll.tenant_id == tenant_id, Poll.deleted_at.is_(None)) + if status: + query = query.filter(Poll.status == status) + if kind: + query = query.filter(Poll.kind == kind) + return query.order_by(Poll.created_at.desc(), Poll.title.asc()).all() + + +def get_poll(session: Session, *, tenant_id: str, poll_id: str) -> Poll: + poll = session.query(Poll).filter(Poll.tenant_id == tenant_id, Poll.id == poll_id, Poll.deleted_at.is_(None)).first() + if poll is None: + raise PollError("Poll not found") + return poll + + +def update_poll(session: Session, *, tenant_id: str, poll_id: str, payload: PollUpdateRequest) -> Poll: + poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id) + if poll.status in {"closed", "decided", "archived"}: + raise PollError("Closed, decided, or archived polls cannot be edited") + for field in ("title", "description", "visibility", "result_visibility", "allow_anonymous", "allow_response_update"): + value = getattr(payload, field) + if value is not None: + setattr(poll, field, value) + if payload.min_choices is not None or payload.max_choices is not None: + min_choices = poll.min_choices if payload.min_choices is None else payload.min_choices + max_choices = poll.max_choices if payload.max_choices is None else payload.max_choices + min_choices, max_choices = _validate_choice_bounds(poll.kind, min_choices, max_choices, len(_active_options(poll))) + poll.min_choices = min_choices + poll.max_choices = max_choices + if payload.opens_at is not None: + poll.opens_at = payload.opens_at + if payload.closes_at is not None: + poll.closes_at = payload.closes_at + if poll.opens_at is not None and poll.closes_at is not None and poll.closes_at <= poll.opens_at: + raise PollError("closes_at must be after opens_at") + if payload.metadata is not None: + poll.metadata_ = payload.metadata + session.flush() + return poll + + +def open_poll(session: Session, *, tenant_id: str, poll_id: str) -> Poll: + poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id) + if poll.status == "archived": + raise PollError("Archived polls cannot be opened") + if poll.closes_at is not None and response_datetime(poll.closes_at) <= _now(): + raise PollError("Poll close time is already in the past") + poll.status = "open" + poll.closed_at = None + session.flush() + return poll + + +def close_poll(session: Session, *, tenant_id: str, poll_id: str) -> Poll: + poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id) + if poll.status == "archived": + raise PollError("Archived polls cannot be closed") + poll.status = "closed" + poll.closed_at = _now() + session.flush() + return poll + + +def decide_poll(session: Session, *, tenant_id: str, poll_id: str, payload: PollDecisionRequest) -> Poll: + poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id) + option = _option_by_id_or_key(poll, option_id=payload.option_id, option_key=payload.option_key) + poll.status = "decided" + poll.decided_option_id = option.id + poll.decided_at = _now() + if poll.closed_at is None: + poll.closed_at = poll.decided_at + session.flush() + return poll + + +def _assert_poll_accepts_responses(poll: Poll, *, now: datetime | None = None) -> None: + now = now or _now() + if poll.status != "open": + raise PollError("Poll is not open") + if poll.opens_at is not None and response_datetime(poll.opens_at) > now: + raise PollError("Poll is not open yet") + if poll.closes_at is not None and response_datetime(poll.closes_at) <= now: + raise PollError("Poll is already closed") + + +def _resolve_answer_option(poll: Poll, answer) -> PollOption: + if answer.option_id is None and answer.option_key is None: + raise PollError("Poll answers must reference an option_id or option_key") + return _option_by_id_or_key(poll, option_id=answer.option_id, option_key=answer.option_key) + + +def _normalize_choice_answers(poll: Poll, payload: PollSubmitResponseRequest) -> list[dict[str, Any]]: + selected: list[PollOption] = [] + seen: set[str] = set() + for answer in payload.answers: + option = _resolve_answer_option(poll, answer) + if option.id in seen: + raise PollError("Poll responses cannot select the same option more than once") + seen.add(option.id) + selected.append(option) + if len(selected) < poll.min_choices: + raise PollError(f"Poll response requires at least {poll.min_choices} option(s)") + if poll.max_choices is not None and len(selected) > poll.max_choices: + raise PollError(f"Poll response allows at most {poll.max_choices} option(s)") + return [{"option_id": option.id, "option_key": option.key, "value": True} for option in selected] + + +def _normalize_ranked_answers(poll: Poll, payload: PollSubmitResponseRequest) -> list[dict[str, Any]]: + selected: list[tuple[int, PollOption]] = [] + seen_options: set[str] = set() + seen_ranks: set[int] = set() + for position, answer in enumerate(payload.answers, start=1): + option = _resolve_answer_option(poll, answer) + rank = answer.rank or position + if option.id in seen_options: + raise PollError("Ranked responses cannot rank the same option more than once") + if rank in seen_ranks: + raise PollError("Ranked responses cannot reuse the same rank") + seen_options.add(option.id) + seen_ranks.add(rank) + selected.append((rank, option)) + if len(selected) < poll.min_choices: + raise PollError(f"Ranked response requires at least {poll.min_choices} ranked option(s)") + if poll.max_choices is not None and len(selected) > poll.max_choices: + raise PollError(f"Ranked response allows at most {poll.max_choices} ranked option(s)") + return [ + {"option_id": option.id, "option_key": option.key, "rank": rank} + for rank, option in sorted(selected, key=lambda item: item[0]) + ] + + +def _normalize_availability_answers(poll: Poll, payload: PollSubmitResponseRequest) -> list[dict[str, Any]]: + selected: list[dict[str, Any]] = [] + seen: set[str] = set() + for answer in payload.answers: + option = _resolve_answer_option(poll, answer) + if option.id in seen: + raise PollError("Availability responses cannot answer the same slot more than once") + if answer.value not in AVAILABILITY_VALUES: + raise PollError("Availability responses must use available, maybe, or unavailable") + seen.add(option.id) + selected.append({"option_id": option.id, "option_key": option.key, "value": answer.value}) + if len(selected) < poll.min_choices: + raise PollError(f"Availability response requires at least {poll.min_choices} slot answer(s)") + if poll.max_choices is not None and len(selected) > poll.max_choices: + raise PollError(f"Availability response allows at most {poll.max_choices} slot answer(s)") + return selected + + +def normalize_response_answers(poll: Poll, payload: PollSubmitResponseRequest) -> list[dict[str, Any]]: + if poll.kind in {"single_choice", "multiple_choice", "yes_no"}: + return _normalize_choice_answers(poll, payload) + if poll.kind == "ranked_choice": + return _normalize_ranked_answers(poll, payload) + if poll.kind == "availability": + return _normalize_availability_answers(poll, payload) + raise PollError(f"Unsupported poll kind: {poll.kind}") + + +def _existing_response(session: Session, *, poll: Poll, respondent_id: str | None) -> PollResponse | None: + if respondent_id is None: + return None + return ( + session.query(PollResponse) + .filter(PollResponse.poll_id == poll.id, PollResponse.respondent_id == respondent_id, PollResponse.deleted_at.is_(None)) + .order_by(PollResponse.submitted_at.desc()) + .first() + ) + + +def submit_poll_response(session: Session, *, tenant_id: str, poll_id: str, payload: PollSubmitResponseRequest) -> PollResponse: + poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id) + _assert_poll_accepts_responses(poll) + if payload.respondent_id is None and not poll.allow_anonymous: + raise PollError("Anonymous responses are not allowed for this poll") + answers = normalize_response_answers(poll, payload) + existing = _existing_response(session, poll=poll, respondent_id=payload.respondent_id) + if existing is not None: + if not poll.allow_response_update: + raise PollError("Response updates are not allowed for this poll") + existing.answers = answers + existing.respondent_label = payload.respondent_label + existing.submitted_at = _now() + existing.metadata_ = payload.metadata + session.flush() + return existing + response = PollResponse( + tenant_id=tenant_id, + poll_id=poll.id, + respondent_id=payload.respondent_id, + respondent_label=payload.respondent_label, + answers=answers, + submitted_at=_now(), + metadata_=payload.metadata, + ) + session.add(response) + session.flush() + return response + + +def poll_option_response(option: PollOption) -> dict[str, Any]: + return { + "id": option.id, + "key": option.key, + "label": option.label, + "description": option.description, + "position": option.position, + "value": option.value, + "metadata": option.metadata_ or {}, + } + + +def poll_response(poll: Poll) -> dict[str, Any]: + return { + "id": poll.id, + "tenant_id": poll.tenant_id, + "slug": poll.slug, + "title": poll.title, + "description": poll.description, + "kind": poll.kind, + "status": poll.status, + "visibility": poll.visibility, + "result_visibility": poll.result_visibility, + "allow_anonymous": poll.allow_anonymous, + "allow_response_update": poll.allow_response_update, + "min_choices": poll.min_choices, + "max_choices": poll.max_choices, + "opens_at": response_datetime(poll.opens_at), + "closes_at": response_datetime(poll.closes_at), + "closed_at": response_datetime(poll.closed_at), + "decided_at": response_datetime(poll.decided_at), + "decided_option_id": poll.decided_option_id, + "created_by_user_id": poll.created_by_user_id, + "created_at": response_datetime(poll.created_at), + "updated_at": response_datetime(poll.updated_at), + "metadata": poll.metadata_ or {}, + "options": [poll_option_response(option) for option in _active_options(poll)], + } + + +def poll_response_item(response: PollResponse) -> dict[str, Any]: + return { + "id": response.id, + "respondent_id": response.respondent_id, + "respondent_label": response.respondent_label, + "answers": response.answers or [], + "submitted_at": response_datetime(response.submitted_at), + "created_at": response_datetime(response.created_at), + "updated_at": response_datetime(response.updated_at), + "metadata": response.metadata_ or {}, + } + + +def list_poll_responses(session: Session, *, tenant_id: str, poll_id: str) -> list[PollResponse]: + get_poll(session, tenant_id=tenant_id, poll_id=poll_id) + return ( + session.query(PollResponse) + .filter(PollResponse.tenant_id == tenant_id, PollResponse.poll_id == poll_id, PollResponse.deleted_at.is_(None)) + .order_by(PollResponse.submitted_at.asc(), PollResponse.id.asc()) + .all() + ) + + +def poll_result_summary(poll: Poll) -> dict[str, Any]: + options = _active_options(poll) + option_results = { + option.id: { + "option_id": option.id, + "option_key": option.key, + "label": option.label, + "count": 0, + "score": 0, + "values": {}, + "ranks": {}, + } + for option in options + } + option_count = len(options) + responses = _active_responses(poll) + for response in responses: + for answer in response.answers or []: + option_id = answer.get("option_id") + if option_id not in option_results: + continue + result = option_results[option_id] + if poll.kind == "ranked_choice": + rank = int(answer.get("rank") or option_count) + result["count"] += 1 + result["score"] += max(option_count - rank + 1, 0) + result["ranks"][rank] = result["ranks"].get(rank, 0) + 1 + elif poll.kind == "availability": + value = str(answer.get("value") or "unavailable") + result["values"][value] = result["values"].get(value, 0) + 1 + if value == "available": + result["count"] += 1 + result["score"] += 2 + elif value == "maybe": + result["score"] += 1 + else: + result["count"] += 1 + result["score"] += 1 + result_list = list(option_results.values()) + score_field = "score" if poll.kind in {"ranked_choice", "availability"} else "count" + best_score = max((int(result[score_field]) for result in result_list), default=0) + leading = [str(result["option_id"]) for result in result_list if best_score > 0 and int(result[score_field]) == best_score] + return { + "poll_id": poll.id, + "kind": poll.kind, + "status": poll.status, + "response_count": len(responses), + "option_results": result_list, + "leading_option_ids": leading, + } + + +def poll_result_summary_by_id(session: Session, *, tenant_id: str, poll_id: str) -> dict[str, Any]: + return poll_result_summary(get_poll(session, tenant_id=tenant_id, poll_id=poll_id)) diff --git a/tests/test_manifest.py b/tests/test_manifest.py index e798731..4244667 100644 --- a/tests/test_manifest.py +++ b/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}) diff --git a/tests/test_service.py b/tests/test_service.py new file mode 100644 index 0000000..56071cf --- /dev/null +++ b/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()