phase1/eval/test_batch_boundary_audit.py

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()