PyBook 01: Foundation

What this notebook proves: "pytest green on a fresh clone with no API keys", and the acceptance rule: "Must pass on a fresh clone with no API keys, cassettes committed."

This book walks the foundation layer: settings, the Clock port, the exception taxonomy, and the cassette replay harness. Every cell runs offline against the installed auradefi package and asserts its own output. A notebook is a test.

import auradefi

print("auradefi", auradefi.__version__)
assert auradefi.__version__ == "0.1.1"

auradefi 0.1.1

Settings: no keys ever needed

The design principle: "an embedding host owns its own configuration story", so Settings is a frozen dataclass the host constructs directly, plus an AURADEFI_-prefixed environment loader for convenience. No dotenv magic, no framework settings object, and no default requires any variable to be set.

from auradefi.config import Settings

# Constructed bare: every field has a working default, no key required.
settings = Settings()
assert settings.etherscan_api_key is None
assert settings.helius_api_key is None
assert settings.http_timeout_s == 10.0
assert settings.sync_min_interval_s == 60

# from_env takes an explicit mapping (os.environ is only the default),
# so this cell depends on nothing outside itself.
assert Settings.from_env({}) == Settings()
loaded = Settings.from_env({
    "AURADEFI_ETHERSCAN_API_KEY": "k_demo",
    "AURADEFI_HTTP_TIMEOUT_S": "2.5",
})
assert loaded.etherscan_api_key == "k_demo"
assert loaded.http_timeout_s == 2.5
print("defaults need no environment; overrides load from AURADEFI_*")

defaults need no environment; overrides load from AURADEFI_*

from auradefi.errors import ConfigError

# Invalid configuration fails at construction, inside the taxonomy.
for build in (
    lambda: Settings(http_timeout_s=0.0),
    lambda: Settings(sync_min_interval_s=-1),
    lambda: Settings.from_env({"AURADEFI_HTTP_TIMEOUT_S": "fast"}),
):
    try:
        build()
    except ConfigError as exc:
        print("rejected:", exc)
    else:
        raise AssertionError("expected ConfigError")

rejected: http_timeout_s must be positive: 0.0
rejected: sync_min_interval_s must be non-negative: -1
rejected: AURADEFI_HTTP_TIMEOUT_S must be a float: 'fast'

The Clock port: "ms epoch, everywhere, always"

Every timestamp in auradefi is an integer of milliseconds since the Unix epoch (initiated_at: int # ms epoch, everywhere, always). Code that needs the current time takes a Clock, a runtime_checkable Protocol, so production injects SystemClock and tests inject FrozenClock, which moves only when advance() is called. That is what makes every time-dependent test deterministic.

from auradefi.clock import Clock, FrozenClock, SystemClock

# Both implementations satisfy the port structurally, no base class.
assert isinstance(SystemClock(), Clock)
assert isinstance(FrozenClock(0), Clock)

clock = FrozenClock(1_754_000_000_000)
assert clock.now_ms() == clock.now_ms() == 1_754_000_000_000  # frozen
clock.advance(86_400_000)  # one day, in ms
assert clock.now_ms() == 1_754_086_400_000

# FrozenClock only advances forward.
try:
    clock.advance(-1)
except ValueError as exc:
    print("guarded:", exc)
else:
    raise AssertionError("expected ValueError")

assert isinstance(SystemClock().now_ms(), int)  # always integer ms
print("frozen clock at", clock.now_ms(), "ms epoch")

guarded: FrozenClock only advances forward
frozen clock at 1754086400000 ms epoch

The exception taxonomy: one base class

Everything auradefi raises derives from a single AuradefiError, "so an embedding host can catch one type at the boundary" (auradefi/errors.py). Exception classes are defined in that one module and nowhere else: the test suite enforces it mechanically, so a new error type is a deliberate, reviewed addition to the public contract.

import inspect

from auradefi import errors

classes = [
    obj
    for obj in vars(errors).values()
    if inspect.isclass(obj) and issubclass(obj, Exception)
]
assert errors.AuradefiError in classes
for cls in classes:
    assert issubclass(cls, errors.AuradefiError), cls
print(len(classes), "exception classes, every one under AuradefiError")

# Subtrees mean a host can be precise or broad, as it likes:
#   CassetteMissError -> CassetteError -> AuradefiError
assert issubclass(errors.CassetteMissError, errors.CassetteError)
assert issubclass(errors.CaipParseError, errors.ValidationError)
try:
    raise errors.CaipParseError("demo")
except errors.AuradefiError as exc:  # the one boundary catch
    print("caught at the boundary:", type(exc).__name__)

23 exception classes, every one under AuradefiError
caught at the boundary: CaipParseError

Cassette replay: recorded HTTP, committed

Again: cassettes are committed JSON recordings so the suite (and any embedding host's tests) run with zero network. A cassette is:

{"interactions": [
    {"request":  {"method": "GET", "url": "https://…?a=1&b=2"},
     "response": {"status": 200, "json": {"ok": true}}}
]}

Matching is by method + host + path + sorted query string, so parameter order never matters. Below we build a small cassette in a temp file and replay it through a real httpx.Client.

import json
import tempfile
from pathlib import Path

from auradefi.testing import cassettes

document = {
    "interactions": [
        {
            "request": {
                "method": "GET",
                "url": "https://api.example.com/v1/balances?address=0xabc&chain=eip155:1",
            },
            "response": {
                "status": 200,
                "headers": {"content-type": "application/json"},
                "json": {"balance": "1234567890123456789", "decimals": 18},
            },
        },
        {
            "request": {"method": "GET", "url": "https://api.example.com/v1/status"},
            "response": {"status": 200, "json": {"state": "pending"}},
        },
        {
            "request": {"method": "GET", "url": "https://api.example.com/v1/status"},
            "response": {"status": 200, "json": {"state": "confirmed"}},
        },
    ]
}
cassette_path = Path(tempfile.mkdtemp()) / "demo_cassette.json"
cassette_path.write_text(json.dumps(document, indent=2), encoding="utf-8")

cassette = cassettes.load(cassette_path)
client = cassette.client()

# Query params in a DIFFERENT order than recorded: still a hit.
response = client.get("https://api.example.com/v1/balances?chain=eip155:1&address=0xabc")
assert response.status_code == 200
body = response.json()
assert body == {"balance": "1234567890123456789", "decimals": 18}
assert isinstance(body["balance"], str)  # rule #2, even in fixtures
print("replayed:", body)

replayed: {'balance': '1234567890123456789', 'decimals': 18}

Repeated identical requests replay their recorded interactions in order, and the final one repeats, so idempotent polling loops work against a finite recording.

states = [client.get("https://api.example.com/v1/status").json()["state"] for _ in range(4)]
assert states == ["pending", "confirmed", "confirmed", "confirmed"]
print("poll sequence:", states)

poll sequence: ['pending', 'confirmed', 'confirmed', 'confirmed']

The offline guarantee: `CassetteMissError`

Any request with no recorded match raises CassetteMissError. "The offline guarantee fails loudly, never by letting a live call escape" (auradefi/testing/cassettes.py). A test that would have touched the network fails instead, and its message lists what is recorded.

from auradefi.errors import AuradefiError, CassetteMissError

try:
    client.get("https://api.example.com/v1/unrecorded")
except CassetteMissError as exc:
    assert "v1/unrecorded" in str(exc)
    assert "Recorded interactions" in str(exc)
    assert isinstance(exc, AuradefiError)  # inside the one taxonomy
    print(exc)
else:
    raise AssertionError("a live call escaped: the offline guarantee is broken")

client.close()
print("\nfoundation walkthrough complete: every assertion held")

GET https://api.example.com/v1/unrecorded is not recorded in demo_cassette.json. Recorded interactions:
  GET api.example.com/v1/balances
  GET api.example.com/v1/status

foundation walkthrough complete: every assertion held