Celes Renata
a72f336ad1
Multi-stage evidence-grounded inference architecture replacing the monolithic 9B model extraction pipeline. CPU-first specialist services handle routine extraction while the 9B vLLM model is preserved for semantic adjudication of ambiguous cases. Key components: - Capability-aware inference gateway (OpenAI-compatible + Ollama) - Endpoint registry with DB migrations and REST API - Sentence-aware document segmenter (property tests) - Deterministic financial parsing with offset integrity - Symbol resolution with ambiguity detection - Specialist service (GLiNER2, dynamic batching, K8s deployment) - Company-specific sentiment (FinBERT, calibration) - Retrieval-based novelty and duplicate detection - Confidence calibration pipeline - Deterministic routing engine (property tests) - 9B adjudication layer with VRAM gating - Stock-specific impact model (features, labels, baseline, trained) - Pipeline orchestrator (state machine, queues, leases, feature flags) - Bounded parallelism (async workers, semaphore, load shedding) - Observability (tracing, metrics, alerts) - Compatibility adapter (v3→v2 golden mapping tests) - Shadow/canary promotion framework - Active learning and fine-tuning pipeline Test results: 1,161 tests pass, ruff lint clean. All 282 spec tasks completed.
629 lines
28 KiB
Python
629 lines
28 KiB
Python
"""Tests for migration 041_v3_pipeline_tables.sql.
|
|
|
|
Validates:
|
|
- SQL is syntactically valid (parseable)
|
|
- Required tables are created for all v3 pipeline stages
|
|
- Required constraints exist (CHECK, UNIQUE, FK references)
|
|
- Idempotency patterns (IF NOT EXISTS) used throughout
|
|
- Immutable-revision triggers are defined
|
|
"""
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
MIGRATION_PATH = (
|
|
Path(__file__).resolve().parent.parent
|
|
/ "infra"
|
|
/ "migrations"
|
|
/ "041_v3_pipeline_tables.sql"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def migration_sql() -> str:
|
|
"""Load the migration SQL content."""
|
|
assert MIGRATION_PATH.exists(), f"Migration file not found: {MIGRATION_PATH}"
|
|
return MIGRATION_PATH.read_text()
|
|
|
|
|
|
class TestMigrationFileExists:
|
|
def test_file_exists(self):
|
|
assert MIGRATION_PATH.exists()
|
|
|
|
def test_file_not_empty(self):
|
|
content = MIGRATION_PATH.read_text()
|
|
assert len(content.strip()) > 500
|
|
|
|
|
|
class TestSQLSyntax:
|
|
"""Basic syntax validation via regex pattern checks."""
|
|
|
|
def test_no_unclosed_parentheses(self, migration_sql: str):
|
|
sql = re.sub(r"--[^\n]*", "", migration_sql)
|
|
sql = re.sub(r"'[^']*'", "''", sql)
|
|
open_count = sql.count("(")
|
|
close_count = sql.count(")")
|
|
assert open_count == close_count, (
|
|
f"Unbalanced parentheses: {open_count} open vs {close_count} close"
|
|
)
|
|
|
|
def test_no_trailing_commas_before_close_paren(self, migration_sql: str):
|
|
sql = re.sub(r"--[^\n]*", "", migration_sql)
|
|
matches = re.findall(r",\s*\)", sql)
|
|
assert len(matches) == 0, f"Trailing commas before ')': {matches}"
|
|
|
|
def test_all_statements_terminated(self, migration_sql: str):
|
|
semicolons = migration_sql.count(";")
|
|
assert semicolons >= 30, f"Expected at least 30 semicolons, got {semicolons}"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 20.1: Pipeline runs and stage runs
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestPipelineRunsTable:
|
|
"""Verify v3_pipeline_runs table structure."""
|
|
|
|
def test_table_created(self, migration_sql: str):
|
|
assert "CREATE TABLE IF NOT EXISTS v3_pipeline_runs" in migration_sql
|
|
|
|
def test_has_id_primary_key(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_pipeline_runs")
|
|
assert "id UUID PRIMARY KEY" in block
|
|
|
|
def test_has_document_id(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_pipeline_runs")
|
|
assert "document_id UUID NOT NULL" in block
|
|
|
|
def test_has_pipeline_version(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_pipeline_runs")
|
|
assert "pipeline_version TEXT" in block
|
|
|
|
def test_has_status_check(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_pipeline_runs")
|
|
assert "CHECK" in block
|
|
assert "pending" in block
|
|
assert "running" in block
|
|
assert "completed" in block
|
|
assert "failed" in block
|
|
|
|
def test_has_idempotency_key_unique(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_pipeline_runs")
|
|
assert "idempotency_key TEXT NOT NULL UNIQUE" in block
|
|
|
|
def test_has_timestamps(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_pipeline_runs")
|
|
assert "started_at TIMESTAMPTZ" in block
|
|
assert "completed_at TIMESTAMPTZ" in block
|
|
assert "created_at TIMESTAMPTZ" in block
|
|
|
|
def test_has_error_field(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_pipeline_runs")
|
|
assert "error TEXT" in block
|
|
|
|
|
|
class TestStageRunsTable:
|
|
"""Verify v3_stage_runs table structure."""
|
|
|
|
def test_table_created(self, migration_sql: str):
|
|
assert "CREATE TABLE IF NOT EXISTS v3_stage_runs" in migration_sql
|
|
|
|
def test_has_pipeline_run_fk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_runs")
|
|
assert "REFERENCES v3_pipeline_runs(id)" in block
|
|
|
|
def test_has_stage_check(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_runs")
|
|
assert "segmentation" in block
|
|
assert "extraction" in block
|
|
assert "sentiment" in block
|
|
assert "novelty" in block
|
|
assert "routing" in block
|
|
assert "adjudication" in block
|
|
assert "impact" in block
|
|
assert "persistence" in block
|
|
|
|
def test_has_status_check(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_runs")
|
|
assert "pending" in block
|
|
assert "running" in block
|
|
assert "completed" in block
|
|
assert "failed" in block
|
|
assert "skipped" in block
|
|
|
|
def test_has_endpoint_and_deployment_refs(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_runs")
|
|
assert "endpoint_id UUID" in block
|
|
assert "deployment_id UUID" in block
|
|
|
|
def test_has_input_output_refs(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_runs")
|
|
assert "input_refs JSONB" in block
|
|
assert "output_refs JSONB" in block
|
|
|
|
def test_has_trace_id(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_runs")
|
|
assert "trace_id TEXT" in block
|
|
|
|
def test_has_model_and_schema_versions(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_runs")
|
|
assert "model_version TEXT" in block
|
|
assert "schema_version TEXT" in block
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 20.2: Document chunks and evidence spans
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestDocumentChunksTable:
|
|
"""Verify v3_document_chunks table structure."""
|
|
|
|
def test_table_created(self, migration_sql: str):
|
|
assert "CREATE TABLE IF NOT EXISTS v3_document_chunks" in migration_sql
|
|
|
|
def test_has_document_id(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_document_chunks")
|
|
assert "document_id UUID NOT NULL" in block
|
|
|
|
def test_has_chunk_id(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_document_chunks")
|
|
assert "chunk_id TEXT NOT NULL" in block
|
|
|
|
def test_has_unique_document_chunk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_document_chunks")
|
|
assert "UNIQUE(document_id, chunk_id)" in block
|
|
|
|
def test_has_section_path_jsonb(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_document_chunks")
|
|
assert "section_path JSONB" in block
|
|
|
|
def test_has_char_offsets(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_document_chunks")
|
|
assert "start_char INTEGER NOT NULL" in block
|
|
assert "end_char INTEGER NOT NULL" in block
|
|
|
|
def test_has_overlap_fields(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_document_chunks")
|
|
assert "overlap_left INTEGER" in block
|
|
assert "overlap_right INTEGER" in block
|
|
|
|
def test_has_boilerplate_score(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_document_chunks")
|
|
assert "boilerplate_score" in block
|
|
|
|
def test_has_document_type(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_document_chunks")
|
|
assert "document_type TEXT" in block
|
|
|
|
|
|
class TestEvidenceSpansTable:
|
|
"""Verify v3_evidence_spans table structure."""
|
|
|
|
def test_table_created(self, migration_sql: str):
|
|
assert "CREATE TABLE IF NOT EXISTS v3_evidence_spans" in migration_sql
|
|
|
|
def test_has_document_id(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_evidence_spans")
|
|
assert "document_id UUID NOT NULL" in block
|
|
|
|
def test_has_char_offsets(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_evidence_spans")
|
|
assert "start_char INTEGER NOT NULL" in block
|
|
assert "end_char INTEGER NOT NULL" in block
|
|
|
|
def test_has_text(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_evidence_spans")
|
|
assert "text TEXT NOT NULL" in block
|
|
|
|
def test_has_checksum(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_evidence_spans")
|
|
assert "checksum TEXT NOT NULL" in block
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 20.3: Extracted entities, facts, relations, and rejected candidates
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestExtractedEntitiesTable:
|
|
"""Verify v3_extracted_entities table structure."""
|
|
|
|
def test_table_created(self, migration_sql: str):
|
|
assert "CREATE TABLE IF NOT EXISTS v3_extracted_entities" in migration_sql
|
|
|
|
def test_has_pipeline_run_fk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_entities")
|
|
assert "REFERENCES v3_pipeline_runs(id)" in block
|
|
|
|
def test_has_entity_type(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_entities")
|
|
assert "entity_type TEXT NOT NULL" in block
|
|
|
|
def test_has_literal_text(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_entities")
|
|
assert "literal_text TEXT NOT NULL" in block
|
|
|
|
def test_has_canonical_id(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_entities")
|
|
assert "canonical_id UUID" in block
|
|
|
|
def test_has_evidence_span_fk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_entities")
|
|
assert "REFERENCES v3_evidence_spans(id)" in block
|
|
|
|
def test_has_confidence(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_entities")
|
|
assert "confidence REAL" in block
|
|
|
|
def test_has_derivation(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_entities")
|
|
assert "derivation TEXT" in block
|
|
|
|
|
|
class TestExtractedFactsTable:
|
|
"""Verify v3_extracted_facts table structure."""
|
|
|
|
def test_table_created(self, migration_sql: str):
|
|
assert "CREATE TABLE IF NOT EXISTS v3_extracted_facts" in migration_sql
|
|
|
|
def test_has_pipeline_run_fk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_facts")
|
|
assert "REFERENCES v3_pipeline_runs(id)" in block
|
|
|
|
def test_has_subject_entity_fk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_facts")
|
|
assert "REFERENCES v3_extracted_entities(id)" in block
|
|
|
|
def test_has_predicate(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_facts")
|
|
assert "predicate TEXT NOT NULL" in block
|
|
|
|
def test_has_literal_and_normalized(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_facts")
|
|
assert "literal_value TEXT NOT NULL" in block
|
|
assert "normalized_value JSONB" in block
|
|
|
|
def test_has_unit(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_facts")
|
|
assert "unit TEXT" in block
|
|
|
|
def test_has_period_jsonb(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_facts")
|
|
assert "period JSONB" in block
|
|
|
|
def test_has_evidence_span_ids_array(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_facts")
|
|
assert "evidence_span_ids UUID[]" in block
|
|
|
|
def test_has_confidence_and_derivation(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_facts")
|
|
assert "confidence REAL" in block
|
|
assert "derivation TEXT" in block
|
|
|
|
|
|
class TestExtractedRelationsTable:
|
|
"""Verify v3_extracted_relations table structure."""
|
|
|
|
def test_table_created(self, migration_sql: str):
|
|
assert "CREATE TABLE IF NOT EXISTS v3_extracted_relations" in migration_sql
|
|
|
|
def test_has_pipeline_run_fk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_relations")
|
|
assert "REFERENCES v3_pipeline_runs(id)" in block
|
|
|
|
def test_has_source_and_target_entity_fks(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_relations")
|
|
assert "source_entity_id UUID NOT NULL REFERENCES v3_extracted_entities(id)" in block
|
|
assert "target_entity_id UUID NOT NULL REFERENCES v3_extracted_entities(id)" in block
|
|
|
|
def test_has_relation_type(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_relations")
|
|
assert "relation_type TEXT NOT NULL" in block
|
|
|
|
def test_has_evidence_span_ids_array(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_extracted_relations")
|
|
assert "evidence_span_ids UUID[]" in block
|
|
|
|
|
|
class TestRejectedCandidatesTable:
|
|
"""Verify v3_rejected_candidates table structure."""
|
|
|
|
def test_table_created(self, migration_sql: str):
|
|
assert "CREATE TABLE IF NOT EXISTS v3_rejected_candidates" in migration_sql
|
|
|
|
def test_has_pipeline_run_fk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_rejected_candidates")
|
|
assert "REFERENCES v3_pipeline_runs(id)" in block
|
|
|
|
def test_has_candidate_type(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_rejected_candidates")
|
|
assert "candidate_type TEXT NOT NULL" in block
|
|
|
|
def test_has_candidate_data_jsonb(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_rejected_candidates")
|
|
assert "candidate_data JSONB NOT NULL" in block
|
|
|
|
def test_has_rejection_reason(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_rejected_candidates")
|
|
assert "rejection_reason TEXT NOT NULL" in block
|
|
|
|
def test_has_stage(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_rejected_candidates")
|
|
assert "stage TEXT NOT NULL" in block
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 20.4: Company signal candidates and probability distributions
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestCompanySignalCandidatesTable:
|
|
"""Verify v3_company_signal_candidates table structure."""
|
|
|
|
def test_table_created(self, migration_sql: str):
|
|
assert "CREATE TABLE IF NOT EXISTS v3_company_signal_candidates" in migration_sql
|
|
|
|
def test_has_pipeline_run_fk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_company_signal_candidates")
|
|
assert "REFERENCES v3_pipeline_runs(id)" in block
|
|
|
|
def test_has_company_fk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_company_signal_candidates")
|
|
assert "REFERENCES companies(id)" in block
|
|
|
|
def test_has_relevance_probability(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_company_signal_candidates")
|
|
assert "relevance_probability REAL" in block
|
|
|
|
def test_has_probability_distributions(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_company_signal_candidates")
|
|
assert "event_probabilities JSONB" in block
|
|
assert "sentiment_probabilities JSONB" in block
|
|
assert "direction_probabilities JSONB" in block
|
|
assert "horizon_probabilities JSONB" in block
|
|
|
|
def test_has_expected_magnitude(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_company_signal_candidates")
|
|
assert "expected_magnitude REAL" in block
|
|
|
|
def test_has_evidence_span_ids_array(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_company_signal_candidates")
|
|
assert "evidence_span_ids UUID[]" in block
|
|
|
|
def test_has_routing_reasons_array(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_company_signal_candidates")
|
|
assert "routing_reasons TEXT[]" in block
|
|
|
|
def test_has_adjudicated_bool(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_company_signal_candidates")
|
|
assert "adjudicated BOOLEAN" in block
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 20.5: Adjudication decisions, routing, and lineage
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestAdjudicationDecisionsTable:
|
|
"""Verify v3_adjudication_decisions table structure."""
|
|
|
|
def test_table_created(self, migration_sql: str):
|
|
assert "CREATE TABLE IF NOT EXISTS v3_adjudication_decisions" in migration_sql
|
|
|
|
def test_has_pipeline_run_fk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_adjudication_decisions")
|
|
assert "REFERENCES v3_pipeline_runs(id)" in block
|
|
|
|
def test_has_question_codes_array(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_adjudication_decisions")
|
|
assert "question_codes TEXT[]" in block
|
|
|
|
def test_has_candidates_jsonb(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_adjudication_decisions")
|
|
assert "candidates JSONB" in block
|
|
|
|
def test_has_decision_jsonb(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_adjudication_decisions")
|
|
assert "decision JSONB" in block
|
|
|
|
def test_has_evidence_span_ids_array(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_adjudication_decisions")
|
|
assert "evidence_span_ids UUID[]" in block
|
|
|
|
def test_has_model_lineage_fk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_adjudication_decisions")
|
|
assert "REFERENCES v3_stage_lineage(id)" in block
|
|
|
|
|
|
class TestRoutingDecisionsTable:
|
|
"""Verify v3_routing_decisions table structure."""
|
|
|
|
def test_table_created(self, migration_sql: str):
|
|
assert "CREATE TABLE IF NOT EXISTS v3_routing_decisions" in migration_sql
|
|
|
|
def test_has_pipeline_run_fk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_routing_decisions")
|
|
assert "REFERENCES v3_pipeline_runs(id)" in block
|
|
|
|
def test_has_route_check(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_routing_decisions")
|
|
assert "fast_path" in block
|
|
assert "adjudication" in block
|
|
assert "CHECK" in block
|
|
|
|
def test_has_reason_codes_array(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_routing_decisions")
|
|
assert "reason_codes TEXT[]" in block
|
|
|
|
def test_has_confidence_features_jsonb(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_routing_decisions")
|
|
assert "confidence_features JSONB" in block
|
|
|
|
|
|
class TestStageLineageTable:
|
|
"""Verify v3_stage_lineage table structure."""
|
|
|
|
def test_table_created(self, migration_sql: str):
|
|
assert "CREATE TABLE IF NOT EXISTS v3_stage_lineage" in migration_sql
|
|
|
|
def test_has_stage_run_fk(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_lineage")
|
|
assert "REFERENCES v3_stage_runs(id)" in block
|
|
|
|
def test_has_endpoint_and_deployment_refs(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_lineage")
|
|
assert "endpoint_id UUID" in block
|
|
assert "deployment_id UUID" in block
|
|
assert "REFERENCES inference_endpoints(id)" in block
|
|
assert "REFERENCES model_deployments(id)" in block
|
|
|
|
def test_has_model_and_protocol(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_lineage")
|
|
assert "model TEXT" in block
|
|
assert "protocol TEXT" in block
|
|
|
|
def test_has_structured_mode(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_lineage")
|
|
assert "structured_mode TEXT" in block
|
|
|
|
def test_has_request_id(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_lineage")
|
|
assert "request_id TEXT" in block
|
|
|
|
def test_has_latency_ms(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_lineage")
|
|
assert "latency_ms INTEGER" in block
|
|
|
|
def test_has_retries(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_lineage")
|
|
assert "retries INTEGER" in block
|
|
|
|
def test_has_trace_id(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_stage_lineage")
|
|
assert "trace_id TEXT" in block
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 20.6: Idempotency and immutable-revision constraints
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
class TestIdempotencyConstraints:
|
|
"""Verify idempotency indexes and constraints."""
|
|
|
|
def test_pipeline_runs_idempotency_key_unique(self, migration_sql: str):
|
|
block = _extract_create_block(migration_sql, "v3_pipeline_runs")
|
|
assert "idempotency_key TEXT NOT NULL UNIQUE" in block
|
|
|
|
def test_stage_runs_idempotent_index(self, migration_sql: str):
|
|
assert "idx_v3_stage_runs_idempotent" in migration_sql
|
|
assert "ON v3_stage_runs(pipeline_run_id, stage)" in migration_sql
|
|
|
|
def test_signal_candidates_idempotent_index(self, migration_sql: str):
|
|
assert "idx_v3_signal_candidates_idempotent" in migration_sql
|
|
assert "ON v3_company_signal_candidates(pipeline_run_id, company_id)" in migration_sql
|
|
|
|
def test_routing_idempotent_index(self, migration_sql: str):
|
|
assert "idx_v3_routing_idempotent" in migration_sql
|
|
assert "ON v3_routing_decisions(pipeline_run_id)" in migration_sql
|
|
|
|
def test_evidence_spans_idempotent_index(self, migration_sql: str):
|
|
assert "idx_v3_evidence_spans_idempotent" in migration_sql
|
|
assert "ON v3_evidence_spans(document_id, checksum)" in migration_sql
|
|
|
|
|
|
class TestImmutableRevisionTriggers:
|
|
"""Verify immutable-row triggers prevent updating completed records."""
|
|
|
|
def test_immutable_function_defined(self, migration_sql: str):
|
|
assert "v3_immutable_completed_row" in migration_sql
|
|
|
|
def test_pipeline_runs_immutable_trigger(self, migration_sql: str):
|
|
assert "trg_v3_pipeline_runs_immutable" in migration_sql
|
|
|
|
def test_stage_runs_immutable_trigger(self, migration_sql: str):
|
|
assert "trg_v3_stage_runs_immutable" in migration_sql
|
|
|
|
def test_trigger_checks_completed_and_failed(self, migration_sql: str):
|
|
# The trigger function should check for both terminal states
|
|
assert "'completed'" in migration_sql or "completed" in migration_sql
|
|
assert "'failed'" in migration_sql or "failed" in migration_sql
|
|
|
|
|
|
class TestIdempotentPatterns:
|
|
"""Verify the migration uses idempotent DDL patterns."""
|
|
|
|
def test_create_table_if_not_exists(self, migration_sql: str):
|
|
creates = re.findall(r"CREATE TABLE\b", migration_sql)
|
|
creates_idempotent = re.findall(r"CREATE TABLE IF NOT EXISTS", migration_sql)
|
|
assert len(creates) == len(creates_idempotent), (
|
|
"All CREATE TABLE should use IF NOT EXISTS"
|
|
)
|
|
|
|
def test_create_index_if_not_exists(self, migration_sql: str):
|
|
indexes = re.findall(r"CREATE INDEX\b", migration_sql)
|
|
indexes_idempotent = re.findall(r"CREATE INDEX IF NOT EXISTS", migration_sql)
|
|
assert len(indexes) == len(indexes_idempotent), (
|
|
"All CREATE INDEX should use IF NOT EXISTS"
|
|
)
|
|
|
|
def test_create_unique_index_if_not_exists(self, migration_sql: str):
|
|
indexes = re.findall(r"CREATE UNIQUE INDEX\b", migration_sql)
|
|
indexes_idempotent = re.findall(r"CREATE UNIQUE INDEX IF NOT EXISTS", migration_sql)
|
|
assert len(indexes) == len(indexes_idempotent), (
|
|
"All CREATE UNIQUE INDEX should use IF NOT EXISTS"
|
|
)
|
|
|
|
def test_trigger_uses_drop_if_exists(self, migration_sql: str):
|
|
drops = re.findall(r"DROP TRIGGER IF EXISTS", migration_sql)
|
|
creates = re.findall(r"CREATE TRIGGER", migration_sql)
|
|
assert len(drops) == len(creates), (
|
|
"Each CREATE TRIGGER should be preceded by DROP TRIGGER IF EXISTS"
|
|
)
|
|
|
|
|
|
class TestIndexes:
|
|
"""Verify key indexes exist for query performance."""
|
|
|
|
def test_pipeline_runs_document_index(self, migration_sql: str):
|
|
assert "idx_v3_pipeline_runs_document" in migration_sql
|
|
|
|
def test_pipeline_runs_status_index(self, migration_sql: str):
|
|
assert "idx_v3_pipeline_runs_status" in migration_sql
|
|
|
|
def test_stage_runs_pipeline_index(self, migration_sql: str):
|
|
assert "idx_v3_stage_runs_pipeline" in migration_sql
|
|
|
|
def test_document_chunks_document_index(self, migration_sql: str):
|
|
assert "idx_v3_document_chunks_document" in migration_sql
|
|
|
|
def test_evidence_spans_document_index(self, migration_sql: str):
|
|
assert "idx_v3_evidence_spans_document" in migration_sql
|
|
|
|
def test_entities_pipeline_index(self, migration_sql: str):
|
|
assert "idx_v3_extracted_entities_pipeline" in migration_sql
|
|
|
|
def test_facts_pipeline_index(self, migration_sql: str):
|
|
assert "idx_v3_extracted_facts_pipeline" in migration_sql
|
|
|
|
def test_signal_candidates_company_index(self, migration_sql: str):
|
|
assert "idx_v3_signal_candidates_company" in migration_sql
|
|
|
|
def test_stage_lineage_stage_run_index(self, migration_sql: str):
|
|
assert "idx_v3_stage_lineage_stage_run" in migration_sql
|
|
|
|
|
|
# ─── Helpers ───────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _extract_create_block(sql: str, table_name: str) -> str:
|
|
"""Extract the CREATE TABLE block for a given table name."""
|
|
pattern = rf"CREATE TABLE IF NOT EXISTS {table_name}\s*\((.*?)\);"
|
|
match = re.search(pattern, sql, re.DOTALL)
|
|
assert match is not None, f"Could not find CREATE TABLE block for {table_name}"
|
|
return match.group(1)
|