tests/test_pipeline_ledger.py
Rama main · réplica de lectura del repositorio original.
"""管線帳本要記得住那八天。
2026-09-01 到 09-09，書籤轉卡每晚跑、每晚以 404 失敗、每晚回報成功，八天 0
產出。當時所有「現況」的數字都正常：卡片 1,603 張、索引 1,632 筆。那八個晚上
在系統裡不存在，因為沒有任何地方記著「昨晚收了 9 筆、產出 0 筆、失敗 9 筆、
原因是找不到模型」。
所以這裡最重要的一個測試是 test_eight_quiet_nights_are_a_fault：把那段歷史
餵進帳本，健檢必須說出來。其餘的測試是為了讓它不要用誤報換到這個能力。
"""
from __future__ import annotations
import json
import os
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest import mock
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
import xkb_ledger # noqa: E402
import health_check_pipeline as hc # noqa: E402
DAY = 86400.0
class _LedgerSandbox(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.path = Path(self._tmp.name) / "pipeline-ledger.jsonl"
patch = mock.patch.object(xkb_ledger, "LEDGER_PATH", self.path)
patch.start()
self.addCleanup(patch.stop)
self.addCleanup(self._tmp.cleanup)
def write(self, rows: list[dict]) -> None:
self.path.write_text(
"\n".join(json.dumps(r, ensure_ascii=False) for r in rows) + "\n",
encoding="utf-8",
)
@staticmethod
def nights(stage: str, count: int, *, produced: int, failed: int,
pending_from: int, reason: str = "", newest_age_hours: float = 2) -> list[dict]:
"""一晚一筆，最新的那筆在 newest_age_hours 小時前。"""
rows = []
now = datetime.now(timezone.utc)
for i in range(count):
age = timedelta(hours=newest_age_hours) + timedelta(days=count - 1 - i)
row = {
"ts": (now - age).isoformat(),
"stage": stage,
"ok": True,
"intake": produced + failed,
"produced": produced,
"failed": failed,
"pending": pending_from + i * failed,
}
if reason and failed:
row["reason"] = reason
rows.append(row)
return rows
@staticmethod
def _msgs(section: dict) -> str:
return "\n".join(c["msg"] for c in section["checks"])
class Recording(_LedgerSandbox):
def test_a_run_is_written_as_one_line(self):
xkb_ledger.record("bookmark-batch", intake=9, produced=0, failed=9,
pending=57, reason="404 model not found")
rows = xkb_ledger.read()
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["stage"], "bookmark-batch")
self.assertEqual(rows[0]["produced"], 0)
self.assertEqual(rows[0]["failed"], 9)
self.assertIn("404", rows[0]["reason"])
def test_reading_filters_by_stage(self):
xkb_ledger.record("bookmark-batch", produced=1)
xkb_ledger.record("ingestion-batch", produced=4)
self.assertEqual(len(xkb_ledger.read("bookmark-batch")), 1)
self.assertEqual(xkb_ledger.read("ingestion-batch")[0]["produced"], 4)
def test_a_corrupt_line_does_not_lose_the_rest(self):
xkb_ledger.record("bookmark-batch", produced=1)
with self.path.open("a", encoding="utf-8") as fh:
fh.write("{ not json\n")
xkb_ledger.record("bookmark-batch", produced=2)
self.assertEqual([r["produced"] for r in xkb_ledger.read()], [1, 2])
def test_an_unwritable_ledger_does_not_raise(self):
"""帳本不能把管線弄掛——但也不會沉默，xkb_failures 會在 stderr 說。"""
blocked = Path(self._tmp.name) / "nope" / "x.jsonl"
with mock.patch.object(xkb_ledger, "LEDGER_PATH", blocked), \
mock.patch.object(Path, "mkdir", side_effect=OSError("read-only")):
entry = xkb_ledger.record("bookmark-batch", produced=1)
self.assertEqual(entry["stage"], "bookmark-batch")
class TypicalGap(_LedgerSandbox):
def test_needs_at_least_three_runs(self):
rows = self.nights("s", 2, produced=1, failed=0, pending_from=0)
self.assertIsNone(xkb_ledger.typical_gap_seconds(rows))
def test_one_catch_up_run_does_not_move_the_median(self):
"""手動補跑會製造一個很短的間隔。門檻不該因為它就變鬆或變緊。"""
rows = self.nights("s", 6, produced=1, failed=0, pending_from=0)
rows.append({**rows[-1], "ts": rows[-1]["ts"]}) # 同一刻再跑一次
gap = xkb_ledger.typical_gap_seconds(rows)
self.assertAlmostEqual(gap, DAY, delta=DAY * 0.1)
def test_a_burst_of_manual_runs_does_not_move_the_median(self):
"""中位數只擋得住「一次」離群。
2026-09-10 我為了驗證連續手動跑了六次書籤批次，每次相隔幾分鐘，中位
間隔被壓成 0 小時——健檢於是說「平常每 0h 跑一次，已經 11h 沒有紀錄」，
把一個正常的階段報成排程死掉。上一版的註解寫著「一次補跑不會把它拉
歪」，而上一版的測試也只測了一次。
"""
from datetime import datetime, timedelta, timezone
now = datetime.now(timezone.utc)
rows = [{"ts": (now - timedelta(days=d)).isoformat(), "stage": "s"}
for d in range(8, 0, -1)]
rows += [{"ts": (now - timedelta(minutes=m)).isoformat(), "stage": "s"}
for m in (40, 32, 25, 18, 9, 2)]
gap = xkb_ledger.typical_gap_seconds(rows)
self.assertIsNotNone(gap)
self.assertAlmostEqual(gap, DAY, delta=DAY * 0.2)
def test_a_history_with_no_rhythm_yet_says_so(self):
"""全部都在同一段操作裡——沒有節奏可比，就不要拿假節奏去判斷。"""
from datetime import datetime, timedelta, timezone
now = datetime.now(timezone.utc)
rows = [{"ts": (now - timedelta(minutes=m)).isoformat(), "stage": "s"}
for m in (30, 20, 10, 4)]
self.assertIsNone(xkb_ledger.typical_gap_seconds(rows))
class HealthCheckReadsIt(_LedgerSandbox):
def test_eight_quiet_nights_are_a_fault(self):
"""這一項存在的理由。"""
self.write(self.nights("bookmark-batch", 8, produced=0, failed=9,
pending_from=13, reason="404 model not found"))
section = hc.check_pipeline_ledger()
self.assertFalse(all(c["ok"] for c in section["checks"]))
msgs = self._msgs(section)
self.assertIn("沒有產出", msgs)
self.assertIn("404", msgs)
def test_a_working_stage_is_not_a_fault(self):
self.write(self.nights("bookmark-batch", 8, produced=20, failed=0,
pending_from=40))
section = hc.check_pipeline_ledger()
self.assertTrue(all(c["ok"] for c in section["checks"]), self._msgs(section))
def test_a_quiet_stage_with_an_empty_queue_is_not_a_fault(self):
"""沒有新書籤時，產出 0 是正常的，不是故障。"""
self.write(self.nights("bookmark-batch", 8, produced=0, failed=0,
pending_from=0))
section = hc.check_pipeline_ledger()
self.assertTrue(all(c["ok"] for c in section["checks"]), self._msgs(section))
def test_a_stage_that_stopped_reporting_is_a_fault(self):
"""排程整個死掉時，帳本會安靜——安靜要看得見。"""
rows = self.nights("bookmark-batch", 8, produced=20, failed=0,
pending_from=40, newest_age_hours=24 * 5)
self.write(rows)
section = hc.check_pipeline_ledger()
self.assertFalse(all(c["ok"] for c in section["checks"]))
self.assertIn("沒有紀錄", self._msgs(section))
def test_the_threshold_follows_the_stage_own_rhythm(self):
"""每 12 小時跑一次的階段，停 20 小時不算停。
寫死 24h 的話，改排程就會變成每天誤報，而每天誤報等於沒有檢查。
"""
now = datetime.now(timezone.utc)
rows = [{
"ts": (now - timedelta(hours=20 + 12 * i)).isoformat(),
"stage": "twice-daily", "ok": True, "produced": 3, "pending": 0,
} for i in range(8)][::-1]
self.write(rows)
section = hc.check_pipeline_ledger()
self.assertTrue(all(c["ok"] for c in section["checks"]), self._msgs(section))
def test_a_batch_that_failed_outright_is_named(self):
rows = self.nights("ingestion-batch", 5, produced=4, failed=0, pending_from=0)
rows[-1] = {**rows[-1], "ok": False, "reason": "向量索引建置離開碼 137"}
self.write(rows)
section = hc.check_pipeline_ledger()
self.assertFalse(all(c["ok"] for c in section["checks"]))
self.assertIn("137", self._msgs(section))
def test_an_empty_ledger_is_not_a_fault(self):
"""剛裝上去、還沒跑過任何排程，不是故障。"""
section = hc.check_pipeline_ledger()
self.assertTrue(all(c["ok"] for c in section["checks"]), self._msgs(section))
class BatchSummary(unittest.TestCase):
"""帳本接上去的第一次實跑就照出這個舊 bug。
worker 的輸出裡有兩行含 done=：佇列同步的累計總數，和這一次的結果。
原本的 sed 抓「最後一個 done=」，平常剛好對；佇列沒事做時 worker 不印
結果那行，於是這次的產出被讀成佇列總數——實測報過「產了 1575 張卡」，
還因此多跑了一次向量索引。
"""
QUEUE_LINE = "✅ queue synced: /x/tiege-queue.json (1591 items, todo=0, done=1575)"
def setUp(self):
sys.path.insert(0, str(ROOT / "scripts"))
import xkb_batch_summary
self.parse = xkb_batch_summary.parse
def test_the_queue_total_is_not_this_run_output(self):
result = self.parse(self.QUEUE_LINE + "\n✅ No todo items found\n")
self.assertEqual(result, (0, 0, "idle"))
def test_a_real_run_is_read_from_the_worker_summary(self):
text = (self.QUEUE_LINE
+ "\n📊 done=7 skipped=0 failed=2 remaining todo=0\n")
self.assertEqual(self.parse(text), (7, 2, "ran"))
def test_an_unreadable_run_is_not_reported_as_zero(self):
"""讀不出結果和「產出 0」是兩件事，混在一起就是這條管線八天沒被發現的原因。"""
self.assertIsNone(self.parse(self.QUEUE_LINE + "\nTraceback (most recent call last):\n"))
self.assertIsNone(self.parse(""))
def test_the_batch_script_treats_unreadable_as_a_fault(self):
text = (ROOT / "scripts" / "run_bookmark_batch.sh").read_text(encoding="utf-8")
self.assertIn("xkb_batch_summary.py", text)
self.assertIn("讀不出 worker 的執行結果", text)
# 舊的抓法不能留著，否則哪天有人改回去也不會有人發現。
self.assertNotIn('sed -n \'s/.*done=', text)
class Wiring(unittest.TestCase):
def test_the_batch_stages_write_to_it(self):
"""階段不寫，帳本就永遠是空的——而空帳本不會報錯。"""
for name, stage in [("run_bookmark_batch.sh", "bookmark-batch"),
("run_ingestion_batch.sh", "ingestion-batch")]:
text = (ROOT / "scripts" / name).read_text(encoding="utf-8")
with self.subTest(script=name):
self.assertIn("xkb_ledger.py", text)
self.assertIn(stage, text)
if __name__ == "__main__":
unittest.main()