107 lines
3.7 KiB
Python
107 lines
3.7 KiB
Python
"""Global VeraeTime client: same wire shape as verae-timestamp/global.Client."""
|
|
|
|
import pytest
|
|
|
|
from verae_mcp_timestamping_server.global_client import GlobalClient, GlobalLedger
|
|
from verae_mcp_timestamping_server.ledger import SHA256_EMPTY, MemoryLedger, RoutingLedger
|
|
|
|
pytestmark = pytest.mark.anyio
|
|
|
|
|
|
class _Resp:
|
|
def __init__(self, status_code: int, body: dict):
|
|
self.status_code = status_code
|
|
self.body = body
|
|
self.text = str(body)
|
|
self.headers = {"Content-Type": "application/json"}
|
|
|
|
def json(self):
|
|
return self.body
|
|
|
|
|
|
class FakeHttp:
|
|
def __init__(self):
|
|
self.calls: list[tuple[str, str, dict, dict]] = []
|
|
|
|
def request(self, method: str, url: str, payload: dict, headers: dict) -> _Resp:
|
|
self.calls.append((method, url, payload, headers))
|
|
if url.endswith("/api/timestamp"):
|
|
return _Resp(
|
|
200,
|
|
{
|
|
"jobId": "job-1",
|
|
"hash": payload.get("data") or payload.get("hash"),
|
|
"timestamp": "2026-08-15T00:00:00Z",
|
|
"blockHash": "0xdead",
|
|
},
|
|
)
|
|
if url.endswith("/api/verify"):
|
|
if payload.get("certificate") == "job-1":
|
|
return _Resp(200, {"valid": True, "hash": SHA256_EMPTY})
|
|
return _Resp(200, {"valid": False})
|
|
if "/api/status/" in url:
|
|
return _Resp(
|
|
200,
|
|
{
|
|
"id": "job-1",
|
|
"status": "completed",
|
|
"result": "job-1",
|
|
"metadata": {"timestamp": "2026-08-15T00:00:00Z", "blockHash": "0xdead"},
|
|
},
|
|
)
|
|
return _Resp(404, {"error": "nope"})
|
|
|
|
|
|
async def test_global_stamp_posts_hash_sync_and_maps_receipt():
|
|
http = FakeHttp()
|
|
client = GlobalClient("https://api.veraetime.net", token="secret", http=http)
|
|
ledger = GlobalLedger(client)
|
|
[rec] = await ledger.stamp(SHA256_EMPTY, scope="global", cube_id="cube-1")
|
|
method, url, payload, headers = http.calls[0]
|
|
assert method == "POST"
|
|
assert url == "https://api.veraetime.net/api/timestamp"
|
|
assert payload["data"] == SHA256_EMPTY
|
|
assert payload["digest"] == SHA256_EMPTY
|
|
assert payload["hashAlg"] == "SHA256"
|
|
assert payload["restamp"] is False
|
|
assert headers["Authorization"] == "Bearer secret"
|
|
assert rec.scope == "global"
|
|
assert rec.receipt_id == "job-1"
|
|
assert rec.merkle == "0xdead"
|
|
assert rec.cube_id == "cube-1"
|
|
|
|
|
|
async def test_global_verify_posts_certificate():
|
|
http = FakeHttp()
|
|
ledger = GlobalLedger(GlobalClient("https://api.veraetime.net", http=http))
|
|
ok = await ledger.verify("job-1", want_scope="global")
|
|
assert ok.valid is True
|
|
assert ok.receipt is not None
|
|
assert ok.receipt.hash == SHA256_EMPTY
|
|
assert ok.receipt.scope == "global"
|
|
_, url, payload, _ = http.calls[0]
|
|
assert url.endswith("/api/verify")
|
|
assert payload == {"certificate": "job-1"}
|
|
|
|
|
|
async def test_routing_sends_only_global_scope_to_central():
|
|
http = FakeHttp()
|
|
local = MemoryLedger()
|
|
routed = RoutingLedger(
|
|
local=local,
|
|
global_ledger=GlobalLedger(GlobalClient("https://api.veraetime.net", http=http)),
|
|
)
|
|
[local_rec] = await routed.stamp(SHA256_EMPTY, scope="local")
|
|
assert local_rec.scope == "local"
|
|
assert http.calls == []
|
|
|
|
[g] = await routed.stamp(SHA256_EMPTY, scope="global", cube_id="cube-1")
|
|
assert g.scope == "global"
|
|
assert g.receipt_id == "job-1"
|
|
assert http.calls[0][1].endswith("/api/timestamp")
|
|
|
|
cached = await routed.hash_report(SHA256_EMPTY, want_scope="global")
|
|
assert cached == []
|
|
|
|
weak = await routed.verify(local_rec.receipt_id, want_scope="global")
|
|
assert weak.valid is False
|