"""Unit tests for canonical merger.""" from __future__ import annotations import copy import pytest from libs.parser.merger import merge, should_queue_for_review _BASE_RULE = { "schema_version": "1.0.0", "document_id": "DOC::test", "parser_kind": "rule", "event_type": "earnings_release", "event_direction": "bullish", "event_date": "2026-01-01", "filing_time_bucket": "post_market", "headline": "", "summary": "Q1 results beat estimates.", "guidance": {"status": "raised", "scope": "annual", "notes": ""}, "signals": { "demand_strength": "strong", "pricing_power": "present", "backlog_or_bookings": "unknown", "customer_expansion": "unknown", "margin_quality": "improving", }, "risk_flags": { "oneoff_item": False, "tax_benefit": False, "valuation_gain": False, "non_gaap_heavy": True, "financing_related": False, "legal_or_regulatory_overhang": False, }, "evidence": [], "confidence": { "overall": 0.80, "event_type": 0.90, "event_direction": 0.75, "guidance": 0.80, "risk_flags": 0.85, }, "warnings": [], } _BASE_LLM = { **copy.deepcopy(_BASE_RULE), "parser_kind": "llm", "headline": "Apple beats Q1 expectations", "summary": "Apple Inc. reported strong Q1 results driven by iPhone sales.", "signals": { "demand_strength": "strong", "pricing_power": "present", "backlog_or_bookings": "present", "customer_expansion": "present", "margin_quality": "improving", }, "confidence": { "overall": 0.85, "event_type": 0.90, "event_direction": 0.80, "guidance": 0.75, "risk_flags": 0.85, }, } @pytest.mark.unit class TestMerge: def test_rule_wins_when_both_confident(self) -> None: """Rule wins on event_type/direction when rule confidence >= 0.70.""" result = merge(_BASE_RULE, _BASE_LLM) assert result["event_type"] == "earnings_release" assert result["event_direction"] == "bullish" assert result["provenance"]["event_type"] == "rule" def test_llm_fills_unknown_signals(self) -> None: """LLM fills in unknown signals that rule left blank.""" result = merge(_BASE_RULE, _BASE_LLM) # backlog_or_bookings was unknown in rule but present in LLM assert result["signals"]["backlog_or_bookings"] == "present" assert result["provenance"]["signals.backlog_or_bookings"] == "llm" # customer_expansion was unknown in rule but present in LLM assert result["signals"]["customer_expansion"] == "present" def test_llm_wins_when_rule_not_confident(self) -> None: """LLM wins on a field when rule confidence is below threshold.""" llm_bearish = copy.deepcopy(_BASE_LLM) llm_bearish["event_direction"] = "bearish" llm_bearish["confidence"]["event_direction"] = 0.80 # LLM confident rule_low = copy.deepcopy(_BASE_RULE) rule_low["confidence"]["event_direction"] = 0.40 # below threshold → not confident result = merge(rule_low, llm_bearish) # Rule not confident → LLM fills in "bearish", no conflict assert result["event_direction"] == "bearish" assert result["provenance"]["event_direction"] == "llm" assert result["rule_llm_conflict"] is False def test_conflict_flags_rule_llm_conflict(self) -> None: """rule_llm_conflict flag is set when both parsers disagree with high confidence.""" llm_conflict = copy.deepcopy(_BASE_LLM) llm_conflict["event_type"] = "guidance_update" # conflicts with "earnings_release" # Both rule and LLM are confident rule_high = copy.deepcopy(_BASE_RULE) rule_high["confidence"]["event_type"] = 0.90 llm_conflict["confidence"]["event_type"] = 0.90 result = merge(rule_high, llm_conflict) # Both confident → rule wins, but conflict flagged assert result["event_type"] == "earnings_release" assert result["rule_llm_conflict"] is True def test_llm_none_all_provenance_is_rule(self) -> None: """With no LLM output, all provenance should be 'rule'.""" result = merge(_BASE_RULE, None) assert result["parser_kind"] == "merged" assert result["rule_llm_conflict"] is False for field, source in result["provenance"].items(): assert source == "rule", f"Expected 'rule' for {field}, got {source}" def test_risk_flags_are_ored(self) -> None: """Risk flags combine with OR: if either parser flags it, it's flagged.""" rule_no_oneoff = copy.deepcopy(_BASE_RULE) rule_no_oneoff["risk_flags"]["oneoff_item"] = False llm_with_oneoff = copy.deepcopy(_BASE_LLM) llm_with_oneoff["risk_flags"]["oneoff_item"] = True result = merge(rule_no_oneoff, llm_with_oneoff) assert result["risk_flags"]["oneoff_item"] is True assert result["provenance"]["risk_flags.oneoff_item"] == "llm" @pytest.mark.unit class TestShouldQueueForReview: def test_no_queue_for_clean_record(self) -> None: """High-confidence, no conflicts, no oneoff flags → no review needed.""" clean = copy.deepcopy(_BASE_RULE) # Remove any flags that would trigger review clean["risk_flags"]["oneoff_item"] = False clean["risk_flags"]["non_gaap_heavy"] = False clean["rule_llm_conflict"] = False clean["provenance"] = {} # type: ignore[typeddict-unknown-key] clean["parser_kind"] = "merged" # event_direction is bullish (not unknown/mixed), confidence is 0.80 should_queue, reasons = should_queue_for_review(clean) assert should_queue is False assert reasons == [] def test_low_confidence_triggers_review(self) -> None: low_conf = copy.deepcopy(_BASE_RULE) low_conf["confidence"]["overall"] = 0.45 low_conf["rule_llm_conflict"] = False should_queue, reasons = should_queue_for_review(low_conf) assert should_queue is True assert "low_confidence" in reasons def test_conflict_triggers_review(self) -> None: conflict = copy.deepcopy(_BASE_RULE) conflict["rule_llm_conflict"] = True should_queue, reasons = should_queue_for_review(conflict) assert should_queue is True assert "rule_llm_conflict" in reasons def test_oneoff_triggers_review(self) -> None: oneoff = copy.deepcopy(_BASE_RULE) oneoff["risk_flags"]["oneoff_item"] = True oneoff["rule_llm_conflict"] = False should_queue, reasons = should_queue_for_review(oneoff) assert should_queue is True assert "oneoff_likely" in reasons