Epictetus
88 lines
3.2 KiB
Python
88 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Guardrail R6 test: first_seen is write-once; re-harvest must NOT overwrite it.
|
|
|
|
Run: python3 -m unittest -v tests.test_store
|
|
Or: python3 tests/test_store.py
|
|
"""
|
|
import os
|
|
import sys
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from adapters._store import upsert_entries, true_first_seen
|
|
|
|
|
|
class TestFirstSeenWriteOnce(unittest.TestCase):
|
|
def setUp(self):
|
|
self.fd, self.db_path = tempfile.mkstemp(suffix=".db")
|
|
os.close(self.fd)
|
|
self.conn = sqlite3.connect(self.db_path)
|
|
self.conn.execute("""
|
|
CREATE TABLE entries (
|
|
id INTEGER PRIMARY KEY,
|
|
source TEXT,
|
|
source_id TEXT,
|
|
url TEXT UNIQUE,
|
|
title TEXT,
|
|
extracted_text TEXT,
|
|
summary TEXT,
|
|
category_tags TEXT,
|
|
signal_score REAL,
|
|
raw_metadata TEXT,
|
|
first_seen TEXT,
|
|
last_updated TEXT
|
|
)
|
|
""")
|
|
self.conn.commit()
|
|
|
|
def tearDown(self):
|
|
self.conn.close()
|
|
os.unlink(self.db_path)
|
|
|
|
def _entry(self, url, true_date, harvest):
|
|
return {
|
|
"source": "hackernews",
|
|
"source_id": url,
|
|
"url": url,
|
|
"title": "t",
|
|
"extracted_text": "",
|
|
"summary": None,
|
|
"category_tags": "[]",
|
|
"signal_score": 1.0,
|
|
"raw_metadata": '{"time": 1783762613}', # Ghost Font = 2026-07-11
|
|
"first_seen": true_date, # true publish date
|
|
"last_updated": harvest,
|
|
}
|
|
|
|
def test_reharvest_preserves_first_seen(self):
|
|
url = "https://example.com/story"
|
|
# First harvest: true date 2026-07-11, harvest time 2026-07-13
|
|
upsert_entries(self.conn, [self._entry(url, "2026-07-11T09:36:53Z", "2026-07-13T04:15:00Z")])
|
|
row = self.conn.execute("SELECT first_seen, last_updated FROM entries WHERE url=?", (url,)).fetchone()
|
|
self.assertEqual(row[0], "2026-07-11T09:36:53Z")
|
|
self.assertEqual(row[1], "2026-07-13T04:15:00Z")
|
|
|
|
# Re-harvest NEXT day: harvest time 2026-07-14. first_seen MUST stay 07-11.
|
|
upsert_entries(self.conn, [self._entry(url, "2026-07-14T00:00:00Z", "2026-07-14T00:00:00Z")])
|
|
row2 = self.conn.execute("SELECT first_seen, last_updated FROM entries WHERE url=?", (url,)).fetchone()
|
|
# This is the guardrail assertion:
|
|
self.assertEqual(row2[0], "2026-07-11T09:36:53Z",
|
|
"GUARDRAIL R6 VIOLATED: first_seen was overwritten on re-harvest")
|
|
# last_updated SHOULD advance
|
|
self.assertEqual(row2[1], "2026-07-14T00:00:00Z")
|
|
|
|
def test_true_first_seen_derives_source_date(self):
|
|
# HN epoch -> ISO
|
|
self.assertEqual(true_first_seen({"time": 1783762613}, "hackernews", "2026-07-13T00:00:00Z"),
|
|
"2026-07-11T09:36:53Z")
|
|
# Reddit ISO passthrough
|
|
self.assertEqual(true_first_seen({"published": "2026-07-11T20:28:05+00:00"}, "reddit", "2026-07-13T00:00:00Z"),
|
|
"2026-07-11T20:28:05Z")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|