94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
"""Model/enum layer must match the migrated schema (issue 004).
|
|
|
|
Guards the drift fixed in .scratch/refactor-base-api-alignment/issues/004:
|
|
TEXT+CHECK status/queue sets, the jobs (document_id, queue) unique
|
|
constraint, and the PrescreenResult mapping for `prescreen_results`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from sqlalchemy import CheckConstraint, UniqueConstraint
|
|
|
|
from contract_check.core.db.enums import (
|
|
DOC_STATUSES,
|
|
QUEUE_NAMES,
|
|
ROUTING_DECISIONS,
|
|
)
|
|
from contract_check.core.db.models import Base, PrescreenResult
|
|
|
|
|
|
class TestEnumsMatchMigrations:
|
|
def test_doc_statuses_cover_migrations_0006_and_0008(self) -> None:
|
|
assert "prescreening" in DOC_STATUSES
|
|
assert "manual_review" in DOC_STATUSES
|
|
|
|
def test_queue_names_cover_migration_0006(self) -> None:
|
|
assert "prescreen" in QUEUE_NAMES
|
|
|
|
def test_routing_decisions_match_prescreen_check(self) -> None:
|
|
assert ROUTING_DECISIONS == ("auto_approve", "manual_review", "deep_analysis")
|
|
|
|
|
|
class TestPrescreenResultModel:
|
|
@pytest.fixture
|
|
def table(self): # type: ignore[no-untyped-def]
|
|
return PrescreenResult.__table__
|
|
|
|
def test_mapped_to_prescreen_results(self, table) -> None: # type: ignore[no-untyped-def]
|
|
assert table.name == "prescreen_results"
|
|
assert "prescreen_results" in Base.metadata.tables
|
|
|
|
def test_check_constraints_match_migration_0006(self, table) -> None: # type: ignore[no-untyped-def]
|
|
by_name = {c.name: c for c in table.constraints if isinstance(c, CheckConstraint)}
|
|
assert str(by_name["prescreen_results_routing_decision_check"].sqltext) == (
|
|
"routing_decision IN ('auto_approve','manual_review','deep_analysis')"
|
|
)
|
|
assert "confidence_score >= 0" in str(by_name["prescreen_results_confidence_check"].sqltext)
|
|
assert str(by_name["prescreen_results_retry_count_nonneg"].sqltext) == "retry_count >= 0"
|
|
|
|
def test_indexes_match_migration_0006(self, table) -> None: # type: ignore[no-untyped-def]
|
|
names = {idx.name for idx in table.indexes}
|
|
assert names == {
|
|
"ix_prescreen_results_document_id",
|
|
"prescreen_results_routing_idx",
|
|
"prescreen_results_confidence_idx",
|
|
}
|
|
|
|
def test_server_defaults_match_migration_0006(self, table) -> None: # type: ignore[no-untyped-def]
|
|
for column in ("routing_decision", "extractor_version", "auto_findings", "retry_count"):
|
|
assert table.c[column].server_default is not None, column
|
|
|
|
|
|
class TestJobsUniqueConstraint:
|
|
def test_document_queue_unique_present(self) -> None:
|
|
jobs = Base.metadata.tables["jobs"]
|
|
uniques = [
|
|
c
|
|
for c in jobs.constraints
|
|
if isinstance(c, UniqueConstraint) and c.name == "jobs_document_queue_unique"
|
|
]
|
|
assert len(uniques) == 1
|
|
assert [c.name for c in uniques[0].columns] == ["document_id", "queue"]
|
|
|
|
|
|
class TestStatusCheckConstraints:
|
|
def test_documents_status_check_matches_migration_0008(self) -> None:
|
|
documents = Base.metadata.tables["documents"]
|
|
check = next(
|
|
c
|
|
for c in documents.constraints
|
|
if isinstance(c, CheckConstraint) and c.name == "documents_status_check"
|
|
)
|
|
sql = str(check.sqltext)
|
|
for status in ("prescreening", "manual_review"):
|
|
assert f"'{status}'" in sql
|
|
|
|
def test_jobs_queue_check_matches_migration_0006(self) -> None:
|
|
jobs = Base.metadata.tables["jobs"]
|
|
check = next(
|
|
c
|
|
for c in jobs.constraints
|
|
if isinstance(c, CheckConstraint) and c.name == "jobs_queue_check"
|
|
)
|
|
assert "'prescreen'" in str(check.sqltext)
|