| 1 | import tempfile |
| 2 | import unittest |
| 3 | from datetime import datetime, timedelta, timezone |
| 4 | from pathlib import Path |
| 5 | from unittest import mock |
| 6 | |
| 7 | import briefing |
| 8 | import store |
| 9 | |
| 10 | |
| 11 | class BriefingV3Tests(unittest.TestCase): |
| 12 | def test_generate_daily_uses_utc_for_last_run(self): |
| 13 | with tempfile.TemporaryDirectory() as tmpdir: |
| 14 | db_path = Path(tmpdir) / "research.db" |
| 15 | briefs_dir = Path(tmpdir) / "briefs" |
| 16 | old_db_override = store._db_override |
| 17 | old_briefs_dir = briefing.BRIEFS_DIR |
| 18 | try: |
| 19 | store._db_override = db_path |
| 20 | briefing.BRIEFS_DIR = briefs_dir |
| 21 | topic = store.add_topic("test topic") |
| 22 | store.record_run(topic["id"], source_mode="v3", status="completed") |
| 23 | result = briefing.generate_daily() |
| 24 | self.assertEqual(result["status"], "ok") |
| 25 | self.assertEqual(result["topics"][0]["name"], "test topic") |
| 26 | self.assertIsNotNone(result["topics"][0]["hours_ago"]) |
| 27 | self.assertGreaterEqual(result["topics"][0]["hours_ago"], 0.0) |
| 28 | finally: |
| 29 | store._db_override = old_db_override |
| 30 | briefing.BRIEFS_DIR = old_briefs_dir |
| 31 | |
| 32 | def test_generate_weekly_ranks_top_findings_by_engagement(self): |
| 33 | """Weekly digest top_findings must be the highest-engagement items, not |
| 34 | the most recent. get_new_findings returns first_seen DESC, so without an |
| 35 | explicit engagement sort the digest headlines recent low-engagement noise.""" |
| 36 | with tempfile.TemporaryDirectory() as tmpdir: |
| 37 | db_path = Path(tmpdir) / "research.db" |
| 38 | briefs_dir = Path(tmpdir) / "briefs" |
| 39 | old_db_override = store._db_override |
| 40 | old_briefs_dir = briefing.BRIEFS_DIR |
| 41 | try: |
| 42 | store._db_override = db_path |
| 43 | briefing.BRIEFS_DIR = briefs_dir |
| 44 | topic = store.add_topic("test topic") |
| 45 | run_id = store.record_run( |
| 46 | topic["id"], source_mode="v3", status="completed" |
| 47 | ) |
| 48 | |
| 49 | now = datetime.now(timezone.utc) |
| 50 | # Finding i: i days ago, engagement i. The most recent (i=0) has |
| 51 | # the lowest engagement, so a recency sort and an engagement sort |
| 52 | # disagree on the top 5. |
| 53 | conn = store._connect() |
| 54 | try: |
| 55 | for i in range(6): |
| 56 | first_seen = (now - timedelta(days=i, hours=1)).strftime( |
| 57 | "%Y-%m-%d %H:%M:%S" |
| 58 | ) |
| 59 | conn.execute( |
| 60 | """INSERT INTO findings |
| 61 | (run_id, topic_id, source, source_url, |
| 62 | source_title, engagement_score, first_seen) |
| 63 | VALUES (?, ?, 'reddit', ?, ?, ?, ?)""", |
| 64 | ( |
| 65 | run_id, |
| 66 | topic["id"], |
| 67 | f"https://example.com/{i}", |
| 68 | f"finding-{i}", |
| 69 | float(i), |
| 70 | first_seen, |
| 71 | ), |
| 72 | ) |
| 73 | conn.commit() |
| 74 | finally: |
| 75 | conn.close() |
| 76 | |
| 77 | result = briefing.generate_weekly() |
| 78 | top = result["topics"][0]["top_findings"] |
| 79 | scores = [f["engagement_score"] for f in top] |
| 80 | |
| 81 | self.assertEqual(len(top), 5) |
| 82 | self.assertEqual(scores, [5.0, 4.0, 3.0, 2.0, 1.0]) |
| 83 | # The most-recent, lowest-engagement finding is dropped, not kept. |
| 84 | self.assertNotIn(0.0, scores) |
| 85 | finally: |
| 86 | store._db_override = old_db_override |
| 87 | briefing.BRIEFS_DIR = old_briefs_dir |
| 88 | |
| 89 | def test_save_briefing_uses_utf8_encoding(self): |
| 90 | with tempfile.TemporaryDirectory() as tmpdir: |
| 91 | old_briefs_dir = briefing.BRIEFS_DIR |
| 92 | try: |
| 93 | briefing.BRIEFS_DIR = Path(tmpdir) / "briefs" |
| 94 | payload = {"status": "ok", "message": "emoji 💬 and accents café"} |
| 95 | with mock.patch("briefing.open", create=True) as mock_open: |
| 96 | handle = mock.Mock() |
| 97 | handle.__enter__ = mock.Mock(return_value=handle) |
| 98 | handle.__exit__ = mock.Mock(return_value=False) |
| 99 | mock_open.return_value = handle |
| 100 | |
| 101 | briefing._save_briefing(payload) |
| 102 | |
| 103 | mock_open.assert_called_once() |
| 104 | _, kwargs = mock_open.call_args |
| 105 | self.assertEqual("w", kwargs["mode"] if "mode" in kwargs else mock_open.call_args.args[1]) |
| 106 | self.assertEqual("utf-8", kwargs["encoding"]) |
| 107 | finally: |
| 108 | briefing.BRIEFS_DIR = old_briefs_dir |
| 109 | |
| 110 | if __name__ == "__main__": |
| 111 | unittest.main() |
| 112 |