64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
"""NatsLedger speaks locald subjects: verae.ts.request / verify / hash.report."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from verae_mcp_timestamping_server.ledger import SHA256_EMPTY, MemoryLedger, Receipt
|
|
from verae_mcp_timestamping_server.nats_ledger import NatsLedger
|
|
|
|
pytestmark = pytest.mark.anyio
|
|
|
|
|
|
class _Msg:
|
|
def __init__(self, data: bytes):
|
|
self.data = data
|
|
|
|
|
|
class _FakeNc:
|
|
"""Request/reply shaped like nats-py, backed by MemoryLedger (locald protocol)."""
|
|
|
|
def __init__(self, inner: MemoryLedger):
|
|
self.inner = inner
|
|
|
|
async def request(self, subject: str, payload: bytes, timeout: float = 2.0):
|
|
body = json.loads(payload)
|
|
if subject == "verae.ts.request":
|
|
receipts = await self.inner.stamp(
|
|
body["hash"],
|
|
scope=body.get("scope") or "local",
|
|
cube_id=body.get("cube_id") or "",
|
|
org_id=body.get("org_id") or "",
|
|
)
|
|
return _Msg(receipts[0].model_dump_json().encode())
|
|
if subject == "verae.ts.verify":
|
|
result = await self.inner.verify(
|
|
body["receipt_id"], want_scope=body.get("want_scope") or ""
|
|
)
|
|
if not result.valid or result.receipt is None:
|
|
return _Msg(json.dumps({"valid": False, "error": result.error}).encode())
|
|
return _Msg(result.receipt.model_dump_json().encode())
|
|
if subject == "verae.hash.report":
|
|
receipts = await self.inner.hash_report(
|
|
body["hash"], org_id=body.get("org_id") or ""
|
|
)
|
|
return _Msg(
|
|
json.dumps({"receipts": [r.model_dump() for r in receipts]}).encode()
|
|
)
|
|
raise AssertionError(f"unexpected subject {subject}")
|
|
|
|
|
|
async def test_nats_stamp_verify_report_roundtrip():
|
|
nc = _FakeNc(MemoryLedger())
|
|
ledger = NatsLedger(nc)
|
|
receipts = await ledger.stamp(SHA256_EMPTY, scope="local", cube_id="cube-1")
|
|
assert receipts[0].hash == SHA256_EMPTY
|
|
assert receipts[0].cube_id == "cube-1"
|
|
|
|
ok = await ledger.verify(receipts[0].receipt_id, want_scope="local")
|
|
assert ok.valid is True
|
|
assert ok.receipt is not None
|
|
|
|
report = await ledger.hash_report(SHA256_EMPTY)
|
|
assert [r.receipt_id for r in report] == [receipts[0].receipt_id]
|
|
assert isinstance(report[0], Receipt)
|