158 lines
6.5 KiB
Python
158 lines
6.5 KiB
Python
"""The publish step: dry run by default, gated, and only ever one block."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from hermes_nano_x402 import nano_crypto as nc
|
|
from hermes_nano_x402.client import X402NanoClient
|
|
from hermes_nano_x402.guard import PaymentGuard
|
|
from hermes_nano_x402.publish import PublishGate, PublishRefused, publish
|
|
from hermes_nano_x402.quote import PaymentRefused, parse_quote
|
|
|
|
from helpers import ONE_CALL_RAW, OTHER_PAY_TO, PAY_TO, FakeSigner, ScriptedTransport, b64, challenge, quote_doc, served
|
|
|
|
|
|
def _quote(**kwargs: Any):
|
|
return parse_quote(b64(quote_doc(**kwargs)), resource_url="https://pursekeeper.dev/v1/echo?msg=hello", now=0)
|
|
|
|
|
|
def _block(link: str = PAY_TO, signature: str = "S" * 128) -> dict:
|
|
return {"type": "send", "link": link, "signature": signature, "balance": "1", "previous": "A" * 64}
|
|
|
|
|
|
def test_dry_run_is_the_default_and_never_touches_the_network(monkeypatch):
|
|
monkeypatch.delenv("NANO_PUBLISH", raising=False)
|
|
gate = PublishGate(pay_to_allowlist=[PAY_TO], max_raw_per_block=ONE_CALL_RAW)
|
|
out = publish("https://rpc.invalid", quote=_quote(), block=_block(), block_hash="H" * 64, gate=gate)
|
|
assert out == {"published": False, "dry_run": True, "hash": "H" * 64}
|
|
|
|
|
|
def test_gate_refuses_a_destination_outside_the_allowlist():
|
|
gate = PublishGate(pay_to_allowlist=[PAY_TO], max_raw_per_block=ONE_CALL_RAW)
|
|
with pytest.raises(PublishRefused) as exc:
|
|
gate.authorize(quote=_quote(pay_to=OTHER_PAY_TO), block=_block(link=OTHER_PAY_TO))
|
|
assert exc.value.code == "pay_to_not_allowed"
|
|
|
|
|
|
def test_gate_refuses_over_the_cap():
|
|
gate = PublishGate(pay_to_allowlist=[PAY_TO], max_raw_per_block=ONE_CALL_RAW - 1)
|
|
with pytest.raises(PublishRefused) as exc:
|
|
gate.authorize(quote=_quote(), block=_block())
|
|
assert exc.value.code == "over_cap"
|
|
|
|
|
|
def test_gate_refuses_a_link_that_is_not_the_quoted_account():
|
|
gate = PublishGate(pay_to_allowlist=[PAY_TO], max_raw_per_block=ONE_CALL_RAW)
|
|
with pytest.raises(PublishRefused) as exc:
|
|
gate.authorize(quote=_quote(), block=_block(link="nano_3" + "c" * 59))
|
|
assert exc.value.code == "link_mismatch"
|
|
|
|
|
|
def test_gate_accepts_a_hex_link_equal_to_the_quoted_account():
|
|
gate = PublishGate(pay_to_allowlist=[PAY_TO], max_raw_per_block=ONE_CALL_RAW)
|
|
hexlink = nc.decode_address(PAY_TO).hex()
|
|
gate.authorize(quote=_quote(), block=_block(link=hexlink))
|
|
|
|
|
|
@pytest.mark.parametrize("signature", ["", "abc"])
|
|
def test_gate_refuses_an_unsigned_block(signature):
|
|
gate = PublishGate(pay_to_allowlist=[PAY_TO], max_raw_per_block=ONE_CALL_RAW)
|
|
with pytest.raises(PublishRefused) as exc:
|
|
gate.authorize(quote=_quote(), block=_block(signature=signature))
|
|
assert exc.value.code == "unsigned"
|
|
|
|
|
|
class _FakeResp:
|
|
def __init__(self, doc: dict) -> None:
|
|
self._doc = doc
|
|
|
|
def read(self) -> bytes:
|
|
return json.dumps(self._doc).encode()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
|
|
def _patch_urlopen(monkeypatch, doc: dict, seen: list | None = None):
|
|
def fake(req, timeout=None):
|
|
if seen is not None:
|
|
seen.append(json.loads(req.data))
|
|
return _FakeResp(doc)
|
|
|
|
monkeypatch.setattr("hermes_nano_x402.publish.urllib.request.urlopen", fake)
|
|
|
|
|
|
def test_publish_broadcasts_one_json_block_when_enabled(monkeypatch):
|
|
seen: list = []
|
|
_patch_urlopen(monkeypatch, {"hash": "H" * 64}, seen)
|
|
gate = PublishGate(pay_to_allowlist=[PAY_TO], max_raw_per_block=ONE_CALL_RAW)
|
|
out = publish(
|
|
"https://rpc.example", quote=_quote(), block=_block(), block_hash="H" * 64, gate=gate, dry_run=False
|
|
)
|
|
assert out["published"] is True
|
|
assert len(seen) == 1
|
|
assert seen[0]["action"] == "process" and seen[0]["subtype"] == "send" and seen[0]["do_work"] is True
|
|
assert seen[0]["json_block"] == "true"
|
|
|
|
|
|
def test_a_block_already_on_chain_counts_as_published(monkeypatch):
|
|
_patch_urlopen(monkeypatch, {"error": "Old block"})
|
|
gate = PublishGate(pay_to_allowlist=[PAY_TO], max_raw_per_block=ONE_CALL_RAW)
|
|
out = publish("https://rpc.example", quote=_quote(), block=_block(), block_hash="H" * 64, gate=gate, dry_run=False)
|
|
assert out["published"] is True
|
|
|
|
|
|
def test_a_node_rejection_is_a_refusal(monkeypatch):
|
|
_patch_urlopen(monkeypatch, {"error": "Bad signature"})
|
|
gate = PublishGate(pay_to_allowlist=[PAY_TO], max_raw_per_block=ONE_CALL_RAW)
|
|
with pytest.raises(PublishRefused) as exc:
|
|
publish("https://rpc.example", quote=_quote(), block=_block(), block_hash="H" * 64, gate=gate, dry_run=False)
|
|
assert exc.value.code == "node_rejected"
|
|
|
|
|
|
def test_publish_is_enabled_only_by_an_explicit_operator_flag(monkeypatch):
|
|
from hermes_nano_x402.publish import publish_enabled
|
|
|
|
monkeypatch.delenv("NANO_PUBLISH", raising=False)
|
|
assert publish_enabled() is False
|
|
monkeypatch.setenv("NANO_PUBLISH", "1")
|
|
assert publish_enabled() is True
|
|
|
|
|
|
def test_client_settles_then_presents_and_reports_it_was_a_dry_run():
|
|
transport = ScriptedTransport(challenge(), served())
|
|
guard = PaymentGuard(allowed_pay_to=(PAY_TO,), allowed_origins=("https://pursekeeper.dev",), per_call_cap_xno=Decimal("0.01"), total_cap_xno=Decimal("0.05"))
|
|
client = X402NanoClient(
|
|
signer=FakeSigner(),
|
|
guard=guard,
|
|
transport=transport,
|
|
publish_gate=PublishGate(pay_to_allowlist=[PAY_TO], max_raw_per_block=ONE_CALL_RAW),
|
|
publish_dry_run=True,
|
|
)
|
|
out = client.get("https://pursekeeper.dev/v1/echo?msg=hello")
|
|
assert out.status == 200 and out.paid is True and out.published is False
|
|
assert len(transport.signed_requests) == 1
|
|
|
|
|
|
def test_client_does_not_present_a_block_it_could_not_settle(monkeypatch):
|
|
transport = ScriptedTransport(challenge(), served())
|
|
guard = PaymentGuard(allowed_pay_to=(PAY_TO,), allowed_origins=("https://pursekeeper.dev",), per_call_cap_xno=Decimal("0.01"), total_cap_xno=Decimal("0.05"))
|
|
client = X402NanoClient(
|
|
signer=FakeSigner(),
|
|
guard=guard,
|
|
transport=transport,
|
|
publish_gate=PublishGate(pay_to_allowlist=[PAY_TO], max_raw_per_block=ONE_CALL_RAW),
|
|
publish_dry_run=False,
|
|
)
|
|
monkeypatch.setattr("hermes_nano_x402.publish.urllib.request.urlopen", lambda *a, **k: (_ for _ in ()).throw(OSError("no route")))
|
|
with pytest.raises(PaymentRefused) as exc:
|
|
client.get("https://pursekeeper.dev/v1/echo?msg=hello")
|
|
assert exc.value.code == "publish_failed"
|
|
assert transport.signed_requests == []
|