Spaces:
Sleeping
Sleeping
| """tests/test_brief_diff.py — unit tests for analytics/brief_diff.py.""" | |
| from __future__ import annotations | |
| from analytics.brief_diff import diff_briefs | |
| def _brief(risks=None, sentiment_label=None, generated_at="2026-02-01T00:00:00Z"): | |
| return { | |
| "generated_at": generated_at, | |
| "risks_categorized": risks or [], | |
| "sentiment": {"metrics": {"score": 1, "label": sentiment_label}} if sentiment_label else {}, | |
| } | |
| def test_diff_briefs_flags_new_signal(): | |
| previous = _brief(risks=[{"text": "Existing supply chain constraint on GPU output.", "impact": "HIGH"}]) | |
| current = _brief(risks=[ | |
| {"text": "Existing supply chain constraint on GPU output.", "impact": "HIGH"}, | |
| {"text": "New China export control risk foreclosing data center market access.", "impact": "HIGH"}, | |
| ]) | |
| result = diff_briefs(current, previous) | |
| assert any("China export control" in s for s in result["new_signals"]) | |
| assert result["resolved_signals"] == [] | |
| def test_diff_briefs_flags_resolved_signal(): | |
| previous = _brief(risks=[ | |
| {"text": "Existing supply chain constraint on GPU output.", "impact": "HIGH"}, | |
| {"text": "Pending litigation risk over patent infringement claims.", "impact": "HIGH"}, | |
| ]) | |
| current = _brief(risks=[{"text": "Existing supply chain constraint on GPU output.", "impact": "HIGH"}]) | |
| result = diff_briefs(current, previous) | |
| assert result["new_signals"] == [] | |
| assert any("litigation" in s for s in result["resolved_signals"]) | |
| def test_diff_briefs_ignores_minor_rewording(): | |
| previous = _brief(risks=[{"text": "Gross margin compressed 12.2pp sequentially to 60.5% amid cost inflation.", "impact": "HIGH"}]) | |
| current = _brief(risks=[{"text": "Gross margin compressed 15pp sequentially to 58% amid cost inflation.", "impact": "HIGH"}]) | |
| result = diff_briefs(current, previous) | |
| # Same underlying fact, different rounding — should NOT show up as new/resolved. | |
| assert result["new_signals"] == [] | |
| assert result["resolved_signals"] == [] | |
| def test_diff_briefs_detects_sentiment_shift(): | |
| previous = _brief(sentiment_label="Bearish") | |
| current = _brief(sentiment_label="Bullish") | |
| result = diff_briefs(current, previous) | |
| assert result["sentiment_shift"] == {"previous": "Bearish", "current": "Bullish"} | |
| def test_diff_briefs_no_sentiment_shift_when_unchanged(): | |
| previous = _brief(sentiment_label="Bullish") | |
| current = _brief(sentiment_label="Bullish") | |
| result = diff_briefs(current, previous) | |
| assert result["sentiment_shift"] is None | |
| def test_diff_briefs_caps_lists_at_five(): | |
| topics = [ | |
| "China export control risk foreclosing data center market access", | |
| "Litigation exposure over alleged patent infringement claims", | |
| "Supply chain concentration risk among key foundry partners", | |
| "Currency translation headwind from strengthening dollar exposure", | |
| "Regulatory scrutiny over antitrust practices in cloud computing", | |
| "Customer concentration risk among top hyperscaler accounts", | |
| "Cybersecurity incident disclosure obligations under new rules", | |
| "Labor relations dispute affecting manufacturing capacity", | |
| ] | |
| risks = [{"text": topic, "impact": "HIGH"} for topic in topics] | |
| previous = _brief(risks=[]) | |
| current = _brief(risks=risks) | |
| result = diff_briefs(current, previous) | |
| assert len(result["new_signals"]) == 5 | |
| def test_diff_briefs_carries_previous_generated_at(): | |
| previous = _brief(generated_at="2026-01-15T00:00:00Z") | |
| current = _brief(generated_at="2026-02-01T00:00:00Z") | |
| result = diff_briefs(current, previous) | |
| assert result["previous_generated_at"] == "2026-01-15T00:00:00Z" | |