Cobb Hayes
c592a58148
cardano-api: strip 'Fix #N:' audit-ticket prefixes from inline comments (was 50+ in main.py), drop hardening-pass changelog blocks from module docstring, rewrite README to drop deploy paths + marketing sections, keep tier/auth/TTL + policy IDs. cardano-checkout-py: drop TradeCraft lineage refs, swap chromaticcraft/tradecraft test fixtures for acme/globex, repository URL → git.sulkta.com.
250 lines
8 KiB
Python
250 lines
8 KiB
Python
"""Monitor integration tests against InMemoryStore.
|
|
|
|
Stubs both Koios (via monkeypatch on ``check_address_utxos``) and the
|
|
ADA/USD oracle so the tests are deterministic and offline. Exercises
|
|
the main state transitions the scheduler cares about:
|
|
|
|
- PENDING → CONFIRMED when UTxOs satisfy the 98% threshold
|
|
- PENDING → OVERPAID when UTxOs exceed the 102% threshold
|
|
- PENDING → UNDERPAID when UTxOs are nonzero but below threshold
|
|
- PENDING (kept) when no UTxOs yet
|
|
- Reprice of an expired invoice → new expected_lovelace + new expires_at
|
|
- Reprice hitting max_repricings → EXPIRED
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from cardano_checkout import InMemoryStore, Invoice, InvoiceStatus
|
|
from cardano_checkout import monitor
|
|
|
|
|
|
def _make(
|
|
id_: str = "inv",
|
|
expected_lovelace: int = 5_000_000,
|
|
expires_in_minutes: int = 15,
|
|
status: InvoiceStatus = InvoiceStatus.PENDING,
|
|
repriced_count: int = 0,
|
|
) -> Invoice:
|
|
now = datetime.now(timezone.utc)
|
|
return Invoice(
|
|
id=id_,
|
|
merchant_id="acme",
|
|
derivation_index=0,
|
|
receive_address="addr1testreceive",
|
|
expected_lovelace=expected_lovelace,
|
|
usd_amount=2.50,
|
|
status=status,
|
|
expires_at=now + timedelta(minutes=expires_in_minutes),
|
|
metadata={"repriced_count": repriced_count} if repriced_count else {},
|
|
)
|
|
|
|
|
|
def _utxo(lovelace: int, tx_hash: str = "aa" * 32) -> dict:
|
|
return {
|
|
"tx_hash": tx_hash,
|
|
"tx_index": 0,
|
|
"value": str(lovelace),
|
|
"asset_list": [],
|
|
}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _patch_koios(monkeypatch):
|
|
"""Default: Koios returns no UTxOs. Individual tests override."""
|
|
|
|
async def fake_utxos(address, koios_url=None, timeout=None):
|
|
return []
|
|
|
|
monkeypatch.setattr(monitor, "check_address_utxos", fake_utxos)
|
|
|
|
|
|
@pytest.fixture
|
|
def price_fn_at_45c():
|
|
"""A deterministic price_fn for tests — USD priced at $0.45/ADA."""
|
|
async def _convert(usd: float) -> int:
|
|
if usd <= 0:
|
|
return 0
|
|
return int((usd / 0.45) * 1_000_000)
|
|
return _convert
|
|
|
|
|
|
@pytest.fixture
|
|
def price_fn_zero():
|
|
"""A price_fn that returns 0 — stand-in for oracle unavailability."""
|
|
async def _zero(usd: float) -> int:
|
|
return 0
|
|
return _zero
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# check_pending_invoices
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_no_utxos_leaves_invoice_pending() -> None:
|
|
store = InMemoryStore()
|
|
await store.create(_make())
|
|
|
|
updated = await monitor.check_pending_invoices(store)
|
|
assert updated == 0
|
|
|
|
inv = await store.get("inv")
|
|
assert inv is not None
|
|
assert inv.status == InvoiceStatus.PENDING
|
|
assert inv.received_lovelace == 0
|
|
|
|
|
|
async def test_confirm_within_tolerance(monkeypatch) -> None:
|
|
store = InMemoryStore()
|
|
await store.create(_make(expected_lovelace=5_000_000))
|
|
|
|
async def fake_utxos(address, koios_url=None, timeout=None):
|
|
# 4.9 ADA — right at the 98% confirm threshold.
|
|
return [_utxo(4_900_000)]
|
|
|
|
monkeypatch.setattr(monitor, "check_address_utxos", fake_utxos)
|
|
|
|
updated = await monitor.check_pending_invoices(store)
|
|
assert updated == 1
|
|
|
|
inv = await store.get("inv")
|
|
assert inv is not None
|
|
assert inv.status == InvoiceStatus.CONFIRMED
|
|
assert inv.received_lovelace == 4_900_000
|
|
assert inv.confirmed_at is not None
|
|
assert inv.tx_hashes == ["aa" * 32]
|
|
|
|
|
|
async def test_overpaid_flag_when_above_threshold(monkeypatch) -> None:
|
|
store = InMemoryStore()
|
|
await store.create(_make(expected_lovelace=5_000_000))
|
|
|
|
async def fake_utxos(address, koios_url=None, timeout=None):
|
|
return [_utxo(5_200_000)]
|
|
|
|
monkeypatch.setattr(monitor, "check_address_utxos", fake_utxos)
|
|
|
|
await monitor.check_pending_invoices(store)
|
|
inv = await store.get("inv")
|
|
assert inv is not None
|
|
assert inv.status == InvoiceStatus.OVERPAID
|
|
assert inv.confirmed_at is not None
|
|
|
|
|
|
async def test_underpaid_when_below_tolerance(monkeypatch) -> None:
|
|
store = InMemoryStore()
|
|
await store.create(_make(expected_lovelace=5_000_000))
|
|
|
|
async def fake_utxos(address, koios_url=None, timeout=None):
|
|
return [_utxo(3_000_000)]
|
|
|
|
monkeypatch.setattr(monitor, "check_address_utxos", fake_utxos)
|
|
|
|
await monitor.check_pending_invoices(store)
|
|
inv = await store.get("inv")
|
|
assert inv is not None
|
|
assert inv.status == InvoiceStatus.UNDERPAID
|
|
assert inv.received_lovelace == 3_000_000
|
|
assert inv.confirmed_at is None
|
|
|
|
|
|
async def test_record_tx_called_for_observed_hashes(monkeypatch) -> None:
|
|
store = InMemoryStore()
|
|
await store.create(_make(expected_lovelace=5_000_000))
|
|
|
|
async def fake_utxos(address, koios_url=None, timeout=None):
|
|
return [_utxo(4_900_000, tx_hash="feed" + "00" * 30)]
|
|
|
|
monkeypatch.setattr(monitor, "check_address_utxos", fake_utxos)
|
|
|
|
await monitor.check_pending_invoices(store)
|
|
|
|
records = store._tx_records()
|
|
assert any(k[0] == "inv" and k[1] == "feed" + "00" * 30 for k in records)
|
|
|
|
|
|
async def test_already_expired_invoices_are_skipped(monkeypatch) -> None:
|
|
"""Invoices past their expiry are left for reprice_expired_invoices to handle."""
|
|
store = InMemoryStore()
|
|
expired = _make(expected_lovelace=5_000_000)
|
|
expired.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1)
|
|
await store.create(expired)
|
|
|
|
calls = []
|
|
|
|
async def fake_utxos(address, koios_url=None, timeout=None):
|
|
calls.append(address)
|
|
return [_utxo(5_000_000)]
|
|
|
|
monkeypatch.setattr(monitor, "check_address_utxos", fake_utxos)
|
|
|
|
await monitor.check_pending_invoices(store)
|
|
# check_pending_invoices must not have polled Koios for an already-expired invoice.
|
|
assert calls == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# reprice_expired_invoices
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_reprice_updates_expected_lovelace_and_extends_expiry(price_fn_at_45c) -> None:
|
|
store = InMemoryStore()
|
|
inv = _make(expected_lovelace=5_000_000)
|
|
inv.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1)
|
|
await store.create(inv)
|
|
|
|
updated = await monitor.reprice_expired_invoices(
|
|
store, price_fn=price_fn_at_45c, window_minutes=15, max_repricings=3
|
|
)
|
|
assert updated == 1
|
|
|
|
fetched = await store.get("inv")
|
|
assert fetched is not None
|
|
# USD $2.50 @ $0.45/ADA = 5.555 ADA = 5555555 lovelace.
|
|
assert fetched.expected_lovelace == 5_555_555
|
|
assert fetched.status == InvoiceStatus.PENDING
|
|
assert fetched.expires_at is not None
|
|
assert fetched.expires_at > datetime.now(timezone.utc)
|
|
assert fetched.metadata["repriced_count"] == 1
|
|
|
|
|
|
async def test_reprice_gives_up_after_max_repricings(price_fn_at_45c) -> None:
|
|
store = InMemoryStore()
|
|
inv = _make(expected_lovelace=5_000_000, repriced_count=3)
|
|
inv.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1)
|
|
await store.create(inv)
|
|
|
|
await monitor.reprice_expired_invoices(
|
|
store, price_fn=price_fn_at_45c, window_minutes=15, max_repricings=3
|
|
)
|
|
|
|
fetched = await store.get("inv")
|
|
assert fetched is not None
|
|
assert fetched.status == InvoiceStatus.EXPIRED
|
|
|
|
|
|
async def test_reprice_noop_when_nothing_expired(price_fn_at_45c) -> None:
|
|
store = InMemoryStore()
|
|
await store.create(_make())
|
|
|
|
updated = await monitor.reprice_expired_invoices(store, price_fn=price_fn_at_45c)
|
|
assert updated == 0
|
|
|
|
|
|
async def test_reprice_skips_when_oracle_returns_zero(price_fn_zero) -> None:
|
|
store = InMemoryStore()
|
|
inv = _make(expected_lovelace=5_000_000)
|
|
inv.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1)
|
|
await store.create(inv)
|
|
|
|
updated = await monitor.reprice_expired_invoices(store, price_fn=price_fn_zero)
|
|
assert updated == 0
|
|
|
|
fetched = await store.get("inv")
|
|
assert fetched is not None
|
|
assert fetched.status == InvoiceStatus.PENDING # not flipped to expired
|