79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
import unittest
|
|
|
|
from ai_daily_report.models import NewsItem, SourceResult
|
|
from ai_daily_report.quality_gate import evaluate_quality_gate
|
|
|
|
|
|
def news_item(item_id, title="Story"):
|
|
return NewsItem(
|
|
id=item_id,
|
|
source_group="AI HOT",
|
|
source_label="AI HOT",
|
|
source_role="primary",
|
|
source_priority=10,
|
|
title_raw=f"{title} {item_id}",
|
|
title_norm=f"{title} {item_id}".lower(),
|
|
summary_raw="summary",
|
|
url=f"https://example.com/{item_id}",
|
|
canonical_url=f"https://example.com/{item_id}",
|
|
)
|
|
|
|
|
|
class QualityGateTests(unittest.TestCase):
|
|
def test_warns_when_stage3_candidates_zero_for_large_item_set(self):
|
|
items = [news_item(str(index)) for index in range(31)]
|
|
report = evaluate_quality_gate(
|
|
items,
|
|
source_results=[],
|
|
reports={"stage3": {"candidate_group_count": 0}},
|
|
config={"warn_when_stage3_candidates_zero_min_items": 30},
|
|
)
|
|
|
|
self.assertIn("stage3_candidates_zero", report["warnings"])
|
|
self.assertEqual(report["blocking_errors"], [])
|
|
|
|
def test_warns_on_enabled_source_failure(self):
|
|
report = evaluate_quality_gate(
|
|
[news_item("a")],
|
|
source_results=[
|
|
SourceResult(
|
|
source="橘鸦AI早报",
|
|
role="supplement",
|
|
ok=False,
|
|
status="error",
|
|
error="HTTPError: 404",
|
|
)
|
|
],
|
|
reports={"stage3": {"candidate_group_count": 1}},
|
|
config={"warn_on_enabled_source_failure": True},
|
|
)
|
|
|
|
self.assertIn("enabled_source_failed:橘鸦AI早报:error", report["warnings"])
|
|
self.assertEqual(report["source_failures"][0]["source"], "橘鸦AI早报")
|
|
|
|
def test_blocks_required_source_failure_when_configured(self):
|
|
report = evaluate_quality_gate(
|
|
[news_item("a")],
|
|
source_results=[
|
|
SourceResult(
|
|
source="AI HOT",
|
|
role="primary",
|
|
ok=False,
|
|
status="timeout",
|
|
error="TimeoutError",
|
|
)
|
|
],
|
|
reports={"stage3": {"candidate_group_count": 1}},
|
|
config={
|
|
"block_on_required_source_failure": True,
|
|
"required_sources": ["AI HOT"],
|
|
},
|
|
)
|
|
|
|
self.assertIn("required_source_failed:AI HOT:timeout", report["blocking_errors"])
|
|
self.assertTrue(report["quality_gate_failed"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|