diff --git a/backend/alembic/versions/b025_add_unique_test_detection_result.py b/backend/alembic/versions/b025_add_unique_test_detection_result.py new file mode 100644 index 0000000..1f5e431 --- /dev/null +++ b/backend/alembic/versions/b025_add_unique_test_detection_result.py @@ -0,0 +1,41 @@ +"""add_unique_test_detection_result + +Enforce one evaluation per (test, detection_rule) pair. Before creating +the constraint, duplicate rows (if any) are collapsed so the migration +never fails on an existing database. + +Revision ID: b025uqtdr +Revises: b024critidx +Create Date: 2026-02-18 14:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op + +revision: str = "b025uqtdr" +down_revision: Union[str, None] = "b024critidx" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # Remove duplicates keeping the most recently evaluated row + op.execute(""" + DELETE FROM test_detection_results + WHERE id NOT IN ( + SELECT DISTINCT ON (test_id, detection_rule_id) id + FROM test_detection_results + ORDER BY test_id, detection_rule_id, evaluated_at DESC NULLS LAST + ) + """) + + op.create_unique_constraint( + "uq_tdr_test_rule", + "test_detection_results", + ["test_id", "detection_rule_id"], + ) + + +def downgrade() -> None: + op.drop_constraint("uq_tdr_test_rule", "test_detection_results", type_="unique") diff --git a/backend/app/models/test_detection_result.py b/backend/app/models/test_detection_result.py index d43aaef..2897bbf 100644 --- a/backend/app/models/test_detection_result.py +++ b/backend/app/models/test_detection_result.py @@ -7,7 +7,7 @@ rule as triggered / not triggered / not applicable, along with notes. import uuid from datetime import datetime -from sqlalchemy import Column, String, Text, Boolean, DateTime, ForeignKey, Index +from sqlalchemy import Column, String, Text, Boolean, DateTime, ForeignKey, Index, UniqueConstraint from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.orm import relationship @@ -52,4 +52,5 @@ class TestDetectionResult(Base): __table_args__ = ( Index('ix_tdr_test', 'test_id'), Index('ix_tdr_rule', 'detection_rule_id'), + UniqueConstraint('test_id', 'detection_rule_id', name='uq_tdr_test_rule'), )