406 lines
18 KiB
Python
406 lines
18 KiB
Python
import json
|
|
import os
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
import urllib.error
|
|
from email.message import Message
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from ingestion import minjust_current
|
|
from ingestion.minjust_current import (
|
|
ApiResponse,
|
|
CurrentCbdClient,
|
|
SourceChanged,
|
|
SourceError,
|
|
stage_document,
|
|
)
|
|
|
|
|
|
def api_response(value):
|
|
body = json.dumps(value, ensure_ascii=False).encode()
|
|
import hashlib
|
|
|
|
return ApiResponse(value, body, hashlib.sha256(body).hexdigest())
|
|
|
|
|
|
def document_record(alias="6", canonical=6, editions=None):
|
|
return api_response(
|
|
{
|
|
"documentCode": canonical,
|
|
"nameRus": "Закон",
|
|
"nameKyr": "Мыйзам",
|
|
"headerRu": "Заголовок",
|
|
"headerKg": "Баш сөз",
|
|
"number": 114,
|
|
"dateAdopted": "1998-08-04",
|
|
"classId": 1,
|
|
"refTypeId": 10,
|
|
"status": 1,
|
|
"editions": editions if editions is not None else [
|
|
{"id": 920492, "nameRus": "Редакция A", "nameKyr": "Редакция A", "textRusType": "docx", "textKyrType": "docx"},
|
|
{"id": 920292, "nameRus": "Редакция B", "nameKyr": "Редакция B", "textRusType": "docx", "textKyrType": "docx"},
|
|
],
|
|
}
|
|
)
|
|
|
|
|
|
class CurrentCbdClientTests(unittest.TestCase):
|
|
def test_rejects_legacy_roots_when_called_from_an_unrelated_working_directory(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
base = Path(temporary)
|
|
repository = base / "repository"
|
|
elsewhere = base / "elsewhere"
|
|
elsewhere.mkdir()
|
|
legacy_roots = [
|
|
repository / "data/minjust-cbd",
|
|
repository / "data/minjust-normalized",
|
|
]
|
|
for legacy in legacy_roots:
|
|
legacy.mkdir(parents=True)
|
|
# Root CI uses umask 000, so make the trusted fixture's parent permissions explicit.
|
|
for directory in (repository, repository / "data", *legacy_roots):
|
|
directory.chmod(0o755)
|
|
original_cwd = Path.cwd()
|
|
client = Mock()
|
|
try:
|
|
os.chdir(elsewhere)
|
|
with patch.object(minjust_current, "REPOSITORY_ROOT", repository):
|
|
for legacy in legacy_roots:
|
|
with self.subTest(legacy=legacy), self.assertRaisesRegex(ValueError, "must not overlap"):
|
|
stage_document(legacy / "staging", "6", client)
|
|
self.assertFalse((legacy / "staging").exists())
|
|
finally:
|
|
os.chdir(original_cwd)
|
|
client.document.assert_not_called()
|
|
|
|
def test_requires_private_owned_staging_root(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
output = Path(temporary) / "shared"
|
|
output.mkdir(mode=0o755)
|
|
output.chmod(0o755)
|
|
|
|
with self.assertRaisesRegex(ValueError, "private directory"):
|
|
stage_document(output, "6", Mock())
|
|
|
|
def test_resume_rejects_symlinked_checkpoints_and_document_paths(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
base = Path(temporary)
|
|
targets = [
|
|
("request-contract.json", False),
|
|
("current-api.sqlite3", False),
|
|
("manifest.sqlite3", False),
|
|
(".work/documents/6", True),
|
|
(".work/documents/6/editions/920492/ru.html", False),
|
|
("documents/6", True),
|
|
]
|
|
client = Mock()
|
|
for index, (relative, directory) in enumerate(targets):
|
|
with self.subTest(relative=relative):
|
|
output = base / f"generation-{index}"
|
|
output.mkdir(mode=0o700)
|
|
contract = output / "request-contract.json"
|
|
if relative != "request-contract.json":
|
|
contract.write_text(json.dumps(minjust_current.REQUEST_CONTRACT), encoding="utf-8")
|
|
outside = base / f"outside-{index}"
|
|
if directory:
|
|
outside.mkdir()
|
|
else:
|
|
outside.write_text("preserve", encoding="utf-8")
|
|
link = output / relative
|
|
link.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
link.symlink_to(outside, target_is_directory=directory)
|
|
except (NotImplementedError, OSError):
|
|
self.skipTest("symlinks are unavailable")
|
|
|
|
with self.assertRaisesRegex(ValueError, "symbolic link"):
|
|
stage_document(output, "6", client, resume=True)
|
|
|
|
client.document.assert_not_called()
|
|
if not directory:
|
|
self.assertEqual(outside.read_text(encoding="utf-8"), "preserve")
|
|
else:
|
|
self.assertEqual(list(outside.iterdir()), [])
|
|
|
|
def test_respects_retry_after_and_retries_server_errors(self):
|
|
headers = Message()
|
|
headers["Retry-After"] = "7"
|
|
response = Mock()
|
|
response.read.return_value = b"{}"
|
|
response.__enter__ = Mock(return_value=response)
|
|
response.__exit__ = Mock(return_value=False)
|
|
opener = Mock(side_effect=[
|
|
urllib.error.HTTPError("https://example.test", 429, "", headers, None),
|
|
urllib.error.HTTPError("https://example.test", 503, "", Message(), None),
|
|
response,
|
|
])
|
|
delays = []
|
|
|
|
CurrentCbdClient(opener=opener, sleep=delays.append, clock=lambda: 0, retries=3).request_json("/GetEdition")
|
|
|
|
self.assertIn(7.0, delays)
|
|
self.assertIn(2, delays)
|
|
self.assertEqual(opener.call_count, 3)
|
|
|
|
def test_enforces_request_interval(self):
|
|
now = [0.0]
|
|
starts = []
|
|
response = Mock()
|
|
response.read.return_value = b"{}"
|
|
response.__enter__ = Mock(return_value=response)
|
|
response.__exit__ = Mock(return_value=False)
|
|
|
|
def opener(_request, timeout):
|
|
starts.append(now[0])
|
|
return response
|
|
|
|
client = CurrentCbdClient(
|
|
opener=opener,
|
|
clock=lambda: now[0],
|
|
sleep=lambda delay: now.__setitem__(0, now[0] + delay),
|
|
)
|
|
client.request_json("/GetEdition")
|
|
client.request_json("/GetEdition")
|
|
|
|
self.assertGreaterEqual(starts[1] - starts[0], 1.0)
|
|
|
|
def test_rejects_oversized_response(self):
|
|
response = Mock()
|
|
response.read.return_value = b"x" * 11
|
|
response.__enter__ = Mock(return_value=response)
|
|
response.__exit__ = Mock(return_value=False)
|
|
with self.assertRaisesRegex(SourceError, "exceeds 10 bytes"):
|
|
CurrentCbdClient(opener=lambda *_args, **_kwargs: response, max_response_bytes=10).request_json("/GetEdition")
|
|
|
|
def test_get_edition_uses_exact_query_and_checks_requested_id(self):
|
|
response = Mock()
|
|
response.read.return_value = json.dumps({"id": 920292, "contentRu": "text", "contentKg": None}).encode()
|
|
response.__enter__ = Mock(return_value=response)
|
|
response.__exit__ = Mock(return_value=False)
|
|
requests = []
|
|
|
|
def opener(request, timeout):
|
|
requests.append(request)
|
|
return response
|
|
|
|
result = CurrentCbdClient(opener=opener, sleep=lambda _delay: None).edition(920292, "kg")
|
|
|
|
self.assertEqual(requests[0].full_url, "https://cbd.minjust.gov.kg/api/v1/GetEdition?editionId=920292&lang=kg&exact=true")
|
|
self.assertEqual(requests[0].method, "GET")
|
|
self.assertEqual(result.value["id"], 920292)
|
|
|
|
def test_rejects_response_id_that_does_not_match_exact_request(self):
|
|
response = Mock()
|
|
response.read.return_value = json.dumps({"id": 920492, "contentRu": "text", "contentKg": None}).encode()
|
|
response.__enter__ = Mock(return_value=response)
|
|
response.__exit__ = Mock(return_value=False)
|
|
|
|
with self.assertRaisesRegex(SourceChanged, "does not match requested ID"):
|
|
CurrentCbdClient(opener=lambda *_args, **_kwargs: response).edition(920292, "ru")
|
|
|
|
def test_stages_alias_mapping_language_quirk_and_official_edition_order(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
base = Path(temporary)
|
|
output = base / "generation"
|
|
old_raw = base / "legacy" / "documents" / "6" / "sentinel"
|
|
old_raw.parent.mkdir(parents=True)
|
|
old_raw.write_text("preserve", encoding="utf-8")
|
|
|
|
class Client:
|
|
def document(self, alias):
|
|
self.alias = alias
|
|
return document_record(alias=alias, canonical=6)
|
|
|
|
def edition(self, identifier, language):
|
|
text = "<p>Russian</p>" if language == "ru" else "<p>Кыргызча</p>"
|
|
name = "Редакция A" if identifier == 920492 else "Редакция B"
|
|
return api_response({
|
|
"id": identifier, "nameRus": name, "nameKyr": name,
|
|
"contentRu": text, "contentKg": None,
|
|
})
|
|
|
|
client = Client()
|
|
result = stage_document(output, "6", client)
|
|
raw = output / "documents" / "6"
|
|
|
|
self.assertEqual(result["state"], "text_complete_incomplete_attachments")
|
|
self.assertEqual((raw / "editions/920492/ru.html").read_text(), "<p>Russian</p>")
|
|
self.assertEqual((raw / "editions/920492/ky.html").read_text(), "<p>Кыргызча</p>")
|
|
first = json.loads((raw / "editions/920492/metadata.json").read_text())
|
|
second = json.loads((raw / "editions/920292/metadata.json").read_text())
|
|
self.assertEqual((first["SourceOrder"], second["SourceOrder"]), (0, 1))
|
|
language_meta = json.loads((raw / "editions/920292/source/ru.json").read_text())
|
|
self.assertEqual((language_meta["requested_edition_id"], language_meta["exact"]), (920292, True))
|
|
with sqlite3.connect(output / "current-api.sqlite3") as connection:
|
|
self.assertTrue(connection.execute("SELECT contract_sha256 FROM generation").fetchone()[0])
|
|
self.assertEqual(old_raw.read_text(), "preserve")
|
|
with sqlite3.connect(output / "current-api.sqlite3") as connection:
|
|
rows = connection.execute(
|
|
"SELECT request_language, archive_language, state FROM languages ORDER BY request_language, edition_id"
|
|
).fetchall()
|
|
self.assertEqual(rows, [("kg", "ky", "complete")] * 2 + [("ru", "ru", "complete")] * 2)
|
|
|
|
def test_resolves_official_alias_to_canonical_staging_directory(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
output = Path(temporary) / "generation"
|
|
|
|
class Client:
|
|
def document(self, alias):
|
|
return document_record(alias=alias, canonical=205064, editions=[])
|
|
|
|
result = stage_document(output, "1-4", Client())
|
|
|
|
self.assertEqual(result["canonical_id"], 205064)
|
|
self.assertTrue((output / "documents/205064/metadata.json").is_file())
|
|
self.assertFalse((output / "documents/1-4").exists())
|
|
|
|
def test_rejects_non_exact_edition_response_id_without_publishing(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
output = Path(temporary) / "generation"
|
|
|
|
class Client:
|
|
def document(self, _alias):
|
|
return document_record(editions=[{"id": 920292, "textRusType": "docx", "textKyrType": "docx"}])
|
|
|
|
def edition(self, _identifier, language):
|
|
return api_response({"id": 920492, "contentRu": f"<{language} edition>", "contentKg": None})
|
|
|
|
with self.assertRaisesRegex(SourceChanged, "exact edition ID mismatch"):
|
|
stage_document(output, "6", Client())
|
|
|
|
self.assertFalse((output / "documents/6").exists())
|
|
with sqlite3.connect(output / "current-api.sqlite3") as connection:
|
|
self.assertEqual(connection.execute("SELECT state FROM languages").fetchone()[0], "error")
|
|
|
|
def test_rejects_exact_edition_metadata_contradiction(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
output = Path(temporary) / "generation"
|
|
client = CurrentCbdClient()
|
|
client.document = lambda _alias: document_record(editions=[{
|
|
"id": 920292, "nameRus": "04.08.1998", "nameKyr": "04.08.1998",
|
|
"textRusType": "docx", "textKyrType": "docx",
|
|
}])
|
|
client.request_json = lambda *_args, **_kwargs: api_response({
|
|
"id": 920292, "nameRus": "13.04.2017", "nameKyr": "13.04.2017",
|
|
"contentRu": "<wrong edition>", "contentKg": None,
|
|
})
|
|
|
|
with self.assertRaisesRegex(SourceChanged, "metadata mismatch"):
|
|
stage_document(output, "6", client)
|
|
|
|
self.assertFalse((output / "documents/6").exists())
|
|
|
|
def test_recovers_atomic_publish_before_manifest_checkpoint(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
output = Path(temporary) / "generation"
|
|
|
|
class Client:
|
|
def document(self, _alias):
|
|
return document_record(editions=[{"id": 920492, "textRusType": "docx", "textKyrType": "docx"}])
|
|
|
|
def edition(self, identifier, language):
|
|
return api_response({"id": identifier, "contentRu": f"<{language}>", "contentKg": None})
|
|
|
|
client = Client()
|
|
first = stage_document(output, "6", client)
|
|
with sqlite3.connect(output / "current-api.sqlite3") as connection:
|
|
connection.execute("UPDATE documents SET state='fetching' WHERE alias='6'")
|
|
with sqlite3.connect(output / "manifest.sqlite3") as connection:
|
|
connection.execute("DELETE FROM documents WHERE code=6")
|
|
connection.execute("DELETE FROM errors WHERE code=6")
|
|
|
|
resumed = stage_document(output, "6", client, resume=True)
|
|
|
|
self.assertEqual(resumed["source_sha256"], first["source_sha256"])
|
|
self.assertTrue(resumed["resumed"])
|
|
with sqlite3.connect(output / "manifest.sqlite3") as connection:
|
|
self.assertEqual(connection.execute("SELECT COUNT(*) FROM documents WHERE code=6").fetchone()[0], 1)
|
|
self.assertEqual(connection.execute("SELECT COUNT(*) FROM errors WHERE code=6").fetchone()[0], 1)
|
|
|
|
def test_retries_after_document_metadata_request_failed_before_checkpoint(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
output = Path(temporary) / "generation"
|
|
|
|
class Client:
|
|
failed = False
|
|
|
|
def document(self, _alias):
|
|
if not self.failed:
|
|
self.failed = True
|
|
raise SourceError("temporary DNS failure")
|
|
return document_record(editions=[])
|
|
|
|
client = Client()
|
|
with self.assertRaisesRegex(SourceError, "DNS"):
|
|
stage_document(output, "6", client)
|
|
|
|
result = stage_document(output, "6", client, resume=True)
|
|
|
|
self.assertEqual(result["canonical_id"], 6)
|
|
with sqlite3.connect(output / "current-api.sqlite3") as connection:
|
|
row = connection.execute(
|
|
"SELECT canonical_id, response_sha256, expected_editions FROM documents WHERE alias='6'"
|
|
).fetchone()
|
|
self.assertEqual(row, (6, document_record(editions=[]).sha256, 0))
|
|
|
|
def test_does_not_resume_generation_without_exact_request_contract(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
output = Path(temporary) / "legacy-generation"
|
|
output.mkdir(mode=0o700)
|
|
sentinel = output / "old-response.json"
|
|
sentinel.write_text("preserve", encoding="utf-8")
|
|
client = Mock()
|
|
|
|
with self.assertRaisesRegex(SourceChanged, "no exact-edition request contract"):
|
|
stage_document(output, "6", client, resume=True)
|
|
|
|
client.document.assert_not_called()
|
|
self.assertEqual(sentinel.read_text(encoding="utf-8"), "preserve")
|
|
|
|
def test_incomplete_translation_resumes_without_rewriting_successful_language(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
output = Path(temporary) / "generation"
|
|
|
|
class Client:
|
|
failed = False
|
|
|
|
def document(self, _alias):
|
|
return document_record(editions=[{"id": 920492, "textRusType": "docx", "textKyrType": "docx"}])
|
|
|
|
def edition(self, identifier, language):
|
|
if language == "kg" and not self.failed:
|
|
self.failed = True
|
|
raise SourceError("simulated interrupted response")
|
|
return api_response({"id": identifier, "contentRu": f"<{language}>", "contentKg": None})
|
|
|
|
client = Client()
|
|
first = stage_document(output, "6", client)
|
|
self.assertIn("incomplete", first)
|
|
self.assertFalse((output / "documents/6").exists())
|
|
second = stage_document(output, "6", client, resume=True)
|
|
self.assertEqual(second["state"], "text_complete_incomplete_attachments")
|
|
self.assertEqual((output / "documents/6/editions/920492/ru.html").read_text(), "<ru>")
|
|
self.assertEqual((output / "documents/6/editions/920492/ky.html").read_text(), "<kg>")
|
|
self.assertEqual(client.failed, True)
|
|
|
|
def test_refuses_budget_overrun_without_publishing_partial_document(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
output = Path(temporary) / "generation"
|
|
|
|
class Client:
|
|
def document(self, _alias):
|
|
return document_record(editions=[{"id": 920492, "textRusType": "docx", "textKyrType": "docx"}])
|
|
|
|
def edition(self, identifier, _language):
|
|
return api_response({"id": identifier, "contentRu": "123456", "contentKg": None})
|
|
|
|
with self.assertRaisesRegex(SourceError, "budget"):
|
|
stage_document(output, "6", Client(), max_document_bytes=5)
|
|
self.assertFalse((output / "documents/6").exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|