194 lines
9.3 KiB
Python
194 lines
9.3 KiB
Python
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_preserves_official_edition_order_instead_of_sorting_ids(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
base = Path(temporary)
|
|
source = base / "source"
|
|
output = base / "normalized"
|
|
document = self.document(source, 6)
|
|
self.edition(document, 920492, {"ru": "<p>Первая редакция</p>"})
|
|
self.edition(document, 920292, {"ru": "<p>Последняя редакция</p>"})
|
|
for code, order in ((920492, 0), (920292, 1)):
|
|
metadata_path = document / "editions" / str(code) / "metadata.json"
|
|
metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
|
|
metadata["SourceOrder"] = order
|
|
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False), encoding="utf-8")
|
|
|
|
normalize_archive(source, output)
|
|
normalized = json.loads((output / "documents/6/document.json").read_text(encoding="utf-8"))
|
|
|
|
self.assertEqual([item["source_code"] for item in normalized["editions"]], ["920492", "920292"])
|
|
self.assertEqual([item["source_order"] for item in normalized["editions"]], [0, 1])
|
|
|
|
def test_rejects_partial_source_edition_order(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
base = Path(temporary)
|
|
source = base / "source"
|
|
output = base / "normalized"
|
|
document = self.document(source, 6)
|
|
self.edition(document, 10, {"ru": "<p>Один</p>"})
|
|
self.edition(document, 20, {"ru": "<p>Два</p>"})
|
|
metadata_path = document / "editions/10/metadata.json"
|
|
metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
|
|
metadata["SourceOrder"] = 0
|
|
metadata_path.write_text(json.dumps(metadata), encoding="utf-8")
|
|
|
|
with self.assertLogs("normalization.minjust_cbd", level="ERROR"):
|
|
result = normalize_archive(source, output)
|
|
|
|
self.assertEqual((result.failed, result.normalized), (1, 0))
|
|
with sqlite3.connect(output / "manifest.sqlite3") as connection:
|
|
error = connection.execute("SELECT error FROM documents WHERE code='6'").fetchone()[0]
|
|
self.assertIn("Invalid source edition order", error)
|
|
|
|
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()
|