GovOPlaN/govoplan-poll
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(poll): enforce active respondent uniqueness

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-22 02:54:09 +02:00
parent e2e816cb21
commit cac7733bee
3 changed files with 321 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/govoplan_poll/backend/db/models.py
View File

@@ -99,6 +99,18 @@ class PollResponse(Base, TimestampMixin):
__table_args__ = ( __table_args__ = (
Index("ix_poll_responses_poll_submitted", "poll_id", "submitted_at"), Index("ix_poll_responses_poll_submitted", "poll_id", "submitted_at"),
Index("ix_poll_responses_poll_respondent", "poll_id", "respondent_id"), Index("ix_poll_responses_poll_respondent", "poll_id", "respondent_id"),
Index(
"uq_poll_responses_active_respondent",
"poll_id",
"respondent_id",
unique=True,
sqlite_where=text(
"deleted_at IS NULL AND respondent_id IS NOT NULL"
),
postgresql_where=text(
"deleted_at IS NULL AND respondent_id IS NOT NULL"
),
),
Index("ix_poll_responses_tenant_poll", "tenant_id", "poll_id"), Index("ix_poll_responses_tenant_poll", "tenant_id", "poll_id"),
) )

99
src/govoplan_poll/backend/migrations/versions/6e7f8a9b0c1d_v0110_poll_response_uniqueness.py Normal file
View File

@@ -0,0 +1,99 @@
"""v0.1.10 active identified Poll response uniqueness
Revision ID: 6e7f8a9b0c1d
Revises: 5d6e7f8a9b0c
Create Date: 2026-07-22 00:00:00.000000
This migration deduplicates active identified responses before creating the
partial unique index. Run it while Poll response writes are quiesced or while
the platform maintenance lock is held. PostgreSQL additionally takes a table
lock so an unexpected writer cannot enter between cleanup and index creation.
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "6e7f8a9b0c1d"
down_revision = "5d6e7f8a9b0c"
branch_labels = None
depends_on = None
_INDEX_NAME = "uq_poll_responses_active_respondent"
_ACTIVE_IDENTIFIED_PREDICATE = (
"deleted_at IS NULL AND respondent_id IS NOT NULL"
)
def _tombstone_duplicate_active_responses() -> None:
responses = sa.table(
"poll_responses",
sa.column("id", sa.String(length=36)),
sa.column("poll_id", sa.String(length=36)),
sa.column("respondent_id", sa.String(length=255)),
sa.column("submitted_at", sa.DateTime(timezone=True)),
sa.column("deleted_at", sa.DateTime(timezone=True)),
sa.column("updated_at", sa.DateTime(timezone=True)),
)
ranked = (
sa.select(
responses.c.id.label("id"),
sa.func.row_number()
.over(
partition_by=(
responses.c.poll_id,
responses.c.respondent_id,
),
order_by=(
responses.c.submitted_at.desc(),
responses.c.id.desc(),
),
)
.label("response_rank"),
)
.where(
responses.c.deleted_at.is_(None),
responses.c.respondent_id.is_not(None),
)
.subquery("ranked_active_poll_responses")
)
duplicate_ids = sa.select(ranked.c.id).where(ranked.c.response_rank > 1)
migration_timestamp = sa.func.current_timestamp()
op.get_bind().execute(
responses.update()
.where(
responses.c.id.in_(duplicate_ids),
responses.c.deleted_at.is_(None),
)
.values(
deleted_at=migration_timestamp,
updated_at=migration_timestamp,
)
)
def upgrade() -> None:
connection = op.get_bind()
if connection.dialect.name == "postgresql":
connection.execute(
sa.text(
"LOCK TABLE poll_responses IN SHARE ROW EXCLUSIVE MODE"
)
)
_tombstone_duplicate_active_responses()
op.create_index(
_INDEX_NAME,
"poll_responses",
["poll_id", "respondent_id"],
unique=True,
sqlite_where=sa.text(_ACTIVE_IDENTIFIED_PREDICATE),
postgresql_where=sa.text(_ACTIVE_IDENTIFIED_PREDICATE),
)
def downgrade() -> None:
# Deduplicated rows remain tombstoned: a downgrade must not silently revive
# responses that were no longer part of the active result set.
op.drop_index(_INDEX_NAME, table_name="poll_responses")

210
tests/test_migrations.py Normal file
View File

@@ -0,0 +1,210 @@
from __future__ import annotations
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from alembic import command
from alembic.runtime.migration import MigrationContext
from sqlalchemy import create_engine, inspect
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from govoplan_core.db.migrations import alembic_config
from govoplan_poll.backend.db.models import PollResponse
from govoplan_poll.backend.manifest import get_manifest
from govoplan_poll.backend.schemas import (
PollCreateRequest,
PollOptionInput,
)
from govoplan_poll.backend.service import _existing_response, create_poll
_PREVIOUS_POLL_HEAD = "5d6e7f8a9b0c"
_POLL_HEAD = "6e7f8a9b0c1d"
_INDEX_NAME = "uq_poll_responses_active_respondent"
class PollMigrationTests(unittest.TestCase):
@staticmethod
def _config(url: str):
return alembic_config(
database_url=url,
enabled_modules=("poll",),
manifest_factories=(get_manifest,),
)
def test_deduplicates_deterministically_and_enforces_partial_index(self) -> None:
with tempfile.TemporaryDirectory(
prefix="govoplan-poll-migration-"
) as directory:
url = f"sqlite:///{Path(directory) / 'poll.db'}"
config = self._config(url)
command.upgrade(config, _PREVIOUS_POLL_HEAD)
engine = create_engine(url)
latest = datetime(2026, 7, 22, 10, 0, tzinfo=timezone.utc)
historical_deleted_at = latest - timedelta(days=2)
historical_updated_at = latest - timedelta(days=1)
try:
with Session(engine) as session:
poll = create_poll(
session,
tenant_id="tenant-1",
user_id="owner-1",
payload=PollCreateRequest(
title="Migration probe",
kind="single_choice",
status="open",
allow_anonymous=True,
options=[
PollOptionInput(key="yes", label="Yes"),
PollOptionInput(key="no", label="No"),
],
),
)
poll_id = poll.id
session.add_all(
[
PollResponse(
id="identified-oldest",
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id="person-1",
answers=[],
submitted_at=latest - timedelta(hours=1),
),
PollResponse(
id="identified-latest-a",
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id="person-1",
answers=[],
submitted_at=latest,
),
PollResponse(
id="identified-latest-b",
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id="person-1",
answers=[],
submitted_at=latest,
),
PollResponse(
id="anonymous-a",
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id=None,
answers=[],
submitted_at=latest,
),
PollResponse(
id="anonymous-b",
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id=None,
answers=[],
submitted_at=latest,
),
PollResponse(
id="already-deleted",
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id="person-1",
answers=[],
submitted_at=latest + timedelta(hours=1),
deleted_at=historical_deleted_at,
updated_at=historical_updated_at,
),
]
)
session.flush()
self.assertEqual(
_existing_response(
session,
poll=poll,
respondent_id="person-1",
).id,
"identified-latest-b",
)
session.commit()
command.upgrade(config, "heads")
with Session(engine) as session:
rows = {
response.id: response
for response in session.query(PollResponse).all()
}
self.assertIsNone(rows["identified-latest-b"].deleted_at)
tombstoned = (
rows["identified-oldest"],
rows["identified-latest-a"],
)
self.assertTrue(
all(response.deleted_at is not None for response in tombstoned)
)
self.assertTrue(
all(
response.deleted_at == response.updated_at
for response in tombstoned
)
)
self.assertEqual(
len({response.deleted_at for response in tombstoned}),
1,
)
self.assertIsNone(rows["anonymous-a"].deleted_at)
self.assertIsNone(rows["anonymous-b"].deleted_at)
self.assertEqual(
rows["already-deleted"].deleted_at,
historical_deleted_at.replace(tzinfo=None),
)
self.assertEqual(
rows["already-deleted"].updated_at,
historical_updated_at.replace(tzinfo=None),
)
session.add(
PollResponse(
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id="person-1",
answers=[],
submitted_at=latest + timedelta(hours=2),
)
)
with self.assertRaises(IntegrityError):
session.flush()
session.rollback()
session.add(
PollResponse(
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id=None,
answers=[],
submitted_at=latest + timedelta(hours=2),
)
)
session.commit()
with engine.connect() as connection:
heads = set(
MigrationContext.configure(connection).get_current_heads()
)
indexes = {
item["name"]: item
for item in inspect(connection).get_indexes(
"poll_responses"
)
}
self.assertIn(_POLL_HEAD, heads)
self.assertIn(_INDEX_NAME, indexes)
self.assertTrue(indexes[_INDEX_NAME]["unique"])
finally:
engine.dispose()
if __name__ == "__main__":
unittest.main()