cardano-checkout-py/tests/test_store_protocol.py

"""InvoiceStore protocol conformance + InMemoryStore round-trips.

These tests exist to catch regressions in the reference implementation
and to document the exact semantics the monitor + scheduler rely on.
Any backend consumers implement should pass the same suite (we don't
parameterize yet — that lands when we ship the SQLAlchemy adapter).
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone

import pytest

from cardano_checkout import InMemoryStore, Invoice, InvoiceStatus, InvoiceStore


def _make_invoice(
    id_: str = "inv_001",
    merchant: str = "acme",
    index: int = 0,
    status: InvoiceStatus = InvoiceStatus.PENDING,
) -> Invoice:
    return Invoice(
        id=id_,
        merchant_id=merchant,
        derivation_index=index,
        receive_address=f"addr1...{index}",
        expected_lovelace=5_000_000,
        usd_amount=2.50,
        status=status,
        expires_at=datetime.now(timezone.utc) + timedelta(minutes=15),
    )


def test_inmemory_store_satisfies_protocol() -> None:
    """Runtime isinstance check against the Protocol — catches API drift."""
    assert isinstance(InMemoryStore(), InvoiceStore)


async def test_create_then_get_round_trips() -> None:
    store = InMemoryStore()
    inv = _make_invoice()

    await store.create(inv)
    fetched = await store.get(inv.id)

    assert fetched is not None
    assert fetched.id == inv.id
    assert fetched.merchant_id == inv.merchant_id
    assert fetched.expected_lovelace == 5_000_000
    assert fetched.status == InvoiceStatus.PENDING


async def test_create_rejects_duplicate_id() -> None:
    store = InMemoryStore()
    inv = _make_invoice()
    await store.create(inv)

    with pytest.raises(ValueError, match="already exists"):
        await store.create(_make_invoice())


async def test_get_missing_returns_none() -> None:
    store = InMemoryStore()
    assert await store.get("does-not-exist") is None


async def test_update_persists_state_changes() -> None:
    store = InMemoryStore()
    inv = _make_invoice()
    await store.create(inv)

    inv.status = InvoiceStatus.CONFIRMED
    inv.received_lovelace = 5_100_000
    inv.confirmed_at = datetime.now(timezone.utc)
    await store.update(inv)

    fetched = await store.get(inv.id)
    assert fetched is not None
    assert fetched.status == InvoiceStatus.CONFIRMED
    assert fetched.received_lovelace == 5_100_000
    assert fetched.confirmed_at is not None


async def test_update_on_missing_raises() -> None:
    store = InMemoryStore()
    with pytest.raises(KeyError):
        await store.update(_make_invoice(id_="never-created"))


async def test_get_is_defensive_copy() -> None:
    """Mutating a fetched invoice must not change stored state."""
    store = InMemoryStore()
    inv = _make_invoice()
    await store.create(inv)

    fetched = await store.get(inv.id)
    assert fetched is not None
    fetched.status = InvoiceStatus.CANCELLED
    fetched.metadata["tampered"] = True

    # Re-fetch — store should still have the original.
    refetched = await store.get(inv.id)
    assert refetched is not None
    assert refetched.status == InvoiceStatus.PENDING
    assert "tampered" not in refetched.metadata


async def test_list_by_status_returns_only_matching() -> None:
    store = InMemoryStore()
    await store.create(_make_invoice(id_="p1", index=0))
    await store.create(_make_invoice(id_="p2", index=1))
    await store.create(
        _make_invoice(id_="c1", index=2, status=InvoiceStatus.CONFIRMED)
    )

    pending = await store.list_by_status(InvoiceStatus.PENDING)
    confirmed = await store.list_by_status(InvoiceStatus.CONFIRMED)
    expired = await store.list_by_status(InvoiceStatus.EXPIRED)

    assert {inv.id for inv in pending} == {"p1", "p2"}
    assert {inv.id for inv in confirmed} == {"c1"}
    assert expired == []


async def test_list_by_status_honours_limit() -> None:
    store = InMemoryStore()
    for i in range(5):
        await store.create(_make_invoice(id_=f"p{i}", index=i))

    results = await store.list_by_status(InvoiceStatus.PENDING, limit=3)
    assert len(results) == 3


async def test_next_derivation_index_is_monotonic_per_merchant() -> None:
    store = InMemoryStore()
    m1 = "acme"
    m2 = "globex"

    assert await store.next_derivation_index(m1) == 0
    assert await store.next_derivation_index(m1) == 1
    assert await store.next_derivation_index(m2) == 0  # independent per merchant
    assert await store.next_derivation_index(m1) == 2


async def test_create_bumps_index_cursor_if_higher() -> None:
    """If a caller creates an invoice at index N, the next derivation MUST skip past it."""
    store = InMemoryStore()
    await store.create(_make_invoice(id_="manual", index=7))

    nxt = await store.next_derivation_index("acme")
    assert nxt == 8


async def test_record_tx_is_idempotent() -> None:
    store = InMemoryStore()
    inv = _make_invoice()
    await store.create(inv)

    await store.record_tx(inv.id, "deadbeef", 5_000_000)
    await store.record_tx(inv.id, "deadbeef", 5_000_000)

    fetched = await store.get(inv.id)
    assert fetched is not None
    # tx_hashes list should contain the hash once, not twice.
    assert fetched.tx_hashes.count("deadbeef") == 1


async def test_record_tx_appends_multiple_distinct_hashes() -> None:
    store = InMemoryStore()
    inv = _make_invoice()
    await store.create(inv)

    await store.record_tx(inv.id, "aaaa", 1_000_000)
    await store.record_tx(inv.id, "bbbb", 2_000_000)

    fetched = await store.get(inv.id)
    assert fetched is not None
    assert set(fetched.tx_hashes) == {"aaaa", "bbbb"}