| 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"<html>verified</html>" |
| 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() |
|
|
