"""Offline tests. Ephemeral test keys, mocked facilitator/RPC. No real funds or settlement.""" import base64 import copy import json import secrets import threading import time from concurrent.futures import ThreadPoolExecutor import pytest from eth_account import Account from fastapi.testclient import TestClient from engine import EXAMPLE, plan from server import (ASSET, AUTH_TOPIC, ENDPOINT, NETWORK, PAYEE, REQUIREMENTS, TRANSFER_TOPIC, LivePayments, b64, make_app, typed_authorization) SIMULATED_TX = "0x" + "00" * 32 def payment_header(**overrides): account = Account.create() auth = {"from": account.address, "to": PAYEE, "value": "10000", "validAfter": str(int(time.time()) - 5), "validBefore": str(int(time.time()) + 240), "nonce": "0x" + secrets.token_hex(32)} auth.update(overrides) signature = "0x" + account.sign_message(typed_authorization(auth)).signature.hex() return b64({"x402Version": 2, "resource": {"url": ENDPOINT}, "accepted": REQUIREMENTS, "payload": {"authorization": auth, "signature": signature}}) class MockPayments: def __init__(self, pending=False, fail=False, reject=False): self.settlements = 0 self.pending, self.fail, self.reject = pending, fail, reject self.lock = threading.Lock() def verify(self, payment): return not self.reject def block(self): return 100 def settle(self, payment): with self.lock: self.settlements += 1 if self.fail: raise TimeoutError("SIMULATED lost response AFTER broadcast") return {"success": True, "transaction": SIMULATED_TX} def reconcile(self, row, auth): if self.pending: return None, row["scan_block"] return {"success": True, "network": NETWORK, "transaction": SIMULATED_TX, "payer": auth["from"], "confirmation": "SIMULATED_SAFE_RECEIPT"}, row["scan_block"] @pytest.fixture def service(tmp_path): payments = MockPayments() return TestClient(make_app(tmp_path / "calls.db", payments)), payments, tmp_path / "calls.db" def test_example_balances(): result = plan(EXAMPLE) assert result["bars_required"] == 3 assert result["balance_verified"] assert result["optimality"] == "proven_bar_count_by_volume_bound" assert result["pieces_delivered"] == 9 def test_unpaid_challenge(service): client, p, _ = service response = client.post("/kerfledger/v1/plan", json=EXAMPLE) assert response.status_code == 402 assert json.loads(base64.b64decode(response.headers["payment-required"])) == response.json() assert response.json()["accepts"] == [REQUIREMENTS] assert p.settlements == 0 def test_paid_path_and_cached_replay(service): client, p, _ = service headers = {"PAYMENT-SIGNATURE": payment_header()} first = client.post("/kerfledger/v1/plan", json=EXAMPLE, headers=headers) second = client.post("/kerfledger/v1/plan", json=EXAMPLE, headers=headers) assert first.status_code == second.status_code == 200 assert first.json() == second.json() == plan(EXAMPLE) assert "payment-response" in second.headers assert p.settlements == 1 def test_changed_input_conflicts(service): client, p, _ = service headers = {"PAYMENT-SIGNATURE": payment_header()} assert client.post("/kerfledger/v1/plan", json=EXAMPLE, headers=headers).status_code == 200 changed = dict(EXAMPLE, stock_mm=7000) assert client.post("/kerfledger/v1/plan", json=changed, headers=headers).status_code == 409 assert p.settlements == 1 def test_restart_recovery(service): client, p, path = service headers = {"PAYMENT-SIGNATURE": payment_header()} first = client.post("/kerfledger/v1/plan", json=EXAMPLE, headers=headers) restarted_payments = MockPayments(reject=True) restarted = TestClient(make_app(path, restarted_payments)) second = restarted.post("/kerfledger/v1/plan", json=EXAMPLE, headers=headers) assert first.content == second.content assert restarted_payments.settlements == 0 def test_lost_response_reconciles_without_second_settle(tmp_path): p = MockPayments(pending=True, fail=True) path = tmp_path / "calls.db" client = TestClient(make_app(path, p)) headers = {"PAYMENT-SIGNATURE": payment_header()} first = client.post("/kerfledger/v1/plan", json=EXAMPLE, headers=headers) assert first.status_code == 202 and first.headers["retry-after"] == "15" p.pending = False client = TestClient(make_app(path, p)) assert client.post("/kerfledger/v1/plan", json=EXAMPLE, headers=headers).status_code == 200 assert p.settlements == 1 def test_concurrent_replay_one_settlement(service): client, p, _ = service header = payment_header() def call(_): return client.post("/kerfledger/v1/plan", json=EXAMPLE, headers={"PAYMENT-SIGNATURE": header}).status_code with ThreadPoolExecutor(max_workers=5) as pool: statuses = list(pool.map(call, range(5))) assert all(s in (200, 202) for s in statuses) assert p.settlements == 1 @pytest.mark.parametrize("changes", [{"value": "9999"}, {"to": "0x" + "11" * 20}, {"validBefore": "1"}]) def test_wrong_payment_rejected(service, changes): client, p, _ = service response = client.post("/kerfledger/v1/plan", json=EXAMPLE, headers={"PAYMENT-SIGNATURE": payment_header(**changes)}) assert response.status_code == 402 assert p.settlements == 0 def test_forged_signature(service): client, p, _ = service payment = json.loads(base64.b64decode(payment_header())) payment["payload"]["authorization"]["from"] = "0x" + "22" * 20 assert client.post("/kerfledger/v1/plan", json=EXAMPLE, headers={"PAYMENT-SIGNATURE": b64(payment)}).status_code == 402 assert p.settlements == 0 @pytest.mark.parametrize("changes", [{"kerf_mm": True}, {"stock_mm": 10}, {"stock_mm": 6000.5}, {"pieces": [{"length_mm": 100, "quantity": 201}]}, {"extra": 1}]) def test_bad_input_no_charge(service, changes): client, p, _ = service assert client.post("/kerfledger/v1/plan", json=dict(EXAMPLE, **changes), headers={"PAYMENT-SIGNATURE": payment_header()}).status_code == 422 assert p.settlements == 0 def test_real_receipt_validator_rejects_wrong_transfer(): payer = "0x" + "11" * 20 nonce = "0x" + "12" * 32 auth = {"from": payer, "nonce": nonce} logs = [{"address": ASSET, "topics": [AUTH_TOPIC, "0x" + payer[2:].zfill(64), nonce], "data": "0x"}, {"address": ASSET, "topics": [TRANSFER_TOPIC, "0x" + payer[2:].zfill(64), "0x" + PAYEE[2:].lower().zfill(64)], "data": hex(10000)}] live = LivePayments() def rpc(method, params): if method == "eth_getBlockByNumber": return {"number": hex(1000)} if method == "eth_getTransactionReceipt": return {"status": "0x1", "blockNumber": hex(999), "logs": logs} raise AssertionError(method) live.rpc = rpc assert live.reconcile({"tx": SIMULATED_TX, "scan_block": 990}, auth)[0]["success"] logs[1]["data"] = hex(9999) assert live.reconcile({"tx": SIMULATED_TX, "scan_block": 990}, auth)[0] is None def test_many_generated_jobs_preserve_pieces_and_balance(): import random from collections import Counter rng = random.Random(402) for _ in range(100): pieces = [{"length_mm": rng.randint(1, 990), "quantity": rng.randint(1, 5)} for i in range(10)] data = {"stock_mm": 1000, "kerf_mm": 3, "pieces": pieces} result = plan(data) expected = Counter() for row in pieces: expected[row["length_mm"]] += row["quantity"] assert Counter(x for b in result["bars"] for x in b["cuts_mm"]) == expected assert result["balance_verified"] assert all(b["remaining_mm"] >= 0 for b in result["bars"])