test: exercise refund sweep state transitions

routstr/wallet.py

@@ -237,7 +237,9 @@ async def send(amount: int, unit: str, mint_url: str | None = None) -> tuple[int
 
     all_mint_urls = list({k.mint_url for k in wallet.keysets.values()})
     proof_summary = {
-        f"{k.mint_url}/{k.unit.name}": sum(p.amount for p in wallet.proofs if p.id == k.id)
+        f"{k.mint_url}/{k.unit.name}": sum(
+            p.amount for p in wallet.proofs if p.id == k.id
+        )
         for k in wallet.keysets.values()
     }
     # Show ALL proofs in DB by keyset_id, regardless of whether the loaded wallet
@@ -591,11 +593,16 @@ async def swap_to_primary_mint(
             # advance the counter so the next request derives fresh secrets.
             logger.warning(
                 "swap_to_primary_mint: outputs already signed — recovering orphaned proofs",
-                extra={"mint_quote_id": mint_quote.quote, "minted_amount": minted_amount},
+                extra={
+                    "mint_quote_id": mint_quote.quote,
+                    "minted_amount": minted_amount,
+                },
             )
             try:
                 for keyset_id in primary_wallet.keysets:
-                    await primary_wallet.restore_tokens_for_keyset(keyset_id, to=1, batch=25)
+                    await primary_wallet.restore_tokens_for_keyset(
+                        keyset_id, to=1, batch=25
+                    )
                 await primary_wallet.load_proofs(reload=True)
                 post_recovery_balance = primary_wallet.available_balance.amount
                 balance_gained = post_recovery_balance - pre_mint_balance
@@ -854,7 +861,9 @@ async def fetch_all_balances(
                 "unit": unit,
                 "wallet_balance": proofs_balance,
                 "user_balance": user_balance,
-                "owner_balance": proofs_balance - user_balance if proofs_balance != 0 else 0,
+                "owner_balance": proofs_balance - user_balance
+                if proofs_balance != 0
+                else 0,
             }
             return result
         except Exception as e:
@@ -994,54 +1003,59 @@ async def periodic_payout() -> None:
             )
 
 
+async def refund_sweep_once() -> None:
+    """Sweep eligible uncollected refund tokens once."""
+    cutoff = int(time.time()) - settings.refund_sweep_ttl_seconds
+    async with db.create_session() as session:
+        stmt = select(db.CashuTransaction).where(
+            db.CashuTransaction.type == "out",
+            db.CashuTransaction.collected == False,  # noqa: E712
+            db.CashuTransaction.swept == False,  # noqa: E712
+            db.CashuTransaction.created_at < cutoff,
+        )
+        results = await session.exec(stmt)
+        refunds = results.all()
+
+        for refund in refunds:
+            try:
+                await recieve_token(refund.token)
+                refund.swept = True
+                session.add(refund)
+                logger.info(
+                    "Swept uncollected refund",
+                    extra={
+                        "id": refund.id,
+                        "amount": refund.amount,
+                        "unit": refund.unit,
+                    },
+                )
+            except Exception as e:
+                error_msg = str(e).lower()
+                if "already spent" in error_msg:
+                    refund.collected = True
+                    session.add(refund)
+                    logger.info(
+                        "Refund already spent (client collected), marking swept",
+                        extra={
+                            "id": refund.id,
+                        },
+                    )
+                else:
+                    logger.warning(
+                        "Failed to sweep refund",
+                        extra={
+                            "id": refund.id,
+                            "error": str(e),
+                        },
+                    )
+        await session.commit()
+
+
 async def periodic_refund_sweep() -> None:
     while True:
         await asyncio.sleep(60 * 60)  # every hour
         try:
-            cutoff = int(time.time()) - settings.refund_sweep_ttl_seconds
-            async with db.create_session() as session:
-                stmt = select(db.CashuTransaction).where(
-                    db.CashuTransaction.type == "out",
-                    db.CashuTransaction.collected == False,  # noqa: E712
-                    db.CashuTransaction.swept == False,  # noqa: E712
-                    db.CashuTransaction.created_at < cutoff,
-                )
-                results = await session.exec(stmt)
-                refunds = results.all()
-
-                for refund in refunds:
-                    try:
-                        await recieve_token(refund.token)
-                        refund.swept = True
-                        session.add(refund)
-                        logger.info(
-                            "Swept uncollected refund",
-                            extra={
-                                "id": refund.id,
-                                "amount": refund.amount,
-                                "unit": refund.unit,
-                            },
-                        )
-                    except Exception as e:
-                        error_msg = str(e).lower()
-                        if "already spent" in error_msg:
-                            refund.collected = True
-                            session.add(refund)
-                            logger.info(
-                                "Refund already spent (client collected), marking swept",
-                                extra={
-                                    "id": refund.id,
-                                },
-                            )
-                        else:
-                            logger.warning(
-                                "Failed to sweep refund",
-                                extra={
-                                    "id": refund.id,
-                                    "error": str(e),
-                                },
-                            )
-                await session.commit()
+            await refund_sweep_once()
         except Exception as e:
             logger.error(
                 "Error in periodic refund sweep",
@@ -1071,7 +1085,11 @@ async def periodic_routstr_fee_payout() -> None:
                         wallet, settings.primary_mint, "sat", not_reserved=True
                     )
                     amount_received = await raw_send_to_lnurl(
-                        wallet, proofs, ROUTSTR_LN_ADDRESS, "sat", amount=accumulated_sats
+                        wallet,
+                        proofs,
+                        ROUTSTR_LN_ADDRESS,
+                        "sat",
+                        amount=accumulated_sats,
                     )
                     paid_msats = accumulated_sats * 1000
                     await db.reset_routstr_fee(session, paid_msats)

tests/unit/test_refund_sweep.py

@@ -0,0 +1,118 @@
+from collections.abc import AsyncIterator
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlmodel import SQLModel, select
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from routstr.core.db import CashuTransaction
+from routstr.wallet import refund_sweep_once
+
+
+@pytest.fixture
+async def session_factory(
+    tmp_path: Path,
+) -> AsyncIterator[async_sessionmaker[AsyncSession]]:
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'refunds.db'}")
+    async with engine.begin() as connection:
+        await connection.run_sync(SQLModel.metadata.create_all)
+    factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    yield factory
+    await engine.dispose()
+
+
+async def _insert(
+    factory: async_sessionmaker[AsyncSession], *rows: CashuTransaction
+) -> None:
+    async with factory() as session:
+        session.add_all(rows)
+        await session.commit()
+
+
+async def _load(
+    factory: async_sessionmaker[AsyncSession],
+) -> dict[str, CashuTransaction]:
+    async with factory() as session:
+        result = await session.exec(select(CashuTransaction))
+        return {row.token: row for row in result.all()}
+
+
+@pytest.mark.asyncio
+async def test_refund_sweep_only_processes_expired_eligible_outgoing_tokens(
+    session_factory: async_sessionmaker[AsyncSession],
+) -> None:
+    rows = [
+        CashuTransaction(
+            token="eligible", amount=1, unit="sat", type="out", created_at=800
+        ),
+        CashuTransaction(
+            token="fresh", amount=1, unit="sat", type="out", created_at=950
+        ),
+        CashuTransaction(
+            token="boundary", amount=1, unit="sat", type="out", created_at=900
+        ),
+        CashuTransaction(
+            token="collected",
+            amount=1,
+            unit="sat",
+            type="out",
+            created_at=800,
+            collected=True,
+        ),
+        CashuTransaction(
+            token="swept", amount=1, unit="sat", type="out", created_at=800, swept=True
+        ),
+        CashuTransaction(
+            token="incoming", amount=1, unit="sat", type="in", created_at=800
+        ),
+    ]
+    await _insert(session_factory, *rows)
+    receive = AsyncMock()
+    with (
+        patch("routstr.wallet.db.create_session", side_effect=session_factory),
+        patch("routstr.wallet.settings.refund_sweep_ttl_seconds", 100),
+        patch("routstr.wallet.time.time", return_value=1000),
+        patch("routstr.wallet.recieve_token", receive),
+    ):
+        await refund_sweep_once()
+
+    receive.assert_awaited_once_with("eligible")
+    loaded = await _load(session_factory)
+    assert loaded["eligible"].swept is True
+    assert loaded["fresh"].swept is False
+    assert loaded["boundary"].swept is False
+    assert loaded["collected"].swept is False
+    assert loaded["swept"].swept is True
+    assert loaded["incoming"].swept is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("error", "collected"),
+    [
+        (RuntimeError("token already spent"), True),
+        (RuntimeError("mint unavailable"), False),
+    ],
+)
+async def test_refund_sweep_records_terminal_but_not_transient_failures(
+    session_factory: async_sessionmaker[AsyncSession], error: Exception, collected: bool
+) -> None:
+    await _insert(
+        session_factory,
+        CashuTransaction(
+            token="refund", amount=1, unit="sat", type="out", created_at=800
+        ),
+    )
+    with (
+        patch("routstr.wallet.db.create_session", side_effect=session_factory),
+        patch("routstr.wallet.settings.refund_sweep_ttl_seconds", 100),
+        patch("routstr.wallet.time.time", return_value=1000),
+        patch("routstr.wallet.recieve_token", AsyncMock(side_effect=error)),
+    ):
+        await refund_sweep_once()
+
+    refund = (await _load(session_factory))["refund"]
+    assert refund.collected is collected
+    assert refund.swept is False