tests/test_common.py

"""Unit tests for the shared ingestion helpers (canlex/_common.py).

Offline only. The corpus-write guard exists because every ingester rebuilds its
whole processed file from a scrape: on 2026-07-27 a rebuilt upstream index
scraped to zero items and the write emptied 1,870 good chunks.
"""
import contextlib
import io
import json
import tempfile
import unittest
from pathlib import Path

from canlex import _common


class SafeToWriteTests(unittest.TestCase):
    def test_refuses_to_replace_a_corpus_with_nothing(self):
        self.assertFalse(_common.safe_to_write(0, 1870))

    def test_refuses_a_large_collapse(self):
        self.assertFalse(_common.safe_to_write(500, 1870))

    def test_allows_normal_churn(self):
        self.assertTrue(_common.safe_to_write(1866, 1870))
        self.assertTrue(_common.safe_to_write(1900, 1870))

    def test_first_ever_run_is_allowed(self):
        self.assertTrue(_common.safe_to_write(0, 0))

    def test_ratio_is_tunable(self):
        self.assertTrue(_common.safe_to_write(500, 1000, ratio=0.5))
        self.assertFalse(_common.safe_to_write(499, 1000, ratio=0.5))


class WriteCorpusTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.path = Path(self.tmp.name) / "corpus.json"
        self.addCleanup(self.tmp.cleanup)

    def _write(self, chunks):
        self.path.write_text(json.dumps(chunks), encoding="utf-8")

    def test_writes_when_there_is_nothing_to_lose(self):
        self.assertTrue(_common.write_corpus(self.path, [{"id": "a"}]))
        self.assertEqual(json.loads(self.path.read_text(encoding="utf-8")),
                         [{"id": "a"}])

    def test_refuses_and_leaves_the_file_untouched(self):
        self._write([{"id": str(i)} for i in range(100)])
        with contextlib.redirect_stdout(io.StringIO()) as out:
            self.assertFalse(_common.write_corpus(self.path, []))
        self.assertIn("REFUSING", out.getvalue())
        self.assertEqual(len(json.loads(self.path.read_text(encoding="utf-8"))),
                         100)

    def test_allow_shrink_overrides(self):
        self._write([{"id": str(i)} for i in range(100)])
        self.assertTrue(_common.write_corpus(self.path, [], allow_shrink=True))
        self.assertEqual(json.loads(self.path.read_text(encoding="utf-8")), [])

    def test_unreadable_stored_file_does_not_block_a_write(self):
        # A truncated/corrupt corpus reads as empty, so recovery is possible.
        self.path.write_text("{not json", encoding="utf-8")
        self.assertTrue(_common.write_corpus(self.path, [{"id": "a"}]))

    def test_creates_missing_parent_directories(self):
        nested = Path(self.tmp.name) / "deep" / "corpus.json"
        self.assertTrue(_common.write_corpus(nested, [{"id": "a"}]))
        self.assertTrue(nested.exists())

    def test_non_ascii_survives_the_round_trip(self):
        chunks = [{"id": "a", "text": "détention à la frontière"}]
        _common.write_corpus(self.path, chunks)
        self.assertEqual(json.loads(self.path.read_text(encoding="utf-8")),
                         chunks)


class StoredChunksTests(unittest.TestCase):
    def test_missing_file_is_empty(self):
        self.assertEqual(_common.stored_chunks(Path("nope-does-not-exist.json")),
                         [])


class PreserveDroppedTests(unittest.TestCase):
    STORED = [{"section": "A", "text": "one"},
              {"section": "A", "text": "two"},
              {"section": "B", "text": "three"}]

    def test_reattaches_a_dropped_item(self):
        chunks, dropped = _common.preserve_dropped([], self.STORED, only={"A"})
        self.assertEqual(dropped, ["A"])
        self.assertEqual(len(chunks), 2)

    def test_only_restricts_to_genuine_failures(self):
        # B vanished deliberately (cancelled upstream); it must not come back.
        chunks, dropped = _common.preserve_dropped([], self.STORED, only={"A"})
        self.assertNotIn("B", dropped)
        self.assertNotIn("three", [c["text"] for c in chunks])

    def test_without_only_every_missing_item_is_restored(self):
        _chunks, dropped = _common.preserve_dropped([], self.STORED)
        self.assertEqual(dropped, ["A", "B"])

    def test_freshly_scraped_items_are_not_overwritten(self):
        fresh = [{"section": "A", "text": "new"}]
        chunks, dropped = _common.preserve_dropped(fresh, self.STORED,
                                                   only={"A"})
        self.assertEqual((chunks, dropped), (fresh, []))

    def test_a_stored_chunk_without_the_identity_field_is_skipped(self):
        # Not a crash: an older-schema corpus must not take down the rebuild
        # the guard exists to protect.
        stored = self.STORED + [{"id": "legacy-1", "text": "no section key"}]
        chunks, dropped = _common.preserve_dropped([], stored, only={"A"})
        self.assertEqual(dropped, ["A"])
        self.assertNotIn("legacy-1", [c.get("id") for c in chunks])

    def test_custom_key(self):
        stored = [{"case": "Khosa", "text": "x"}]
        chunks, dropped = _common.preserve_dropped(
            [], stored, key=lambda c: c["case"], only={"Khosa"})
        self.assertEqual((len(chunks), dropped), (1, ["Khosa"]))


if __name__ == "__main__":
    unittest.main()