import pytest from pydantic import ValidationError from agent.schemas import ( SourcedFact, BriefOutput, TrendPoint, ManagementCommentaryTopic, MDASection, CategorizedRisk, GuidancePoint, EarningsQualitySignal, AnalyticalTension, SectionSentiment, SubtextRead, MarketExpectations, ) from analysis.textdiff import _detect_trend def test_sourced_fact_valid(): fact = SourcedFact(text="Revenue grew 5%", source="10-Q", reliability="HIGH", evidence_snippet="Revenue grew 5%") assert fact.source == "10-Q" assert fact.reliability == "HIGH" assert fact.evidence_snippet == "Revenue grew 5%" def test_sourced_fact_rejects_invalid_source(): with pytest.raises(ValidationError): SourcedFact(text="x", source="bloomberg", reliability="HIGH") def test_sourced_fact_rejects_invalid_reliability(): with pytest.raises(ValidationError): SourcedFact(text="x", source="10-Q", reliability="VERY_HIGH") def test_sourced_fact_requires_evidence_snippet(): with pytest.raises(ValidationError): SourcedFact(text="x", source="10-Q", reliability="HIGH") def test_market_expectations_cannot_self_attest_alignment_without_evidence(): expectations = MarketExpectations( consensus_eps_est=2.5, consensus_rev_est_bn=10.0, revision_30d_pct=4.2, period_aligned=True, comparison_allowed=True, d1_price_reaction_pct=88.0, event_aligned=True, event_comparison_allowed=True, rationale="Claimed alignment without evidence.", ) assert expectations.period_aligned is False assert expectations.comparison_allowed is False assert expectations.event_aligned is False assert expectations.event_comparison_allowed is False assert expectations.consensus_eps_est is None assert expectations.consensus_rev_est_bn is None assert expectations.revision_30d_pct is None assert expectations.d1_price_reaction_pct is None def _minimal_brief(**overrides) -> BriefOutput: """Build a minimal but fully valid BriefOutput for testing.""" sf = SourcedFact(text="Revenue up 5%", source="10-Q", reliability="HIGH", evidence_snippet="Revenue up 5%") defaults = dict( ticker="AAPL", company_name="Apple Inc.", filing_date="2024-11-01", what_matters_most="iPhone demand remains strong.", standout_number=sf, what_changed=[sf], bull_points=[sf], bear_points=[sf], what_to_watch=["Q2 iPhone shipments"], trends=[TrendPoint(period="Q3 2025", revenue_bn=50.5)], mda_summary=MDASection( drivers=[sf], headwinds=[sf], language_shift="Tone unchanged.", key_quote=sf, ), risks_categorized=[CategorizedRisk( category="Macro", text="Macro risk.", source="10-K", reliability="HIGH", is_new_this_filing=False, )], management_commentary=[ManagementCommentaryTopic( topic="Revenue", summary="Revenue grew.", source="10-Q", reliability="HIGH", evidence_snippet="Revenue grew 5%.", )], guidance_history=[GuidancePoint(period="Q3 2025", text="Revenue guided flat.", source="10-Q")], ) defaults.update(overrides) return BriefOutput(**defaults) def test_brief_output_valid(): brief = _minimal_brief( what_changed=[SourcedFact(text="Revenue up 5%", source="10-Q", reliability="HIGH", evidence_snippet="Revenue up 5%")], bull_points=[SourcedFact(text="Services growing", source="transcript", reliability="MEDIUM", evidence_snippet="Services growing")], bear_points=[SourcedFact(text="China headwinds", source="news", reliability="LOW", evidence_snippet="China headwinds")], what_to_watch=["Q2 iPhone shipments", "AI feature adoption"], evidence_notes=["Revenue growth corroborated by filing and transcript."], ) assert brief.ticker == "AAPL" assert len(brief.what_changed) == 1 assert brief.what_to_watch == ["Q2 iPhone shipments", "AI feature adoption"] def test_brief_output_tolerates_truncated_content_hash_in_evidence_ref(): truncated_hash = "496fb91760e2df7fe4e59bb606127879006da861385c98" evidence_ref = { "evidence_id": "ev_abc123", "source": "10-Q", "content_hash": truncated_hash, "document_id": "sec:AAPL:0001", } payload = _minimal_brief().model_dump(mode="json") payload["bear_points"] = [{ "text": "China headwinds persist.", "source": "10-Q", "reliability": "HIGH", "evidence_snippet": "China headwinds persist.", "evidence_ref": evidence_ref, }] payload["risks_categorized"] = [{ "category": "Demand", "text": "China demand may weaken.", "source": "10-Q", "reliability": "HIGH", "is_new_this_filing": False, "evidence_snippet": "China demand may weaken.", "evidence_ref": evidence_ref, }] brief = BriefOutput.model_validate(payload) assert brief.bear_points[0].evidence_ref.content_hash == truncated_hash assert brief.risks_categorized[0].evidence_ref.content_hash == truncated_hash def test_brief_output_drops_only_invalid_bull_point(capsys): payload = _minimal_brief().model_dump(mode="json") payload["bull_points"] = [ { "text": "Services revenue grew.", "source": "10-Q", "reliability": "HIGH", "evidence_snippet": "Services revenue grew.", }, { "text": "Unusable composite source.", "source": "bloomberg", "reliability": "LOW", "evidence_snippet": "Unusable composite source.", }, { "text": "Margins expanded.", "source": "transcript", "reliability": "MEDIUM", "evidence_snippet": "Margins expanded.", }, ] brief = BriefOutput.model_validate(payload) assert [item.text for item in brief.bull_points] == [ "Services revenue grew.", "Margins expanded.", ] assert "[brief-sanitizer] dropping invalid bull_points item:" in capsys.readouterr().err def test_brief_output_invalid_standout_number_becomes_none(): payload = _minimal_brief().model_dump(mode="json") payload["standout_number"] = { "source": "10-Q", "reliability": "HIGH", "evidence_snippet": "Revenue grew.", } brief = BriefOutput.model_validate(payload) assert brief.standout_number is None def test_brief_output_evidence_ref_ignores_extra_content_key(): payload = _minimal_brief().model_dump(mode="json") payload["bull_points"][0]["evidence_ref"] = { "evidence_id": "ev_abc123", "source": "10-Q", "content_hash": "short-hash-is-tolerated", "document_id": "sec:AAPL:0001", "content": "This belongs to an evidence record, not an evidence ref.", } brief = BriefOutput.model_validate(payload) assert brief.bull_points[0].evidence_ref.evidence_id == "ev_abc123" assert not hasattr(brief.bull_points[0].evidence_ref, "content") def test_brief_output_defaults_llm_singletons_and_status(capsys): payload = _minimal_brief().model_dump(mode="json") payload.pop("standout_number") payload.pop("mda_summary") payload["status"] = {"unexpected": "shape"} brief = BriefOutput.model_validate(payload) assert brief.standout_number is None assert brief.mda_summary == MDASection() assert brief.status == "PARTIAL" assert "[brief-status]" in capsys.readouterr().err def test_brief_output_accepts_metrics_sources_for_risk_and_guidance(): brief = _minimal_brief( risks_categorized=[dict( category="Financial", text="Margin compression risk.", source="metrics", reliability="HIGH", is_new_this_filing=False, )], guidance_history=[dict( period="Q3 2025", text="Revenue guidance narrowed.", source="metrics", )], ) assert brief.risks_categorized[0].source == "metrics" assert brief.guidance_history[0].source == "metrics" def test_sourced_fact_ignores_extra_fields(): # ConfigDict(extra="ignore") — extra fields are silently dropped, not rejected. # This is intentional: LLM output may include unexpected keys. fact = SourcedFact( text="Revenue grew 5%", source="10-Q", reliability="HIGH", evidence_snippet="Revenue grew", unexpected_field="oops", ) assert fact.text == "Revenue grew 5%" assert not hasattr(fact, "unexpected_field") def test_trend_point_valid(): tp = TrendPoint(period="Q3 2025", revenue_bn=50.5, revenue_yoy_pct=8.2, operating_margin=0.31, eps=1.52) assert tp.period == "Q3 2025" assert tp.revenue_bn == 50.5 def test_trend_point_all_optional_metrics(): tp = TrendPoint(period="Q1 2024") assert tp.revenue_bn is None def test_brief_output_evidence_notes_defaults_empty(): # evidence_notes omitted — should default to [] brief = _minimal_brief(filing_date="2025-01-01") assert brief.evidence_notes == [] # ── ManagementCommentaryTopic ──────────────────────────────────────────────── def test_management_commentary_topic_from_filing(): t = ManagementCommentaryTopic( topic="Revenue drivers", summary="Revenue grew 5% driven by Services.", source="10-Q", reliability="HIGH", evidence_snippet="Revenue grew 5% driven by Services.", ) assert t.source == "10-Q" assert t.reliability == "HIGH" def test_management_commentary_topic_accepts_transcript(): t = ManagementCommentaryTopic( topic="AI roadmap", summary="Management highlighted upcoming AI features.", source="transcript", reliability="MEDIUM", evidence_snippet="We are incredibly excited about Apple Intelligence.", ) assert t.source == "transcript" def test_management_commentary_topic_rejects_news_source(): with pytest.raises(ValidationError): ManagementCommentaryTopic( topic="Analyst reaction", summary="Morgan Stanley raised price target.", source="news", reliability="LOW", evidence_snippet="Morgan Stanley raised target to $250.", ) def test_management_commentary_topic_rejects_metrics_source(): with pytest.raises(ValidationError): ManagementCommentaryTopic( topic="Revenue guidance", summary="Revenue guidance was updated.", source="metrics", reliability="HIGH", evidence_snippet="Revenue guidance was updated.", ) def test_management_commentary_topic_trims_long_snippet(): snippet = " ".join([f"word{i}" for i in range(40)]) t = ManagementCommentaryTopic( topic="Long quote", summary="Test.", source="10-K", reliability="HIGH", evidence_snippet=snippet, ) assert len(t.evidence_snippet.split()) <= 30 def test_brief_output_has_management_commentary_field(): assert "management_commentary" in BriefOutput.model_fields assert "transcript_topics" not in BriefOutput.model_fields def test_brief_output_drops_news_commentary_mixed(): """BriefOutput drops news-sourced items but keeps valid ones (filing/transcript).""" brief = _minimal_brief( management_commentary=[ {"topic": "AI strategy", "summary": "Management outlined AI plans.", "source": "transcript", "reliability": "MEDIUM", "evidence_snippet": "We are fully committed to AI integration."}, {"topic": "Breaking news", "summary": "CNBC reported earnings beat.", "source": "news", "reliability": "LOW", "evidence_snippet": "CNBC reported strong results."}, {"topic": "Revenue guidance", "summary": "Guided 3-5% growth.", "source": "10-Q", "reliability": "HIGH", "evidence_snippet": "We expect revenue growth of 3-5%."}, ] ) assert len(brief.management_commentary) == 2 topics = [t.topic for t in brief.management_commentary] assert "AI strategy" in topics assert "Revenue guidance" in topics assert "Breaking news" not in topics def test_brief_output_all_news_commentary_yields_empty_list(): """BriefOutput validates even when all management_commentary items are news-sourced (MSFT regression).""" brief = _minimal_brief( management_commentary=[ {"topic": f"Item {i}", "summary": "News summary.", "source": "news", "reliability": "LOW", "evidence_snippet": f"News snippet {i}."} for i in range(5) ] ) assert brief.management_commentary == [] # ── CategorizedRisk category normalization ─────────────────────────────────── def _risk(**kw) -> CategorizedRisk: defaults = dict(text="Some risk.", source="10-K", reliability="HIGH", is_new_this_filing=False) return CategorizedRisk(**(defaults | kw)) def test_categorized_risk_accepts_geopolitical(): r = _risk(category="Geopolitical") assert r.category == "Geopolitical" def test_categorized_risk_accepts_demand(): r = _risk(category="Demand") assert r.category == "Demand" def test_categorized_risk_accepts_metrics_source(): r = _risk(category="Financial", source="metrics") assert r.source == "metrics" def test_categorized_risk_accepts_analyst_source(): r = _risk(category="Financial", source="analyst") assert r.source == "analyst" def test_categorized_risk_case_folds_lowercase(): r = _risk(category="regulatory") assert r.category == "Regulatory" def test_categorized_risk_alias_legal_to_regulatory(): r = _risk(category="Legal") assert r.category == "Regulatory" def test_categorized_risk_alias_cybersecurity_to_operational(): r = _risk(category="Cybersecurity") assert r.category == "Operational" def test_categorized_risk_alias_tariffs_to_geopolitical(): r = _risk(category="Tariffs") assert r.category == "Geopolitical" def test_categorized_risk_unknown_coerces_to_operational_with_warning(capsys): r = _risk(category="Quantum") assert r.category == "Operational" captured = capsys.readouterr() assert "[risk-category]" in captured.err assert "Quantum" in captured.err def test_brief_output_validates_with_geopolitical_risk(): brief = _minimal_brief( risks_categorized=[CategorizedRisk( category="Geopolitical", text="Trade war escalation risk.", source="10-K", reliability="HIGH", is_new_this_filing=True, )] ) assert brief.risks_categorized[0].category == "Geopolitical" # ── _normalize_source — SourcedFact ───────────────────────────────────────── def _sf(source: str) -> SourcedFact: return SourcedFact(text="x", source=source, reliability="HIGH", evidence_snippet="x") def test_sourced_fact_rejects_compound_10q_transcript(): with pytest.raises(ValidationError): _sf("10-Q, transcript") def test_sourced_fact_coerces_compound_transcript_10q(): # filing takes precedence regardless of order with pytest.raises(ValidationError): _sf("transcript, 10-Q") def test_sourced_fact_coerces_compound_10k_transcript(): with pytest.raises(ValidationError): _sf("10-K, transcript") def test_sourced_fact_coerces_compound_10q_news(): with pytest.raises(ValidationError): _sf("10-Q, news") def test_sourced_fact_coerces_compound_10q_10k_first_wins(): # '10-Q' appears before '10-K' in this string → 10-Q with pytest.raises(ValidationError): _sf("10-Q, 10-K") def test_sourced_fact_coerces_compound_10k_10q_first_wins(): # '10-K' appears before '10-Q' in this string → 10-K with pytest.raises(ValidationError): _sf("10-K, 10-Q") def test_sourced_fact_coerces_compound_transcript_news(): with pytest.raises(ValidationError): _sf("transcript, news") def test_sourced_fact_coerces_lowercase_10q(): # case-folding: "10-q" → "10-Q" assert _sf("10-q").source == "10-Q" def test_sourced_fact_regression_invalid_source_still_rejected(): # _normalize_source passes unknown strings through → Literal still rejects them with pytest.raises(ValidationError): _sf("bloomberg") # ── _normalize_source — other models ──────────────────────────────────────── def test_categorized_risk_coerces_compound_source(): with pytest.raises(ValidationError): CategorizedRisk( category="Macro", text="Macro risk.", source="10-Q, 10-K", reliability="HIGH", is_new_this_filing=False, ) def test_management_commentary_coerces_compound_source(): with pytest.raises(ValidationError): ManagementCommentaryTopic( topic="AI capex", summary="Capex rising.", source="transcript, 10-Q", reliability="HIGH", evidence_snippet="Capex rising.", ) def test_guidance_point_coerces_compound_source(): with pytest.raises(ValidationError): GuidancePoint(period="Q3 2025", text="Revenue guided flat.", source="10-K, transcript") # ── BriefOutput end-to-end: reproduces the original NVDA failure ───────────── def test_brief_output_tolerates_compound_source_in_analytical_tension(): """Reproduce the NVDA 14-error failure: compound source inside a nested SourcedFact.""" from agent.schemas import AnalyticalTension tension = AnalyticalTension( headline="Data center growth strong but sequential deceleration notable.", bullish_reading="Revenue beat consensus by 4%.", bearish_reading="Sequential growth slowing despite beat.", weight="watch", bullish_evidence=SourcedFact( text="Revenue grew 78% YoY.", source="10-Q", reliability="HIGH", evidence_snippet="Revenue grew 78% YoY driven by data center.", ), bearish_evidence=SourcedFact( text="Sequential growth decelerated.", source="10-Q", reliability="HIGH", evidence_snippet="Sequential revenue growth slowed to 8%.", ), ) assert tension.bullish_evidence.source == "10-Q" assert tension.bearish_evidence.source == "10-Q" brief = _minimal_brief(analytical_tensions=[tension]) assert brief.analytical_tensions[0].bullish_evidence.source == "10-Q" assert brief.analytical_tensions[0].bearish_evidence.source == "10-Q" # ── EarningsQualitySignal.assessment ───────────────────────────────────────── def _eqs(**kw) -> EarningsQualitySignal: defaults = dict( dimension="guidance_dynamics", assessment="positive", rationale="Guidance was raised.", evidence=SourcedFact(text="x", source="10-Q", reliability="HIGH", evidence_snippet="x"), ) return EarningsQualitySignal(**(defaults | kw)) def test_eqs_assessment_valid_passthrough(): assert _eqs(assessment="concerning").assessment == "concerning" assert _eqs(assessment="positive").assessment == "positive" assert _eqs(assessment="neutral").assessment == "neutral" def test_eqs_assessment_coerces_negative_to_concerning(capsys): s = _eqs(assessment="negative") assert s.assessment == "concerning" assert "[quality-assessment]" in capsys.readouterr().err def test_eqs_assessment_coerces_bearish_to_concerning(): assert _eqs(assessment="bearish").assessment == "concerning" def test_eqs_assessment_coerces_bullish_to_positive(): assert _eqs(assessment="bullish").assessment == "positive" def test_eqs_assessment_coerces_mixed_to_neutral(): assert _eqs(assessment="mixed").assessment == "neutral" def test_eqs_assessment_unknown_falls_back_to_neutral(capsys): s = _eqs(assessment="unclear") assert s.assessment == "neutral" assert "[quality-assessment]" in capsys.readouterr().err # ── EarningsQualitySignal.dimension ────────────────────────────────────────── def test_eqs_dimension_valid_passthrough(): assert _eqs(dimension="consensus_beat_mix").dimension == "consensus_beat_mix" assert _eqs(dimension="capital_allocation").dimension == "capital_allocation" def test_eqs_dimension_coerces_guidance_synonym(capsys): s = _eqs(dimension="guidance") assert s.dimension == "guidance_dynamics" assert "[quality-dimension]" in capsys.readouterr().err def test_eqs_dimension_coerces_beat_mix_synonym(): assert _eqs(dimension="beat_mix").dimension == "consensus_beat_mix" def test_eqs_dimension_coerces_narrative_synonym(): assert _eqs(dimension="tone").dimension == "narrative_vs_numbers" def test_eqs_dimension_coerces_segment_synonym(): assert _eqs(dimension="segment").dimension == "segment_mix" def test_eqs_dimension_coerces_capital_synonym(): assert _eqs(dimension="capex").dimension == "capital_allocation" def test_eqs_dimension_unknown_still_raises(): """Unknown dimension passes through _normalize_dimension unchanged → Literal rejects it.""" with pytest.raises(ValidationError): _eqs(dimension="completely_unknown_dim") # ── BriefOutput._sanitize_quality_signals ──────────────────────────────────── def test_brief_output_drops_unknown_dimension_keeps_valid(capsys): """Unknown-dimension item is dropped at brief level; valid item is kept.""" sf = {"text": "x", "source": "10-Q", "reliability": "HIGH", "evidence_snippet": "x"} brief = _minimal_brief( earnings_quality_signals=[ {"dimension": "totally_unknown", "assessment": "positive", "rationale": "r", "evidence": sf}, {"dimension": "guidance_dynamics", "assessment": "neutral", "rationale": "r", "evidence": sf}, ] ) assert len(brief.earnings_quality_signals) == 1 assert brief.earnings_quality_signals[0].dimension == "guidance_dynamics" assert "[quality-dimension]" in capsys.readouterr().err def test_brief_output_all_unknown_dimension_yields_empty_list(): """All items with unknown dimension → empty list, brief still validates.""" sf = {"text": "x", "source": "10-Q", "reliability": "HIGH", "evidence_snippet": "x"} brief = _minimal_brief( earnings_quality_signals=[ {"dimension": "foo", "assessment": "positive", "rationale": "r", "evidence": sf}, {"dimension": "bar", "assessment": "neutral", "rationale": "r", "evidence": sf}, ] ) assert brief.earnings_quality_signals == [] # ── NVDA regression: the exact 5-error failure from 2026-06-10 ─────────────── def test_eqs_dimension_coerces_tax_rate_and_effective_leverage(): """tax_rate_and_effective_leverage → consensus_beat_mix (below-the-line alias).""" s = _eqs(dimension="tax_rate_and_effective_leverage") assert s.dimension == "consensus_beat_mix" def test_eqs_dimension_coerces_tax_rate(): assert _eqs(dimension="tax_rate").dimension == "consensus_beat_mix" def test_eqs_dimension_coerces_effective_leverage(): assert _eqs(dimension="effective_leverage").dimension == "consensus_beat_mix" def test_eqs_dimension_coerces_earnings_quality(): assert _eqs(dimension="earnings_quality").dimension == "consensus_beat_mix" def test_guidance_verdict_compound_becomes_none(capsys): """Compound LLM verdict like 'beat_revenue_missed_eps' coerces to None.""" g = _gp(verdict="beat_revenue_missed_eps") assert g.verdict is None assert "[guidance-verdict]" in capsys.readouterr().err def test_brief_output_nvda_5error_payload_validates(): """End-to-end reproduction of the NVDA 5-validation-error failure. After all three fixes: - management_commentary items with source='news' → dropped (empty list) - earnings_quality_signals.4 with dimension='tax_rate_and_effective_leverage' → coerced to 'consensus_beat_mix', NOT dropped - guidance_history.1 with verdict='beat_revenue_missed_eps' → verdict=None """ sf = {"text": "t", "source": "10-Q", "reliability": "HIGH", "evidence_snippet": "e"} base = _minimal_brief().model_dump() base["management_commentary"] = [ {"topic": f"T{i}", "summary": "s", "source": "news", "reliability": "LOW", "evidence_snippet": f"snip {i}"} for i in range(3) ] base["earnings_quality_signals"] = [ {"dimension": "guidance_dynamics", "assessment": "positive", "rationale": "r", "evidence": sf}, {"dimension": "narrative_vs_numbers", "assessment": "neutral", "rationale": "r", "evidence": sf}, {"dimension": "segment_mix", "assessment": "positive", "rationale": "r", "evidence": sf}, {"dimension": "capital_allocation", "assessment": "neutral", "rationale": "r", "evidence": sf}, {"dimension": "tax_rate_and_effective_leverage", "assessment": "neutral", "rationale": "r", "evidence": sf}, ] base["guidance_history"] = [ {"period": "Q1 2025", "text": "g", "source": "10-Q", "verdict": "beat"}, {"period": "Q4 2024", "text": "g", "source": "10-Q", "verdict": "beat_revenue_missed_eps"}, ] brief = BriefOutput.model_validate(base) assert brief.management_commentary == [], "news-sourced items must be dropped" dims = [s.dimension for s in brief.earnings_quality_signals] assert "consensus_beat_mix" in dims, "tax_rate_and_effective_leverage must coerce to consensus_beat_mix" assert len(brief.earnings_quality_signals) == 5, "all 5 signals kept (no drop)" assert brief.guidance_history[1].verdict is None, "compound verdict must coerce to None" # ── AnalyticalTension.weight ────────────────────────────────────────────────── def _tension(**kw) -> AnalyticalTension: sf = SourcedFact(text="x", source="10-Q", reliability="HIGH", evidence_snippet="x") defaults = dict( headline="H", bullish_reading="B", bearish_reading="Be", weight="watch", bullish_evidence=sf, bearish_evidence=sf, ) return AnalyticalTension(**(defaults | kw)) def test_tension_weight_valid_passthrough(): assert _tension(weight="material").weight == "material" assert _tension(weight="watch").weight == "watch" assert _tension(weight="minor").weight == "minor" def test_tension_weight_coerces_high_to_material(capsys): t = _tension(weight="high") assert t.weight == "material" assert "[tension-weight]" in capsys.readouterr().err def test_tension_weight_coerces_medium_to_watch(): assert _tension(weight="medium").weight == "watch" def test_tension_weight_coerces_low_to_minor(): assert _tension(weight="low").weight == "minor" def test_tension_weight_coerces_critical_to_material(): assert _tension(weight="critical").weight == "material" def test_tension_weight_unknown_falls_back_to_watch(capsys): t = _tension(weight="extreme") assert t.weight == "watch" assert "[tension-weight]" in capsys.readouterr().err # ── GuidancePoint.verdict ───────────────────────────────────────────────────── def _gp(**kw) -> GuidancePoint: defaults = dict(period="Q1 2025", text="Revenue guided flat.", source="10-Q") return GuidancePoint(**(defaults | kw)) def test_guidance_point_accepts_metrics_source(): g = _gp(source="metrics") assert g.source == "metrics" def test_guidance_point_accepts_analyst_source(): g = _gp(source="analyst") assert g.source == "analyst" def test_guidance_verdict_valid_passthrough(): assert _gp(verdict="beat").verdict == "beat" assert _gp(verdict="in-line").verdict == "in-line" assert _gp(verdict="missed").verdict == "missed" assert _gp(verdict="pending").verdict == "pending" def test_guidance_verdict_none_passthrough(): assert _gp(verdict=None).verdict is None def test_guidance_verdict_coerces_miss_to_missed(capsys): g = _gp(verdict="miss") assert g.verdict == "missed" assert "[guidance-verdict]" in capsys.readouterr().err def test_guidance_verdict_coerces_inline_to_in_line(): assert _gp(verdict="inline").verdict == "in-line" def test_guidance_verdict_coerces_met_to_in_line(): assert _gp(verdict="met").verdict == "in-line" def test_guidance_verdict_coerces_above_to_beat(): assert _gp(verdict="above").verdict == "beat" def test_guidance_verdict_unknown_becomes_none(capsys): g = _gp(verdict="partial") assert g.verdict is None assert "[guidance-verdict]" in capsys.readouterr().err # ── SectionSentiment.score ──────────────────────────────────────────────────── def _ss(**kw) -> SectionSentiment: defaults = dict(rationale="Revenue beat.") return SectionSentiment(**(defaults | kw)) def test_sentiment_score_valid_passthrough(): for s in (-2, -1, 0, 1, 2): assert _ss(score=s).score == s def test_sentiment_score_none_passthrough(): assert _ss(score=None).score is None def test_sentiment_score_coerces_string_int(capsys): s = _ss(score="1") assert s.score == 1 assert "[sentiment-score]" in capsys.readouterr().err def test_sentiment_score_coerces_positive_string(): assert _ss(score="+2").score == 2 def test_sentiment_score_coerces_float(): assert _ss(score=1.0).score == 1 def test_sentiment_score_clamps_out_of_range(capsys): s = _ss(score=3) assert s.score == 2 assert "[sentiment-score]" in capsys.readouterr().err def test_sentiment_score_non_numeric_becomes_none(capsys): s = _ss(score="abc") assert s.score is None assert "[sentiment-score]" in capsys.readouterr().err # ── SectionSentiment.label ──────────────────────────────────────────────────── def test_sentiment_label_valid_passthrough(): for lbl in ("Strongly Bearish", "Bearish", "Neutral", "Bullish", "Strongly Bullish"): assert _ss(label=lbl).label == lbl def test_sentiment_label_none_passthrough(): assert _ss(label=None).label is None def test_sentiment_label_coerces_positive_to_bullish(capsys): s = _ss(label="positive") assert s.label == "Bullish" assert "[sentiment-label]" in capsys.readouterr().err def test_sentiment_label_coerces_negative_to_bearish(): assert _ss(label="negative").label == "Bearish" def test_sentiment_label_unknown_becomes_none(capsys): s = _ss(label="meh") assert s.label is None assert "[sentiment-label]" in capsys.readouterr().err # ── SubtextRead ────────────────────────────────────────────────────────────── def _str(signal_type="language_drift", **kw) -> SubtextRead: sf = SourcedFact(text="x", source="10-Q", reliability="HIGH", evidence_snippet="x") defaults = dict( observation="Management used 'we expect growth' instead of 'we expect strong growth'.", reading="The softening qualifier signals reduced conviction in guidance.", signal_type=signal_type, implication="Watch for a guidance cut next quarter if this trend continues.", evidence=sf, ) defaults.update(kw) return SubtextRead(**defaults) def test_subtext_read_valid_signal_types(): for st in ("language_drift", "qa_evasion", "omission", "emphasis_shift", "accounting_quality"): assert _str(signal_type=st).signal_type == st def test_subtext_read_coerces_evasion_to_qa_evasion(capsys): s = _str(signal_type="evasion") assert s.signal_type == "qa_evasion" assert "[signal-type]" in capsys.readouterr().err def test_subtext_read_coerces_drift_to_language_drift(): assert _str(signal_type="drift").signal_type == "language_drift" def test_subtext_read_coerces_kpi_dropped_to_emphasis_shift(): assert _str(signal_type="kpi_dropped").signal_type == "emphasis_shift" def test_subtext_read_coerces_silence_to_omission(): assert _str(signal_type="silence").signal_type == "omission" def test_subtext_read_coerces_accounting_quality_alias(): assert _str(signal_type="earnings_quality").signal_type == "accounting_quality" def test_subtext_read_unknown_falls_back_to_language_drift(capsys): s = _str(signal_type="completely_unknown_signal") assert s.signal_type == "language_drift" assert "[signal-type]" in capsys.readouterr().err def test_subtext_read_uppercase_canonical_coerced(capsys): s = _str(signal_type="Accounting_Quality") assert s.signal_type == "accounting_quality" captured = capsys.readouterr() assert captured.err == "" # .lower() hits canonical directly — no coercion warning def test_subtext_read_evidence_is_sourced_fact(): s = _str() assert isinstance(s.evidence, SourcedFact) def test_brief_output_between_the_lines_empty_by_default(): brief = _minimal_brief() assert brief.between_the_lines == [] def test_brief_output_accepts_between_the_lines_list(): sf = {"text": "x", "source": "10-Q", "reliability": "HIGH", "evidence_snippet": "x"} item = { "observation": "Analyst asked about China; management pivoted.", "reading": "Evasion signals China pricing under pressure.", "signal_type": "qa_evasion", "implication": "Watch for China revenue disclosure next quarter.", "evidence": sf, } brief = _minimal_brief(between_the_lines=[item]) assert len(brief.between_the_lines) == 1 assert brief.between_the_lines[0].signal_type == "qa_evasion" # ── _detect_trend ──────────────────────────────────────────────────────────── def test_detect_trend_rising_4(): assert _detect_trend([1, 3, 5, 8]) == "rising 4 quarters" def test_detect_trend_falling_4(): assert _detect_trend([8, 5, 3, 1]) == "falling 4 quarters" def test_detect_trend_rising_3_at_tail(): assert _detect_trend([1, 5, 2, 4, 6]) == "rising 3 quarters" def test_detect_trend_flat_returns_none(): assert _detect_trend([1, 2, 2, 4]) is None def test_detect_trend_too_short_returns_none(): assert _detect_trend([1, 3]) is None def test_detect_trend_rising_exactly_3(): """Exactly 3 strictly increasing values — the minimum for a valid trend.""" assert _detect_trend([1, 2, 3]) == "rising 3 quarters" def test_detect_trend_run_length_2_returns_none(): """Tail run of exactly 2 — below the 3-quarter minimum threshold.""" assert _detect_trend([1, 2, 1, 2]) is None def test_detect_trend_empty_returns_none(): assert _detect_trend([]) is None def test_detect_trend_single_value_returns_none(): assert _detect_trend([5]) is None def test_detect_trend_falling_3_at_tail(): assert _detect_trend([10, 2, 8, 6, 4]) == "falling 3 quarters"