375 lines
18 KiB
Python
375 lines
18 KiB
Python
import hashlib
|
||
import io
|
||
import json
|
||
import sqlite3
|
||
import tempfile
|
||
import unittest
|
||
import urllib.error
|
||
from contextlib import closing
|
||
from pathlib import Path
|
||
from unittest.mock import patch
|
||
|
||
from search.catalog import legal_force_code, labels, source_code
|
||
from search.minjust_opensearch import bulk_batches, document_codes, export_bulk, load_bulk, request_json, switch_alias
|
||
|
||
|
||
class MinjustOpenSearchTest(unittest.TestCase):
|
||
def test_unknown_document_type_keeps_source_label_and_has_no_inferred_force(self):
|
||
value = {"ru": "Решение", "ky": "Решение"}
|
||
|
||
self.assertEqual(source_code("document_type", value), "other")
|
||
self.assertEqual(labels("document_type", "other"), {"ru": "Прочие документы", "ky": "Башка документтер"})
|
||
self.assertIsNone(legal_force_code(value))
|
||
self.assertEqual(source_code("status", {"ru": "Не вступил в силу", "ky": "Күчүнө кире элек"}), "not_yet_effective")
|
||
self.assertEqual(legal_force_code({"ru": "Конституционный закон", "ky": "Конституционный закон"}), "constitutional")
|
||
self.assertEqual(legal_force_code({"ru": "Указ", "ky": "Жарлыгы"}), "subordinate")
|
||
self.assertEqual(source_code("document_type", {"ru": "устав"}), "charter")
|
||
|
||
def test_switches_alias_atomically_after_successful_load(self):
|
||
with patch("search.minjust_opensearch.request_json", return_value={"acknowledged": True}) as request:
|
||
switch_alias("http://127.0.0.1:9200/", "akyldash-fragments-v2", "akyldash-fragments-current")
|
||
|
||
self.assertEqual(request.call_args.args[:2], ("http://127.0.0.1:9200/_aliases", "POST"))
|
||
body = json.loads(request.call_args.args[2])
|
||
self.assertEqual(body["actions"][0], {"remove": {"index": "*", "alias": "akyldash-fragments-current", "must_exist": False}})
|
||
self.assertEqual(body["actions"][1], {"add": {"index": "akyldash-fragments-v2", "alias": "akyldash-fragments-current"}})
|
||
|
||
with self.assertRaisesRegex(ValueError, "differ"):
|
||
switch_alias("http://127.0.0.1:9200", "same", "same")
|
||
with self.assertRaisesRegex(ValueError, "differ"):
|
||
switch_alias("http://127.0.0.1:9200", "index", "")
|
||
with patch("search.minjust_opensearch.request_json", return_value={"acknowledged": False}):
|
||
with self.assertRaisesRegex(RuntimeError, "did not acknowledge"):
|
||
switch_alias("http://127.0.0.1:9200", "index", "alias")
|
||
|
||
def test_exports_atomic_bulk_and_rejects_mismatched_fragment(self):
|
||
with tempfile.TemporaryDirectory() as temporary:
|
||
root = Path(temporary)
|
||
document_root = root / "normalized/documents/7"
|
||
document_root.mkdir(parents=True)
|
||
with closing(sqlite3.connect(root / "normalized/manifest.sqlite3")) as connection:
|
||
with connection:
|
||
connection.execute("CREATE TABLE documents (code TEXT, state TEXT)")
|
||
connection.execute("INSERT INTO documents VALUES ('7', 'success')")
|
||
connection.execute("INSERT INTO documents VALUES ('8', 'success')")
|
||
(document_root / "document.json").write_text(
|
||
json.dumps(
|
||
{
|
||
"schema_version": "1",
|
||
"source_code": "7",
|
||
"name": {"ru": "Закон", "ky": "Мыйзам"},
|
||
"type": {"ru": "Закон", "ky": "Мыйзам"},
|
||
"status": {"ru": "Действует", "ky": "Күчүндө"},
|
||
"number": "1",
|
||
"dates": {"DateAdopted": "2026-01-01"},
|
||
"authority_paths": [{"ru": ["Кабинет"], "ky": ["Кабинет"]}],
|
||
},
|
||
ensure_ascii=False,
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
empty = root / "normalized/documents/8"
|
||
empty.mkdir()
|
||
(empty / "document.json").write_text(
|
||
json.dumps({"schema_version": "1", "source_code": "8"}), encoding="utf-8"
|
||
)
|
||
for language, text in (("ru", "Текст \"RU\"\nстрока"), ("ky", "Кыргызча текст")):
|
||
path = document_root / f"editions/10/{language}/fragments.json"
|
||
path.parent.mkdir(parents=True)
|
||
path.write_text(
|
||
json.dumps(
|
||
[{
|
||
"id": f"document:7:edition:10:lang:{language}:fragment:1",
|
||
"document_code": "7",
|
||
"edition_code": "10",
|
||
"language": language,
|
||
"position": 1,
|
||
"type": "paragraph",
|
||
"text": text,
|
||
"text_sha256": hashlib.sha256(text.encode()).hexdigest(),
|
||
"source_path": f"documents/7/editions/10/{language}.html",
|
||
"source_sha256": "b" * 64,
|
||
}],
|
||
ensure_ascii=False,
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
output = root / "bulk.ndjson"
|
||
self.assertEqual(export_bulk(root / "normalized", output, "test-index"), (2, 2))
|
||
content = output.read_bytes()
|
||
self.assertTrue(content.endswith(b"\n"))
|
||
lines = [json.loads(line) for line in content.splitlines()]
|
||
self.assertEqual(len(lines), 4)
|
||
self.assertEqual(lines[0]["index"]["_id"], "document:7:edition:10:lang:ky:fragment:1")
|
||
self.assertIn("text_ky", lines[1])
|
||
self.assertTrue(lines[1]["is_current_edition"])
|
||
self.assertNotIn("text_ru", lines[1])
|
||
self.assertEqual(lines[1]["legal_force_code"], "legislative")
|
||
self.assertEqual(lines[3]["text_ru"], "Текст \"RU\"\nстрока")
|
||
self.assertEqual(
|
||
list(bulk_batches(iter((("7", b"a\nb\n"), ("8", b"c\nd\n"))), 4)),
|
||
[("7", b"a\nb\n"), ("8", b"c\nd\n")],
|
||
)
|
||
self.assertEqual(document_codes(root / "normalized", 2, "8"), ["8"])
|
||
with self.assertRaisesRegex(ValueError, "selected range"):
|
||
document_codes(root / "normalized", 1, "8")
|
||
|
||
checkpoint = root / "checkpoint.json"
|
||
with patch("search.minjust_opensearch.request_json") as request:
|
||
request.side_effect = [
|
||
{},
|
||
{"cluster_uuid": "cluster-1"},
|
||
{"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
|
||
{"errors": False, "items": [{"index": {}}, {"index": {}}]},
|
||
{"acknowledged": True},
|
||
]
|
||
self.assertEqual(
|
||
load_bulk(
|
||
root / "normalized",
|
||
"http://127.0.0.1:9200",
|
||
"test-index",
|
||
maximum_bytes=4096,
|
||
checkpoint=checkpoint,
|
||
alias="test-current",
|
||
),
|
||
(2, 2),
|
||
)
|
||
self.assertEqual(request.call_args_list[-2].args[3], "application/x-ndjson")
|
||
self.assertEqual(request.call_args_list[-1].args[:2], ("http://127.0.0.1:9200/_aliases", "POST"))
|
||
state = json.loads(checkpoint.read_text(encoding="utf-8"))
|
||
self.assertEqual(state["last_document_code"], "7")
|
||
self.assertTrue(state["complete"])
|
||
|
||
legacy = state.copy()
|
||
legacy.pop("alias")
|
||
legacy["schema_version"] = 1
|
||
legacy["last_document_code"] = "8"
|
||
legacy["complete"] = False
|
||
checkpoint.write_text(json.dumps(legacy), encoding="utf-8")
|
||
with patch("search.minjust_opensearch.request_json") as request:
|
||
request.side_effect = [
|
||
{"cluster_uuid": "cluster-1"},
|
||
{"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
|
||
]
|
||
self.assertEqual(
|
||
load_bulk(
|
||
root / "normalized",
|
||
"http://127.0.0.1:9200",
|
||
"test-index",
|
||
maximum_bytes=4096,
|
||
resume=True,
|
||
checkpoint=checkpoint,
|
||
),
|
||
(1, 0),
|
||
)
|
||
self.assertEqual(json.loads(checkpoint.read_text(encoding="utf-8"))["schema_version"], 2)
|
||
|
||
state["last_document_code"] = "8"
|
||
state["complete"] = False
|
||
checkpoint.write_text(json.dumps(state), encoding="utf-8")
|
||
with patch("search.minjust_opensearch.request_json") as request:
|
||
request.side_effect = [
|
||
{"cluster_uuid": "cluster-2"},
|
||
{"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
|
||
]
|
||
with self.assertRaisesRegex(ValueError, "does not match"):
|
||
load_bulk(
|
||
root / "normalized",
|
||
"http://127.0.0.1:9200",
|
||
"test-index",
|
||
maximum_bytes=4096,
|
||
resume=True,
|
||
checkpoint=checkpoint,
|
||
alias="test-current",
|
||
)
|
||
with patch("search.minjust_opensearch.request_json") as request:
|
||
request.side_effect = [
|
||
{"cluster_uuid": "cluster-1"},
|
||
{"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
|
||
]
|
||
with self.assertRaisesRegex(ValueError, "alias"):
|
||
load_bulk(
|
||
root / "normalized",
|
||
"http://127.0.0.1:9200",
|
||
"test-index",
|
||
maximum_bytes=4096,
|
||
resume=True,
|
||
checkpoint=checkpoint,
|
||
)
|
||
with patch("search.minjust_opensearch.request_json") as request:
|
||
request.side_effect = [
|
||
{"cluster_uuid": "cluster-1"},
|
||
{"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
|
||
]
|
||
with self.assertRaisesRegex(ValueError, "limit"):
|
||
load_bulk(
|
||
root / "normalized",
|
||
"http://127.0.0.1:9200",
|
||
"test-index",
|
||
maximum_bytes=4096,
|
||
limit=1,
|
||
resume=True,
|
||
checkpoint=checkpoint,
|
||
alias="test-current",
|
||
)
|
||
with patch("search.minjust_opensearch.request_json") as request:
|
||
request.side_effect = [
|
||
{"cluster_uuid": "cluster-1"},
|
||
{"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
|
||
{"acknowledged": True},
|
||
]
|
||
self.assertEqual(
|
||
load_bulk(
|
||
root / "normalized",
|
||
"http://127.0.0.1:9200",
|
||
"test-index",
|
||
maximum_bytes=4096,
|
||
resume=True,
|
||
checkpoint=checkpoint,
|
||
alias="test-current",
|
||
),
|
||
(1, 0),
|
||
)
|
||
self.assertTrue(all(call.args[1] == "GET" for call in request.call_args_list[:-1]))
|
||
self.assertEqual(request.call_args_list[-1].args[1], "POST")
|
||
|
||
state["last_document_code"] = "9"
|
||
state["complete"] = False
|
||
checkpoint.write_text(json.dumps(state), encoding="utf-8")
|
||
with patch("search.minjust_opensearch.request_json") as request:
|
||
request.side_effect = [
|
||
{"cluster_uuid": "cluster-1"},
|
||
{"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
|
||
]
|
||
with self.assertRaisesRegex(ValueError, "Resume document not found"):
|
||
load_bulk(
|
||
root / "normalized",
|
||
"http://127.0.0.1:9200",
|
||
"test-index",
|
||
maximum_bytes=4096,
|
||
resume=True,
|
||
checkpoint=checkpoint,
|
||
alias="test-current",
|
||
)
|
||
|
||
failed_checkpoint = root / "failed-checkpoint.json"
|
||
failure = {
|
||
"errors": True,
|
||
"items": [{"index": {"_id": "bad-id", "error": {"type": "mapper", "reason": "bad value"}}}],
|
||
}
|
||
failed_requests = [
|
||
{},
|
||
{"cluster_uuid": "cluster-1"},
|
||
{"failed-index": {"settings": {"index": {"uuid": "index-2"}}}},
|
||
failure,
|
||
]
|
||
with patch("search.minjust_opensearch.request_json", side_effect=failed_requests):
|
||
with self.assertRaisesRegex(RuntimeError, "bad-id: mapper: bad value"):
|
||
load_bulk(
|
||
root / "normalized",
|
||
"http://127.0.0.1:9200",
|
||
"failed-index",
|
||
maximum_bytes=4096,
|
||
checkpoint=failed_checkpoint,
|
||
)
|
||
failed_state = json.loads(failed_checkpoint.read_text(encoding="utf-8"))
|
||
self.assertIsNone(failed_state["last_document_code"])
|
||
self.assertFalse(failed_state["complete"])
|
||
|
||
state["last_document_code"] = "8"
|
||
state["complete"] = False
|
||
checkpoint.write_text(json.dumps(state), encoding="utf-8")
|
||
with closing(sqlite3.connect(root / "normalized/manifest.sqlite3")) as connection:
|
||
with connection:
|
||
connection.execute("INSERT INTO documents VALUES ('9', 'success')")
|
||
with patch("search.minjust_opensearch.request_json") as request:
|
||
request.side_effect = [
|
||
{"cluster_uuid": "cluster-1"},
|
||
{"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
|
||
]
|
||
with self.assertRaisesRegex(ValueError, "manifest changed"):
|
||
load_bulk(
|
||
root / "normalized",
|
||
"http://127.0.0.1:9200",
|
||
"test-index",
|
||
maximum_bytes=4096,
|
||
resume=True,
|
||
checkpoint=checkpoint,
|
||
alias="test-current",
|
||
)
|
||
|
||
http_error = urllib.error.HTTPError(
|
||
"http://127.0.0.1:9200/test",
|
||
429,
|
||
"busy",
|
||
{},
|
||
io.BytesIO(b"busy"),
|
||
)
|
||
with (
|
||
patch("search.minjust_opensearch.urllib.request.urlopen", side_effect=[http_error, io.BytesIO(b"{}")]),
|
||
patch("search.minjust_opensearch.time.sleep") as sleep,
|
||
):
|
||
self.assertEqual(
|
||
request_json("http://127.0.0.1:9200/test", "GET", None, "application/json", attempts=2),
|
||
{},
|
||
)
|
||
sleep.assert_called_once_with(1)
|
||
|
||
with (
|
||
patch(
|
||
"search.minjust_opensearch.urllib.request.urlopen",
|
||
side_effect=[io.BytesIO(b"{"), io.BytesIO(b"{}")],
|
||
),
|
||
patch("search.minjust_opensearch.time.sleep") as sleep,
|
||
):
|
||
self.assertEqual(
|
||
request_json(
|
||
"http://127.0.0.1:9200/_bulk",
|
||
"POST",
|
||
b"{}\n{}\n",
|
||
"application/x-ndjson",
|
||
attempts=2,
|
||
retry_invalid_json=True,
|
||
),
|
||
{},
|
||
)
|
||
sleep.assert_called_once_with(1)
|
||
|
||
bad = document_root / "editions/10/ru/fragments.json"
|
||
fragments = json.loads(bad.read_text(encoding="utf-8"))
|
||
fragments[0]["document_code"] = "8"
|
||
bad.write_text(json.dumps(fragments, ensure_ascii=False), encoding="utf-8")
|
||
with self.assertRaisesRegex(ValueError, "identity"):
|
||
export_bulk(root / "normalized", output, "test-index")
|
||
self.assertEqual(output.read_bytes(), content)
|
||
|
||
fragments[0]["document_code"] = "7"
|
||
fragments[0]["text"] = "Повреждено"
|
||
bad.write_text(json.dumps(fragments, ensure_ascii=False), encoding="utf-8")
|
||
with self.assertRaisesRegex(ValueError, "checksum"):
|
||
export_bulk(root / "normalized", output, "test-index")
|
||
with self.assertRaisesRegex(ValueError, "inside --input"):
|
||
export_bulk(root / "normalized", root / "normalized/manifest.sqlite3", "test-index")
|
||
self.assertEqual(output.read_bytes(), content)
|
||
|
||
bad.write_text("", encoding="utf-8")
|
||
with self.assertRaisesRegex(ValueError, "fragments.json"):
|
||
export_bulk(root / "normalized", output, "test-index")
|
||
self.assertEqual(output.read_bytes(), content)
|
||
|
||
definition = json.loads(
|
||
(Path(__file__).parent / "search/minjust-fragments-index.json").read_text(encoding="utf-8")
|
||
)
|
||
mapping = definition["mappings"]
|
||
self.assertEqual(definition["settings"]["index"]["number_of_shards"], 1)
|
||
self.assertEqual(definition["settings"]["index"]["number_of_replicas"], 0)
|
||
self.assertEqual(mapping["dynamic"], "strict")
|
||
self.assertEqual(mapping["properties"]["position"]["type"], "integer")
|
||
self.assertEqual(mapping["properties"]["text_ru"]["analyzer"], "russian")
|
||
self.assertEqual(mapping["properties"]["text_ky"]["analyzer"], "icu_analyzer")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|