Practical lab guide
MicroBank 5: verify a real transaction with Python
Create synthetic local data, check idempotency, and wait for the corresponding Ledger entry.
On this page
Test the implemented contract
Run this only against your microbank-study local profile. The probe's addresses are fixed loopback URLs, it disables environment HTTP proxies and redirects, and it uses a synthetic .invalid email. It creates one account and a 1,000-cent deposit per --create invocation. It never calls a real banking service.
This checks sequential repetition of one idempotency key, not concurrent exactly-once processing. It verifies the actual Ledger API because the source has no Accounts settlement consumer. Accounts can still report pending; do not invent a settled result.
Save scripts/study/probe.py:
import argparse
import json
from pathlib import Path
import sys
import time
import uuid
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener
ACCOUNTS = "http://127.0.0.1:8000"
LEDGER = "http://127.0.0.1:8001"
class NoRedirect(HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
raise ValueError("Unexpected redirect from a local API")
HTTP = build_opener(ProxyHandler({}), NoRedirect())
def request(method, url, payload=None, key=None):
if not (url.startswith(ACCOUNTS + "/") or url.startswith(LEDGER + "/")):
raise ValueError("The probe only supports its two local services")
headers = {"Accept": "application/json"}
data = None
if payload is not None:
data = json.dumps(payload).encode()
headers["Content-Type"] = "application/json"
if key:
headers["Idempotency-Key"] = key
with HTTP.open(Request(url, data=data, headers=headers, method=method), timeout=3) as response:
return json.load(response)
def create_case(call=request):
identity = str(uuid.uuid4())
user = call("POST", ACCOUNTS + "/v1/users/", {"email": f"study-{identity}@example.invalid"})
account = str(uuid.UUID(user["account_id"]))
payload = {"account_id": account, "kind": "deposit", "amount_cents": 1000}
first = call("POST", ACCOUNTS + "/v1/transactions/", payload, identity)
repeated = call("POST", ACCOUNTS + "/v1/transactions/", payload, identity)
if first["id"] != repeated["id"]:
raise RuntimeError("Repeated key returned a different transaction")
return {"account_id": account, "tx_id": str(uuid.UUID(first["id"])), "expected_cents": 1000}
def verify_case(case, call=request, pause=time.sleep, attempts=15):
account = str(uuid.UUID(case["account_id"]))
transaction = str(uuid.UUID(case["tx_id"]))
for attempt in range(attempts):
balance = call("GET", LEDGER + f"/v1/balances/{account}")
entries = call("GET", LEDGER + f"/v1/entries/{account}")
matching = [entry for entry in entries if entry.get("txId") == transaction]
if balance.get("balanceCents") == case["expected_cents"] and len(matching) == 1:
return {**case, "ledger_entries_for_tx": 1, "ledger_verified": True}
if attempt + 1 < attempts:
pause(2)
raise RuntimeError("Ledger result did not converge within the bounded attempts")
def main():
parser = argparse.ArgumentParser(description="Local MicroBank transaction check")
mode = parser.add_mutually_exclusive_group(required=True)
mode.add_argument("--create", action="store_true")
mode.add_argument("--verify", type=Path, metavar="CASE_JSON")
parser.add_argument("--case-file", type=Path, default=Path("evidence/case.json"))
args = parser.parse_args()
try:
if args.create:
if args.case_file.exists():
raise ValueError("Choose a new --case-file to preserve the previous case")
case = create_case()
args.case_file.parent.mkdir(parents=True, exist_ok=True)
with args.case_file.open("x") as output:
json.dump(case, output, indent=2)
else:
case = json.loads(args.verify.read_text())
print(json.dumps(verify_case(case), sort_keys=True))
return 0
except (OSError, ValueError, KeyError, TypeError, RuntimeError) as error:
print(f"MicroBank check failed: {error}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())
An API error fails immediately. Ledger convergence has 15 attempts, two-second gaps, and three-second request timeouts. A saved case survives a convergence failure so recovery can verify the same transaction without submitting it again. Interrupted creation before the case is saved requires inspecting the disposable data; this small probe is not a production transaction client.
Read the probe in four responsibilities
request owns the HTTP exchange, including the fixed local transport behavior. create_case makes synthetic records and repeats a deposit's idempotency key. verify_case checks the matching Ledger result with bounded polling. main separates creating a new case from verifying a saved one and reports failure through the process exit status.
The injected call and pause functions let the unit tests substitute fake responses and avoid real waiting. That tests the probe's decisions without contacting Accounts or Ledger. A saved case file connects a later recovery check to the same transaction, instead of accidentally creating another deposit every time you want to observe progress.
Before a live run, predict where failures belong: rejected HTTP request, malformed response, duplicate/inconsistent result, or Ledger not ready within the bounded attempts. Do not treat every timeout as proof that the deposit was never accepted; inspect the saved case and application state.
Unit-test failures before the live run
Save tests/study/test_probe.py. These are fake-transport tests of the probe, not proof that MicroBank is running:
import sys
from pathlib import Path
import unittest
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "scripts/study"))
from probe import NoRedirect, create_case, request, verify_case
ACCOUNT = "11111111-1111-4111-8111-111111111111"
TX = "22222222-2222-4222-8222-222222222222"
CASE = {"account_id": ACCOUNT, "tx_id": TX, "expected_cents": 1000}
class ProbeTests(unittest.TestCase):
def test_create_reuses_payload_and_key(self):
calls = []
def fake(method, url, payload=None, key=None):
calls.append((method, url, payload, key))
return {"account_id": ACCOUNT} if url.endswith("/users/") else {"id": TX}
self.assertEqual(create_case(fake), CASE)
self.assertEqual(calls[1], calls[2])
self.assertTrue(calls[1][3])
def test_different_transaction_is_failure(self):
replies = iter([{"account_id": ACCOUNT}, {"id": TX}, {"id": ACCOUNT}])
with self.assertRaises(RuntimeError):
create_case(lambda *args: next(replies))
def test_waits_for_ledger(self):
replies = iter([{"balanceCents": 0}, [], {"balanceCents": 1000}, [{"txId": TX}]])
pauses = []
result = verify_case(CASE, lambda *args: next(replies), pauses.append, attempts=2)
self.assertTrue(result["ledger_verified"])
self.assertEqual(pauses, [2])
def test_duplicate_entry_is_failure(self):
def fake(method, url):
return {"balanceCents": 1000} if "/balances/" in url else [{"txId": TX}, {"txId": TX}]
with self.assertRaises(RuntimeError):
verify_case(CASE, fake, attempts=1)
def test_nonlocal_endpoint_is_rejected(self):
with self.assertRaises(ValueError):
request("GET", "https://example.invalid/v1/health")
def test_redirect_is_rejected(self):
with self.assertRaises(ValueError):
NoRedirect().redirect_request(None, None, 302, "Found", {}, "https://example.invalid/")
def test_transport_failure_propagates(self):
def broken(*args):
raise TimeoutError("fixture timeout")
with self.assertRaises(TimeoutError):
verify_case(CASE, broken, attempts=1)
if __name__ == "__main__":
unittest.main()
python3 -m unittest discover -s tests/study -v
python3 scripts/study/probe.py --create --case-file evidence/first-case.json
python3 scripts/study/probe.py --verify evidence/first-case.json
Expected acceptance: the live command exits 0, prints ledger_verified: true, and records one matching entry with a 1,000-cent balance for this fresh account. If it fails, trace Accounts logs/outbox → requested topic → ledger queue → Ledger logs/database. Do not change the expected result to make the exercise pass.
For a small repeated local check after success, run a bounded read-only loop from the repository root:
for run in 1 2 3; do
python3 scripts/study/probe.py --verify evidence/first-case.json || break
sleep 30
done
This is a local API-based synthetic check, not a browser-user simulation or Argo Rollouts canary. More complete user journeys and scheduled runners belong to the later path. Do not execute this probe on a hosted CI runner or point it at cloud services.
Checkpoint and sources
Keep the source revision, image IDs, local endpoints, case identifiers, result, and any failure investigation. Sources: MicroBank routes and entities↗, Python urllib↗, unittest↗.
Your notes and evidence
Record observations, questions, or links to your work. Keep credentials out of your notes.
Back up or restore this path
Progress and notes stay in this browser. A backup contains only this learning path.