"""Tests for evidence verification, entailment, coverage metrics, rejected store, and metrics. Covers: - Valid offset verification - Invalid offset detection (text mismatch, out of bounds) - Entity-evidence association - Numeric consistency (value found / not found in evidence) - Rejected candidate storage with reason codes - RejectedCandidateStore (store, get_by_pipeline_run, get_by_reason) - Entailment baseline (keyword overlap and exact match) - Coverage metrics computation - VerificationMetrics aggregation (unsupported-claim and evidence-coverage rates) - Full verification report """ from __future__ import annotations import pytest from services.intelligence_pipeline_v3.verification.coverage import ( FieldEvidence, compute_coverage, ) from services.intelligence_pipeline_v3.verification.entailment import ( EntailmentVerifier, ) from services.intelligence_pipeline_v3.verification.metrics import ( VerificationMetrics, compute_verification_metrics, ) from services.intelligence_pipeline_v3.verification.models import ( RejectedCandidate, RejectionReason, VerificationReport, ) from services.intelligence_pipeline_v3.verification.rejected_store import ( RejectedCandidateStore, ) from services.intelligence_pipeline_v3.verification.verifier import ( Candidate, Entity, EvidenceSpan, EvidenceVerifier, NumericFact, ) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- SOURCE_TEXT = ( "Apple Inc. reported revenue of $94.8 billion for Q1 2024, " "beating analyst expectations of $92.0 billion. " "CEO Tim Cook said the company saw strong growth in services." ) @pytest.fixture def source_text() -> str: return SOURCE_TEXT @pytest.fixture def valid_spans(source_text: str) -> list[EvidenceSpan]: """Spans that exactly match the source text at declared offsets.""" return [ EvidenceSpan( id="span-1", start_char=0, end_char=10, text=source_text[0:10], # "Apple Inc." ), EvidenceSpan( id="span-2", start_char=11, end_char=58, text=source_text[11:58], ), EvidenceSpan( id="span-3", start_char=60, end_char=107, text=source_text[60:107], ), ] @pytest.fixture def verifier() -> EvidenceVerifier: return EvidenceVerifier() # --------------------------------------------------------------------------- # Test: Valid offset verification # --------------------------------------------------------------------------- class TestOffsetVerification: def test_valid_offsets_pass( self, verifier: EvidenceVerifier, valid_spans: list[EvidenceSpan], source_text: str ): results = verifier.verify_offsets(valid_spans, source_text) assert len(results) == 3 assert all(r.valid for r in results) assert all(r.reason is None for r in results) def test_text_mismatch_detected(self, verifier: EvidenceVerifier, source_text: str): """Span with text that doesn't match source at the declared offset.""" bad_span = EvidenceSpan( id="span-bad", start_char=0, end_char=10, text="Google LLC", # Wrong — source has "Apple Inc." ) results = verifier.verify_offsets([bad_span], source_text) assert len(results) == 1 assert not results[0].valid assert "Text mismatch" in results[0].reason def test_offset_out_of_bounds(self, verifier: EvidenceVerifier, source_text: str): """Span with end_char beyond source text length.""" bad_span = EvidenceSpan( id="span-oob", start_char=0, end_char=len(source_text) + 100, text="doesn't matter", ) results = verifier.verify_offsets([bad_span], source_text) assert len(results) == 1 assert not results[0].valid assert "out of bounds" in results[0].reason.lower() def test_invalid_range_end_before_start(self, verifier: EvidenceVerifier, source_text: str): """Span where end_char <= start_char.""" bad_span = EvidenceSpan( id="span-reversed", start_char=10, end_char=5, text="x", ) results = verifier.verify_offsets([bad_span], source_text) assert len(results) == 1 assert not results[0].valid assert "Invalid range" in results[0].reason # --------------------------------------------------------------------------- # Test: Entity-evidence association # --------------------------------------------------------------------------- class TestEntityAssociation: def test_entity_found_in_evidence( self, verifier: EvidenceVerifier, valid_spans: list[EvidenceSpan] ): entity = Entity( id="ent-1", literal_text="Apple Inc.", evidence_ids=["span-1"] ) result = verifier.verify_entity_association(entity, valid_spans) assert result.valid assert result.reason is None def test_entity_case_insensitive( self, verifier: EvidenceVerifier, valid_spans: list[EvidenceSpan] ): """Entity matching should be case-insensitive.""" entity = Entity( id="ent-2", literal_text="apple inc.", evidence_ids=["span-1"] ) result = verifier.verify_entity_association(entity, valid_spans) assert result.valid def test_entity_not_in_evidence( self, verifier: EvidenceVerifier, valid_spans: list[EvidenceSpan] ): """Entity text is not present in any linked span.""" entity = Entity( id="ent-3", literal_text="Microsoft", evidence_ids=["span-1", "span-2"] ) result = verifier.verify_entity_association(entity, valid_spans) assert not result.valid assert "not found" in result.reason.lower() def test_entity_with_nonexistent_span_id( self, verifier: EvidenceVerifier, valid_spans: list[EvidenceSpan] ): """Entity references a span ID that doesn't exist.""" entity = Entity( id="ent-4", literal_text="Apple", evidence_ids=["span-nonexistent"] ) result = verifier.verify_entity_association(entity, valid_spans) assert not result.valid # --------------------------------------------------------------------------- # Test: Numeric consistency # --------------------------------------------------------------------------- class TestNumericConsistency: def test_literal_value_found( self, verifier: EvidenceVerifier, valid_spans: list[EvidenceSpan] ): """Literal value string is found directly in evidence text.""" fact = NumericFact( id="fact-1", literal_value="$94.8 billion", normalized_value=94.8, evidence_ids=["span-2"], ) result = verifier.verify_numeric_consistency(fact, valid_spans) assert result.valid assert result.found_value == "$94.8 billion" def test_normalized_value_match( self, verifier: EvidenceVerifier, valid_spans: list[EvidenceSpan] ): """Normalized value matches a number in evidence (without exact literal).""" fact = NumericFact( id="fact-2", literal_value="92 billion", # Not exact match normalized_value=92.0, evidence_ids=["span-3"], ) result = verifier.verify_numeric_consistency(fact, valid_spans) assert result.valid assert result.found_value == "92.0" def test_value_not_found( self, verifier: EvidenceVerifier, valid_spans: list[EvidenceSpan] ): """Value doesn't appear in any linked evidence.""" fact = NumericFact( id="fact-3", literal_value="$200 billion", normalized_value=200.0, evidence_ids=["span-2", "span-3"], ) result = verifier.verify_numeric_consistency(fact, valid_spans) assert not result.valid assert result.found_value is None assert "not found" in result.reason.lower() def test_tolerance_matching(self, valid_spans: list[EvidenceSpan]): """Values within tolerance should match.""" verifier = EvidenceVerifier(numeric_tolerance=0.02) # 2% tolerance fact = NumericFact( id="fact-4", literal_value="93.8", normalized_value=93.8, # Within 2% of 94.8 evidence_ids=["span-2"], ) result = verifier.verify_numeric_consistency(fact, valid_spans) assert result.valid # --------------------------------------------------------------------------- # Test: Rejected candidate storage (in verifier) # --------------------------------------------------------------------------- class TestRejectedCandidates: def test_offset_rejection_stored(self, verifier: EvidenceVerifier, source_text: str): bad_span = EvidenceSpan( id="span-bad", start_char=0, end_char=10, text="WRONG TEXT", ) verifier.verify_offsets([bad_span], source_text) rejected = verifier.rejected_candidates assert len(rejected) == 1 assert rejected[0].rejection_reason == RejectionReason.TEXT_MISMATCH assert rejected[0].candidate_type == "evidence_span" assert rejected[0].stage == "offset_verification" def test_entity_rejection_stored( self, verifier: EvidenceVerifier, valid_spans: list[EvidenceSpan] ): entity = Entity( id="ent-bad", literal_text="Nonexistent Corp", evidence_ids=["span-1"] ) verifier.verify_entity_association(entity, valid_spans) rejected = verifier.rejected_candidates assert len(rejected) == 1 assert rejected[0].rejection_reason == RejectionReason.ENTITY_NOT_IN_EVIDENCE assert rejected[0].candidate_type == "entity" assert rejected[0].candidate_data["entity_id"] == "ent-bad" def test_numeric_rejection_stored( self, verifier: EvidenceVerifier, valid_spans: list[EvidenceSpan] ): fact = NumericFact( id="fact-bad", literal_value="$999", normalized_value=999.0, evidence_ids=["span-2"], ) verifier.verify_numeric_consistency(fact, valid_spans) rejected = verifier.rejected_candidates assert len(rejected) == 1 assert rejected[0].rejection_reason == RejectionReason.NUMERIC_INCONSISTENCY assert rejected[0].candidate_type == "fact" def test_reset_clears_rejected( self, verifier: EvidenceVerifier, valid_spans: list[EvidenceSpan] ): entity = Entity( id="ent-x", literal_text="Nothing", evidence_ids=["span-1"] ) verifier.verify_entity_association(entity, valid_spans) assert len(verifier.rejected_candidates) == 1 verifier.reset() assert len(verifier.rejected_candidates) == 0 def test_rejection_has_timestamp( self, verifier: EvidenceVerifier, valid_spans: list[EvidenceSpan] ): entity = Entity( id="ent-ts", literal_text="Nobody", evidence_ids=["span-1"] ) verifier.verify_entity_association(entity, valid_spans) rejected = verifier.rejected_candidates assert rejected[0].timestamp is not None # --------------------------------------------------------------------------- # Test: RejectedCandidateStore # --------------------------------------------------------------------------- class TestRejectedCandidateStore: def test_store_and_retrieve_by_run(self): store = RejectedCandidateStore() rc = RejectedCandidate( candidate_type="entity", candidate_data={"entity_id": "e1", "text": "Apple"}, rejection_reason=RejectionReason.ENTITY_NOT_IN_EVIDENCE, stage="entity_verification", ) store.store(rc, run_id="run-001") results = store.get_by_pipeline_run("run-001") assert len(results) == 1 assert results[0].candidate_data["entity_id"] == "e1" def test_retrieve_empty_run(self): store = RejectedCandidateStore() results = store.get_by_pipeline_run("nonexistent-run") assert results == [] def test_store_and_retrieve_by_reason(self): store = RejectedCandidateStore() rc1 = RejectedCandidate( candidate_type="entity", candidate_data={"id": "e1"}, rejection_reason=RejectionReason.ENTITY_NOT_IN_EVIDENCE, stage="entity_verification", ) rc2 = RejectedCandidate( candidate_type="fact", candidate_data={"id": "f1"}, rejection_reason=RejectionReason.NUMERIC_INCONSISTENCY, stage="numeric_verification", ) rc3 = RejectedCandidate( candidate_type="entity", candidate_data={"id": "e2"}, rejection_reason=RejectionReason.ENTITY_NOT_IN_EVIDENCE, stage="entity_verification", ) store.store(rc1, run_id="run-1") store.store(rc2, run_id="run-1") store.store(rc3, run_id="run-2") by_entity = store.get_by_reason(RejectionReason.ENTITY_NOT_IN_EVIDENCE) assert len(by_entity) == 2 by_numeric = store.get_by_reason(RejectionReason.NUMERIC_INCONSISTENCY) assert len(by_numeric) == 1 def test_store_batch(self): store = RejectedCandidateStore() batch = [ RejectedCandidate( candidate_type="entity", candidate_data={"id": f"e{i}"}, rejection_reason=RejectionReason.INVALID_OFFSET, stage="offset_verification", ) for i in range(5) ] store.store_batch(batch, run_id="run-batch") assert store.count() == 5 assert len(store.get_by_pipeline_run("run-batch")) == 5 def test_count_by_reason(self): store = RejectedCandidateStore() store.store(RejectedCandidate( candidate_type="span", candidate_data={}, rejection_reason=RejectionReason.INVALID_OFFSET, stage="offset", )) store.store(RejectedCandidate( candidate_type="span", candidate_data={}, rejection_reason=RejectionReason.INVALID_OFFSET, stage="offset", )) store.store(RejectedCandidate( candidate_type="fact", candidate_data={}, rejection_reason=RejectionReason.NUMERIC_INCONSISTENCY, stage="numeric", )) counts = store.count_by_reason() assert counts["invalid_offset"] == 2 assert counts["numeric_inconsistency"] == 1 def test_clear(self): store = RejectedCandidateStore() store.store(RejectedCandidate( candidate_type="entity", candidate_data={}, rejection_reason=RejectionReason.ENTITY_NOT_IN_EVIDENCE, stage="test", ), run_id="run-1") assert store.count() == 1 store.clear() assert store.count() == 0 assert store.get_by_pipeline_run("run-1") == [] assert store.get_by_reason(RejectionReason.ENTITY_NOT_IN_EVIDENCE) == [] # --------------------------------------------------------------------------- # Test: Entailment baseline (keyword overlap) # --------------------------------------------------------------------------- class TestEntailment: def test_exact_match_entailment(self): ev = EntailmentVerifier() result = ev.verify_claim( claim="reported revenue of $94.8 billion", evidence="Apple Inc. reported revenue of $94.8 billion for Q1 2024", ) assert result.entailed assert result.confidence == 1.0 assert result.method == "exact_match" def test_keyword_overlap_entailed(self): ev = EntailmentVerifier(keyword_threshold=0.5) result = ev.verify_claim( claim="Apple revenue grew significantly", evidence="Apple Inc. reported record revenue growth of 15% year-over-year", ) assert result.entailed assert result.method == "keyword_overlap" assert result.confidence >= 0.5 def test_keyword_overlap_not_entailed(self): ev = EntailmentVerifier(keyword_threshold=0.6) result = ev.verify_claim( claim="Microsoft acquired a gaming company", evidence="Apple Inc. reported revenue of $94.8 billion for Q1 2024", ) assert not result.entailed assert result.method == "keyword_overlap" assert result.confidence < 0.6 def test_empty_claim(self): ev = EntailmentVerifier() result = ev.verify_claim(claim="", evidence="Some evidence text") assert not result.entailed assert result.confidence == 0.0 def test_empty_evidence(self): ev = EntailmentVerifier() result = ev.verify_claim(claim="Some claim", evidence="") assert not result.entailed assert result.confidence == 0.0 def test_batch_verification(self): ev = EntailmentVerifier() claims = [ "reported revenue", "completely unrelated topic about cats", ] evidence = "Apple reported revenue of $94.8 billion" results = ev.verify_claims_batch(claims, evidence) assert len(results) == 2 assert results[0].entailed # "reported revenue" is in evidence assert not results[1].entailed # cats not related def test_model_version_present(self): """EntailmentResult includes model_version field.""" ev = EntailmentVerifier() result = ev.verify_claim( claim="revenue growth", evidence="The company reported strong revenue growth this quarter.", ) assert result.model_version == "keyword_overlap_v1" # --------------------------------------------------------------------------- # Test: Coverage metrics # --------------------------------------------------------------------------- class TestCoverageMetrics: def test_full_coverage(self): fields = [ FieldEvidence(field_id="f1", field_name="revenue", evidence_ids=["s1", "s2"]), FieldEvidence(field_id="f2", field_name="eps", evidence_ids=["s2"]), ] verified = {"s1", "s2", "s3"} metrics = compute_coverage(fields, verified) assert metrics.total_fields == 2 assert metrics.supported_fields == 2 assert metrics.coverage_rate == 1.0 assert metrics.unsupported_claims == [] assert metrics.unsupported_rate == 0.0 def test_partial_coverage(self): fields = [ FieldEvidence(field_id="f1", field_name="revenue", evidence_ids=["s1"]), FieldEvidence(field_id="f2", field_name="eps", evidence_ids=["s4"]), FieldEvidence(field_id="f3", field_name="guidance", evidence_ids=["s2"]), ] verified = {"s1", "s2", "s3"} metrics = compute_coverage(fields, verified) assert metrics.total_fields == 3 assert metrics.supported_fields == 2 assert metrics.coverage_rate == pytest.approx(2 / 3) assert metrics.unsupported_claims == ["f2"] assert metrics.unsupported_rate == pytest.approx(1 / 3) def test_no_coverage(self): fields = [ FieldEvidence(field_id="f1", field_name="revenue", evidence_ids=["s99"]), FieldEvidence(field_id="f2", field_name="eps", evidence_ids=["s100"]), ] verified = {"s1", "s2"} metrics = compute_coverage(fields, verified) assert metrics.total_fields == 2 assert metrics.supported_fields == 0 assert metrics.coverage_rate == 0.0 assert len(metrics.unsupported_claims) == 2 assert metrics.unsupported_rate == 1.0 def test_empty_fields(self): """No fields to verify means perfect coverage by definition.""" metrics = compute_coverage([], {"s1", "s2"}) assert metrics.total_fields == 0 assert metrics.coverage_rate == 1.0 assert metrics.unsupported_rate == 0.0 def test_field_with_no_evidence_ids(self): """Field with empty evidence_ids is unsupported.""" fields = [ FieldEvidence(field_id="f1", field_name="revenue", evidence_ids=[]), ] verified = {"s1", "s2"} metrics = compute_coverage(fields, verified) assert metrics.supported_fields == 0 assert metrics.unsupported_claims == ["f1"] # --------------------------------------------------------------------------- # Test: VerificationMetrics (unsupported-claim and evidence-coverage rates) # --------------------------------------------------------------------------- class TestVerificationMetrics: def test_single_report_all_pass(self): reports = [ VerificationReport( total_candidates=10, verified=10, rejected=0, coverage_rate=1.0, rejection_breakdown={}, ) ] metrics = compute_verification_metrics(reports) assert metrics.total_checked == 10 assert metrics.passed_count == 10 assert metrics.failed_count == 0 assert metrics.evidence_coverage_rate == 1.0 assert metrics.unsupported_claim_rate == 0.0 assert metrics.per_reason_counts == {} def test_single_report_some_failures(self): reports = [ VerificationReport( total_candidates=10, verified=7, rejected=3, coverage_rate=0.7, rejection_breakdown={ "entity_not_in_evidence": 2, "unsupported_claim": 1, }, ) ] metrics = compute_verification_metrics(reports) assert metrics.total_checked == 10 assert metrics.passed_count == 7 assert metrics.failed_count == 3 assert metrics.evidence_coverage_rate == 0.7 assert metrics.unsupported_claim_rate == pytest.approx(0.1) assert metrics.per_reason_counts["entity_not_in_evidence"] == 2 assert metrics.per_reason_counts["unsupported_claim"] == 1 def test_multiple_reports_aggregated(self): reports = [ VerificationReport( total_candidates=5, verified=4, rejected=1, coverage_rate=0.8, rejection_breakdown={"invalid_offset": 1}, ), VerificationReport( total_candidates=10, verified=8, rejected=2, coverage_rate=0.8, rejection_breakdown={ "numeric_inconsistency": 1, "unsupported_claim": 1, }, ), ] metrics = compute_verification_metrics(reports) assert metrics.total_checked == 15 assert metrics.passed_count == 12 assert metrics.failed_count == 3 assert metrics.evidence_coverage_rate == pytest.approx(12 / 15) assert metrics.unsupported_claim_rate == pytest.approx(1 / 15) assert metrics.per_reason_counts["invalid_offset"] == 1 assert metrics.per_reason_counts["numeric_inconsistency"] == 1 assert metrics.per_reason_counts["unsupported_claim"] == 1 def test_empty_reports(self): metrics = compute_verification_metrics([]) assert metrics.total_checked == 0 assert metrics.passed_count == 0 assert metrics.failed_count == 0 assert metrics.evidence_coverage_rate == 1.0 assert metrics.unsupported_claim_rate == 0.0 def test_metrics_is_frozen_dataclass(self): """VerificationMetrics should be immutable.""" metrics = compute_verification_metrics([]) assert isinstance(metrics, VerificationMetrics) # --------------------------------------------------------------------------- # Test: Full verification report # --------------------------------------------------------------------------- class TestFullVerificationReport: def test_all_candidates_verified(self, source_text: str): verifier = EvidenceVerifier() spans = [ EvidenceSpan( id="s1", start_char=0, end_char=10, text=source_text[0:10], ), ] candidates = [ Candidate( candidate_type="entity", candidate_id="c1", candidate_data={"name": "Apple Inc."}, evidence_ids=["s1"], literal_text="Apple Inc.", ), ] report = verifier.verify_all(candidates, spans, source_text) assert report.total_candidates == 1 assert report.verified == 1 assert report.rejected == 0 assert report.coverage_rate == 1.0 def test_mixed_verification(self, source_text: str): verifier = EvidenceVerifier() spans = [ EvidenceSpan( id="s1", start_char=0, end_char=10, text=source_text[0:10], ), EvidenceSpan( id="s2", start_char=11, end_char=58, text=source_text[11:58], ), ] candidates = [ Candidate( candidate_type="entity", candidate_id="c1", candidate_data={"name": "Apple"}, evidence_ids=["s1"], literal_text="Apple Inc.", ), Candidate( candidate_type="entity", candidate_id="c2", candidate_data={"name": "Microsoft"}, evidence_ids=["s1", "s2"], literal_text="Microsoft", ), ] report = verifier.verify_all(candidates, spans, source_text) assert report.total_candidates == 2 assert report.verified == 1 assert report.rejected == 1 assert report.coverage_rate == 0.5 assert RejectionReason.ENTITY_NOT_IN_EVIDENCE.value in report.rejection_breakdown def test_invalid_span_cascades_to_candidate(self, source_text: str): """If a candidate's only span is invalid, the candidate is rejected.""" verifier = EvidenceVerifier() bad_span = EvidenceSpan( id="s-bad", start_char=0, end_char=10, text="WRONG TEXT", # Doesn't match source ) candidates = [ Candidate( candidate_type="entity", candidate_id="c1", candidate_data={"name": "test"}, evidence_ids=["s-bad"], literal_text="Apple", ), ] report = verifier.verify_all(candidates, [bad_span], source_text) assert report.rejected == 1 assert report.verified == 0 def test_numeric_candidate_in_full_report(self, source_text: str): verifier = EvidenceVerifier() spans = [ EvidenceSpan( id="s1", start_char=11, end_char=58, text=source_text[11:58], ), ] candidates = [ Candidate( candidate_type="fact", candidate_id="c1", candidate_data={"type": "revenue"}, evidence_ids=["s1"], literal_text="$94.8 billion", normalized_value=94.8, ), ] report = verifier.verify_all(candidates, spans, source_text) assert report.verified == 1 assert report.rejected == 0