davideandClaude Sonnet 5
f1a1145cda
credit_confirmed_utxos only ever credited new UTXOs; a UTXO spent by something other than the app's own bet/withdrawal/payout flow (e.g. someone using the raw derived privkey directly) never got its spent_txid set, so cached_balance_sats kept counting it forever. detect_external_spends mirrors the same listunspent refresh in the other direction: anything still marked unspent in our DB but missing from the address's current unspent set gets spent_txid="external-spend", an audit_log entry, and an immediate balance recompute — wired into the same ElectrumListener._refresh_user call that already runs on every scripthash notification and on listener (re)connect. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
86 lines
3.5 KiB
Python
86 lines
3.5 KiB
Python
import pytest
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.db.base import Base
|
|
from app.db.models import AuditLog, User, UtxoEvent
|
|
from app.deposits.service import credit_confirmed_utxos, detect_external_spends
|
|
|
|
|
|
@pytest.fixture
|
|
async def session_factory():
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield async_sessionmaker(engine, expire_on_commit=False)
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
async def user_id(session_factory):
|
|
async with session_factory() as session:
|
|
user = User(username="alice", password_hash="x", derivation_index=0, address="plm1qxxx")
|
|
session.add(user)
|
|
await session.commit()
|
|
return user.id
|
|
|
|
|
|
async def test_credits_confirmed_utxo_and_updates_balance(session_factory, user_id):
|
|
entries = [{"tx_hash": "aa" * 32, "tx_pos": 0, "height": 100, "value": 10_000_000}]
|
|
async with session_factory() as session:
|
|
credited = await credit_confirmed_utxos(session, user_id, entries)
|
|
assert credited == 1
|
|
user = await session.get(User, user_id)
|
|
assert user.cached_balance_sats == 10_000_000
|
|
|
|
|
|
async def test_unconfirmed_entry_is_ignored(session_factory, user_id):
|
|
entries = [{"tx_hash": "bb" * 32, "tx_pos": 0, "height": 0, "value": 5_000_000}]
|
|
async with session_factory() as session:
|
|
credited = await credit_confirmed_utxos(session, user_id, entries)
|
|
assert credited == 0
|
|
user = await session.get(User, user_id)
|
|
assert user.cached_balance_sats == 0
|
|
|
|
|
|
async def test_idempotent_on_repeated_notification(session_factory, user_id):
|
|
entries = [{"tx_hash": "cc" * 32, "tx_pos": 0, "height": 100, "value": 7_000_000}]
|
|
async with session_factory() as session:
|
|
first = await credit_confirmed_utxos(session, user_id, entries)
|
|
async with session_factory() as session:
|
|
second = await credit_confirmed_utxos(session, user_id, entries)
|
|
user = await session.get(User, user_id)
|
|
assert first == 1
|
|
assert second == 0
|
|
assert user.cached_balance_sats == 7_000_000
|
|
|
|
|
|
async def test_external_spend_marks_utxo_spent_and_corrects_balance(session_factory, user_id):
|
|
entries = [{"tx_hash": "dd" * 32, "tx_pos": 0, "height": 100, "value": 20_000_000}]
|
|
async with session_factory() as session:
|
|
await credit_confirmed_utxos(session, user_id, entries)
|
|
|
|
async with session_factory() as session:
|
|
spent = await detect_external_spends(session, user_id, [])
|
|
assert spent == 1
|
|
user = await session.get(User, user_id)
|
|
assert user.cached_balance_sats == 0
|
|
|
|
utxo = (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one()
|
|
assert utxo.spent_txid == "external-spend"
|
|
|
|
audit_events = (await session.scalars(select(AuditLog))).all()
|
|
assert any(e.event_type == "utxo_spent_externally" for e in audit_events)
|
|
|
|
|
|
async def test_no_spend_detected_when_utxo_still_unspent(session_factory, user_id):
|
|
entries = [{"tx_hash": "ee" * 32, "tx_pos": 0, "height": 100, "value": 3_000_000}]
|
|
async with session_factory() as session:
|
|
await credit_confirmed_utxos(session, user_id, entries)
|
|
|
|
async with session_factory() as session:
|
|
spent = await detect_external_spends(session, user_id, entries)
|
|
assert spent == 0
|
|
user = await session.get(User, user_id)
|
|
assert user.cached_balance_sats == 3_000_000
|