8ccd6827cb
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1029d-2c93-7500-a0e0-e3e089668b29
60 lines
2.4 KiB
Python
60 lines
2.4 KiB
Python
"""Offline producer checks using the real reviewed NetKingdom validator."""
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
NK = ROOT.parent / "net-kingdom"
|
|
|
|
|
|
def module(name, path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
result = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(result)
|
|
return result
|
|
|
|
|
|
publisher = module("publisher", ROOT / "scripts/root_grant_publish.py")
|
|
validator = module("custody_validator", NK / "tools/custody-record/validate.py")
|
|
|
|
|
|
class ProducerTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.first = json.loads((NK / "tools/custody-record/fixtures/valid/grant.json").read_text())
|
|
self.first["record_id"] = publisher.RECORD_ID
|
|
self.next = copy.deepcopy(self.first)
|
|
self.next["revision"] += 1
|
|
|
|
def test_bootstrap_requires_absent_baseline_and_cas_zero(self):
|
|
self.assertEqual(publisher.prepare_payload(self.first, None, 0, validator)["options"], {"cas": 0})
|
|
for old, version in [(None, 1), (self.first, 0), (None, True), (None, -1)]:
|
|
with self.assertRaises(ValueError):
|
|
publisher.prepare_payload(self.first, old, version, validator)
|
|
|
|
def test_update_is_whole_document_and_exact_cas(self):
|
|
result = publisher.prepare_payload(self.next, self.first, 7, validator)
|
|
self.assertEqual(result["options"], {"cas": 7})
|
|
self.assertEqual(result["data"]["document"], self.next)
|
|
result["data"]["document"]["entries"].clear()
|
|
self.assertTrue(self.next["entries"])
|
|
|
|
def test_rollback_same_revision_and_entry_deletion_denied(self):
|
|
for doc in [self.first, dict(self.next, revision=1), dict(self.next, entries=[])]:
|
|
with self.assertRaises(ValueError):
|
|
publisher.prepare_payload(doc, self.first, 1, validator)
|
|
|
|
def test_wrong_record_and_secret_errors_do_not_echo(self):
|
|
for doc in [dict(self.first, record_id="nk-custody:other"), dict(self.first, password="DO-NOT-ECHO")]:
|
|
with self.assertRaises(ValueError) as raised:
|
|
publisher.prepare_payload(doc, None, 0, validator)
|
|
self.assertNotIn("DO-NOT-ECHO", str(raised.exception))
|
|
|
|
def test_invalid_previous_cannot_reset_baseline(self):
|
|
with self.assertRaises(ValueError):
|
|
publisher.prepare_payload(self.next, {}, 1, validator)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|