from __future__ import annotations import gzip import hashlib import json import sqlite3 import tempfile import unittest from pathlib import Path from phase1.ik_ingest.apply_source_probe_proposals import apply, plan from phase1.ik_ingest.probe_source_candidates import REPORT_VERSION class ApplySourceProbeProposalsTest(unittest.TestCase): def setUp(self) -> None: self.temporary = tempfile.TemporaryDirectory() self.workspace = Path(self.temporary.name) (self.workspace / "state").mkdir() (self.workspace / "reports").mkdir() self.database = self.workspace / "state" / "crawl.sqlite3" with sqlite3.connect(self.database) as connection: connection.executescript( """ CREATE TABLE targets( target_doc_id TEXT PRIMARY KEY, source_id TEXT, match_score REAL, match_method TEXT, status TEXT, error TEXT, updated_at TEXT ); CREATE TABLE candidates(source_id TEXT PRIMARY KEY, source_url TEXT); CREATE TABLE fetches(source_id TEXT, status TEXT); CREATE TABLE events( event_type TEXT, payload_json TEXT, created_at TEXT ); """ ) connection.execute( "INSERT INTO targets(target_doc_id,status) VALUES(?,?)", ("2000 INSC 1", "pending"), ) connection.execute( "INSERT INTO candidates VALUES(?,?)", ("100", "https://indiankanoon.org/doc/100/"), ) probe_dir = ( self.workspace / "checkpoints" / "source_resolution" / "probes" / "2000_INSC_1" ) probe_dir.mkdir(parents=True) html = b"verified" with gzip.open(probe_dir / "100.html.gz", "wb") as handle: handle.write(html) probe = { "report_version": REPORT_VERSION, "target_doc_id": "2000 INSC 1", "source_id": "100", "html_sha256": hashlib.sha256(html).hexdigest(), "evaluation": { "safe_proposal": True, "verified_rule": "source_native_case_number", "features": {"score": 0.9}, }, } (probe_dir / "100.json").write_text(json.dumps(probe), encoding="utf-8") proposal = { "target_doc_id": "2000 INSC 1", "source_id": "100", "source_url": "https://indiankanoon.org/doc/100/", "verified_rule": "source_native_case_number", } (self.workspace / "reports" / "source_candidate_probe_proposals.jsonl").write_text( json.dumps(proposal) + "\n", encoding="utf-8" ) def tearDown(self) -> None: self.temporary.cleanup() def test_dry_run_then_execute_with_backup(self) -> None: report, rows = plan(self.workspace) self.assertFalse(report["database_mutated"]) self.assertEqual(report["valid_rows"], 1) with sqlite3.connect(self.database) as connection: self.assertIsNone( connection.execute( "SELECT source_id FROM targets WHERE target_doc_id='2000 INSC 1'" ).fetchone()[0] ) result = apply(self.workspace, report, rows) self.assertEqual(result["applied"], 1) self.assertTrue(Path(result["backup_path"]).exists()) with sqlite3.connect(self.database) as connection: row = connection.execute( "SELECT source_id,match_method FROM targets WHERE target_doc_id='2000 INSC 1'" ).fetchone() self.assertEqual(row[0], "100") self.assertEqual( row[1], "source_probe_v2:source_native_case_number" ) if __name__ == "__main__": unittest.main()