akyldash/backend/test_minjust_normalization.py

import json
import os
import sqlite3
import tempfile
import unittest
from pathlib import Path

from normalization.minjust_cbd import normalize_archive


class MinjustNormalizationTest(unittest.TestCase):
    def write_json(self, path: Path, value: object) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")

    def edition(self, root: Path, code: int, languages: dict[str, str]) -> None:
        directory = root / "editions" / str(code)
        self.write_json(
            directory / "metadata.json",
            {"Code": code, "Name": {"Rus": "Редакция", "Kyr": "Редакция"}, "Type": "edition", "Images": []},
        )
        for language, content in languages.items():
            (directory / f"{language}.html").write_text(content, encoding="utf-8")

    def document(self, root: Path, code: int = 1) -> Path:
        directory = root / "documents" / str(code)
        self.write_json(
            directory / "metadata.json",
            {
                "Code": code,
                "Class": {"Rus": "Акты", "Kyr": "Актылар"},
                "Type": {"Rus": "Закон", "Kyr": "Мыйзам"},
                "Title": {"Rus": " ", "Kyr": None},
                "Name": {"Rus": "Документ", "Kyr": "Документ"},
                "Status": {"Rus": "Действует", "Kyr": "Күчүндө"},
                "Number": "1",
                "DateAdopted": "2026-01-01",
                "IsPublicInCdb": True,
                "IsPublicInRegister": True,
                "Authorities": [],
                "SourcePublications": [],
                "Keywords": [],
                "GeneralClassifiers": [],
                "References": [],
            },
        )
        return directory

    def test_languages_safety_empty_document_and_deterministic_fragments(self):
        with tempfile.TemporaryDirectory() as temporary:
            base = Path(temporary)
            source = base / "source"
            output = base / "normalized"
            document = self.document(source)
            self.edition(
                document,
                10,
                {
                    "ru": '<meta charset=unicode><style>x</style><p style="color:red">Статья 1 Закон<script>bad()</script></p><a href="javascript:bad">ссылка</a><a href="http://[">сломанная ссылка</a><img src="http://[">',
                },
            )
            self.edition(document, 20, {"ky": "<p>1. Кыргызча жобо</p>"})
            self.edition(document, 30, {"ru": "<p>Русский</p>", "ky": "<p>Кыргызча</p>"})
            self.edition(document, 40, {})

            first = normalize_archive(source, output)
            fragments_path = output / "documents/1/editions/10/ru/fragments.json"
            fragments = fragments_path.read_bytes()
            second = normalize_archive(source, output)

            self.assertEqual((first.normalized, second.skipped), (1, 1))
            self.assertEqual(fragments, fragments_path.read_bytes())
            safe_html = (output / "documents/1/editions/10/ru/content.html").read_text(encoding="utf-8")
            self.assertNotIn("script", safe_html)
            self.assertNotIn("style=", safe_html)
            self.assertNotIn("javascript:", safe_html)
            self.assertNotIn("http://[", safe_html)
            self.assertIn("Статья 1 Закон", safe_html)
            parsed = json.loads(fragments)
            self.assertEqual(parsed[0]["type"], "article")
            self.assertEqual(parsed[0]["id"], "document:1:edition:10:lang:ru:fragment:1")
            self.assertEqual(len(parsed[0]["text_sha256"]), 64)
            canonical = json.loads((output / "documents/1/document.json").read_text(encoding="utf-8"))
            self.assertEqual(canonical["available_languages"], ["ru", "ky"])
            self.assertIsNone(canonical["title"]["ru"])
            empty = json.loads((output / "documents/1/editions/40/edition.json").read_text(encoding="utf-8"))
            self.assertFalse(empty["quality"]["has_html"])

    def test_continues_after_bad_document_and_clears_repaired_error(self):
        with tempfile.TemporaryDirectory() as temporary:
            base = Path(temporary)
            source = base / "source"
            output = base / "normalized"
            bad = source / "documents/1"
            bad.mkdir(parents=True)
            (bad / "metadata.json").write_text("not json", encoding="utf-8")
            good = self.document(source, 2)
            self.edition(good, 10, {"ky": "<p>Берене 1 Текст</p>"})

            with self.assertLogs("normalization.minjust_cbd", level="ERROR"):
                failed = normalize_archive(source, output)
            with sqlite3.connect(output / "manifest.sqlite3") as connection:
                state, failed_at = connection.execute(
                    "SELECT state, failed_at FROM documents WHERE code='1'"
                ).fetchone()
                self.assertEqual(state, "error")
                self.assertIsNotNone(failed_at)
            self.write_json(bad / "metadata.json", {"Code": 1, "Name": {"Rus": "Исправлен", "Kyr": None}})
            repaired = normalize_archive(source, output)

            self.assertEqual((failed.failed, failed.normalized), (1, 1))
            self.assertEqual((repaired.normalized, repaired.skipped, repaired.failed), (1, 1, 0))
            with sqlite3.connect(output / "manifest.sqlite3") as connection:
                self.assertEqual(
                    connection.execute(
                        "SELECT state, error, failed_at FROM documents WHERE code='1'"
                    ).fetchone(),
                    ("success", None, None),
                )

    def test_rejects_overlapping_input_and_output(self):
        with tempfile.TemporaryDirectory() as temporary:
            source = Path(temporary) / "source"
            metadata = self.document(source) / "metadata.json"
            original = metadata.read_bytes()

            for output in (source, source / "normalized", source.parent):
                with self.subTest(output=output):
                    with self.assertRaisesRegex(ValueError, "must not overlap"):
                        normalize_archive(source, output)
                    self.assertEqual(metadata.read_bytes(), original)

    def test_recovers_interrupted_directory_publication_before_skip(self):
        with tempfile.TemporaryDirectory() as temporary:
            base = Path(temporary)
            source = base / "source"
            output = base / "normalized"
            self.document(source)
            first = normalize_archive(source, output)
            target = output / "documents/1"
            backup = output / "documents/.1.previous"
            os.replace(target, backup)

            second = normalize_archive(source, output)

            self.assertEqual((first.normalized, second.skipped), (1, 1))
            self.assertTrue((target / "document.json").is_file())
            self.assertFalse(backup.exists())


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