| 1 | import unittest |
| 2 | |
| 3 | from lib import schema |
| 4 | |
| 5 | |
| 6 | class SchemaV3Tests(unittest.TestCase): |
| 7 | def test_report_roundtrip(self): |
| 8 | report = schema.Report( |
| 9 | topic="test topic", |
| 10 | range_from="2026-02-14", |
| 11 | range_to="2026-03-16", |
| 12 | generated_at="2026-03-16T00:00:00+00:00", |
| 13 | provider_runtime=schema.ProviderRuntime( |
| 14 | reasoning_provider="gemini", |
| 15 | planner_model="gemini-3.1-flash-lite", |
| 16 | rerank_model="gemini-3.1-flash-lite", |
| 17 | ), |
| 18 | query_plan=schema.QueryPlan( |
| 19 | intent="breaking_news", |
| 20 | freshness_mode="strict_recent", |
| 21 | cluster_mode="story", |
| 22 | raw_topic="test topic", |
| 23 | subqueries=[schema.SubQuery(label="primary", search_query="test topic", ranking_query="What happened with test topic?", sources=["grounding"])], |
| 24 | source_weights={"grounding": 1.0}, |
| 25 | ), |
| 26 | clusters=[schema.Cluster(cluster_id="cluster-1", title="Title", candidate_ids=["c1"], representative_ids=["c1"], sources=["grounding"], score=90)], |
| 27 | ranked_candidates=[schema.Candidate( |
| 28 | candidate_id="c1", |
| 29 | item_id="i1", |
| 30 | source="grounding", |
| 31 | sources=["grounding", "reddit"], |
| 32 | title="Title", |
| 33 | url="https://example.com", |
| 34 | snippet="Snippet", |
| 35 | subquery_labels=["primary"], |
| 36 | native_ranks={"primary:grounding": 1}, |
| 37 | local_relevance=0.8, |
| 38 | freshness=90, |
| 39 | engagement=None, |
| 40 | source_quality=1.0, |
| 41 | rrf_score=0.02, |
| 42 | rerank_score=91, |
| 43 | final_score=90, |
| 44 | source_items=[ |
| 45 | schema.SourceItem(item_id="i1", source="grounding", title="Title", body="Body", url="https://example.com", published_at="2026-03-16") |
| 46 | ], |
| 47 | )], |
| 48 | items_by_source={"grounding": [schema.SourceItem(item_id="i1", source="grounding", title="Title", body="Body", url="https://example.com")]}, |
| 49 | errors_by_source={}, |
| 50 | warnings=["warning"], |
| 51 | artifacts={"grounding": []}, |
| 52 | ) |
| 53 | restored = schema.report_from_dict(schema.to_dict(report)) |
| 54 | self.assertEqual(report.topic, restored.topic) |
| 55 | self.assertEqual(report.provider_runtime.planner_model, restored.provider_runtime.planner_model) |
| 56 | self.assertEqual(report.ranked_candidates[0].candidate_id, restored.ranked_candidates[0].candidate_id) |
| 57 | self.assertEqual(report.ranked_candidates[0].sources, restored.ranked_candidates[0].sources) |
| 58 | self.assertEqual(report.items_by_source["grounding"][0].title, restored.items_by_source["grounding"][0].title) |
| 59 | |
| 60 | def test_source_item_from_dict_preserves_zero_valued_signals(self): |
| 61 | item = schema.source_item_from_dict( |
| 62 | { |
| 63 | "item_id": "x1", |
| 64 | "source": "x", |
| 65 | "title": "Title", |
| 66 | "body": "Body", |
| 67 | "url": "https://example.com", |
| 68 | "relevance_hint": 0.0, |
| 69 | "local_relevance": 0.0, |
| 70 | "freshness": 0, |
| 71 | "engagement_score": 0, |
| 72 | "source_quality": 0.0, |
| 73 | "local_rank_score": 0.0, |
| 74 | } |
| 75 | ) |
| 76 | self.assertEqual(0.0, item.relevance_hint) |
| 77 | self.assertEqual(0.0, item.local_relevance) |
| 78 | self.assertEqual(0, item.freshness) |
| 79 | self.assertEqual(0, item.engagement_score) |
| 80 | self.assertEqual(0.0, item.source_quality) |
| 81 | self.assertEqual(0.0, item.local_rank_score) |
| 82 | |
| 83 | if __name__ == "__main__": |
| 84 | unittest.main() |
| 85 |