import json import sqlite3 import sys import tempfile import unittest from pathlib import Path HERE = Path(__file__).resolve().parent REPO_ROOT = HERE.parent.parent sys.path.insert(0, str(REPO_ROOT)) from phase1.ik_ingest.audit_batch_boundaries import ( boundary_rows, build_report, ) class BatchBoundaryAuditTest(unittest.TestCase): def create_workspace(self, root: Path) -> Path: workspace = root / "workspace" state = workspace / "state" state.mkdir(parents=True) database = state / "crawl.sqlite3" with sqlite3.connect(database) as connection: connection.executescript( """ CREATE TABLE fetches ( source_id TEXT PRIMARY KEY, status TEXT NOT NULL, completed_at TEXT ); CREATE TABLE events ( id INTEGER PRIMARY KEY, event_type TEXT NOT NULL, payload_json TEXT NOT NULL, created_at TEXT NOT NULL ); """ ) connection.executemany( "INSERT INTO fetches VALUES(?,?,?)", [ ("a", "complete", "2026-07-30T00:10:00+00:00"), ("b", "complete", "2026-07-30T00:20:00+00:00"), ("c", "failed", "2026-07-30T00:20:40+00:00"), ], ) connection.executemany( "INSERT INTO events VALUES(?,?,?,?)", [ ( 10, "document_fetch_completed", json.dumps( { "jobs_selected": 1, "completed": 1, "failed": 0, "ready": 1, "quarantined": 0, } ), "2026-07-30T00:10:30+00:00", ), ( 11, "document_fetch_completed", json.dumps( { "jobs_selected": 2, "completed": 1, "failed": 1, "ready": 1, "quarantined": 0, } ), "2026-07-30T00:20:45+00:00", ), ], ) return workspace def test_uses_last_successful_fetch_in_each_event_window(self): with tempfile.TemporaryDirectory() as folder: workspace = self.create_workspace(Path(folder)) rows = boundary_rows(workspace / "state" / "crawl.sqlite3") self.assertEqual([row["batch_number"] for row in rows], [1, 2]) self.assertEqual( [row["post_fetch_seconds"] for row in rows], [30.0, 45.0], ) self.assertEqual(rows[1]["failed"], 1) def test_report_is_read_only_and_summarizes_observed_batches(self): with tempfile.TemporaryDirectory() as folder: workspace = self.create_workspace(Path(folder)) report = build_report(workspace) self.assertFalse(report["corpus_state_mutated"]) self.assertEqual(report["completed_batches"], 2) self.assertEqual(report["latest_post_fetch_seconds"], 45.0) self.assertEqual(report["maximum_post_fetch_seconds"], 45.0) if __name__ == "__main__": unittest.main()