56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
|
"""B-41: own_address_for is the single place tx/confirmation.py and
|
||
|
|
tx/reconcile.py derive a PendingTransaction's own address from — a payout's
|
||
|
|
address must always be the pool's, everything else the actual user's."""
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||
|
|
|
||
|
|
from app.config import settings
|
||
|
|
from app.db.base import Base
|
||
|
|
from app.db.models import User
|
||
|
|
from app.tx.pending_address import own_address_for
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
async def session_factory(tmp_path, monkeypatch):
|
||
|
|
monkeypatch.setattr(settings, "master_key_path", str(tmp_path / "master.xprv.enc"))
|
||
|
|
monkeypatch.setattr(
|
||
|
|
settings,
|
||
|
|
"xprv_encryption_key",
|
||
|
|
__import__("cryptography.fernet", fromlist=["Fernet"]).Fernet.generate_key().decode(),
|
||
|
|
)
|
||
|
|
from app.wallet import hd
|
||
|
|
|
||
|
|
hd._account_key = None
|
||
|
|
hd.generate_master_key()
|
||
|
|
|
||
|
|
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()
|
||
|
|
hd._account_key = None
|
||
|
|
|
||
|
|
|
||
|
|
async def test_payout_uses_the_pool_address_regardless_of_user_id(session_factory):
|
||
|
|
from app.wallet.hd import derive_pool_address
|
||
|
|
|
||
|
|
async with session_factory() as session:
|
||
|
|
address = await own_address_for(session, "payout", None)
|
||
|
|
|
||
|
|
assert address == derive_pool_address()
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("kind", ["bet", "withdrawal"])
|
||
|
|
async def test_bet_and_withdrawal_use_the_users_own_address(session_factory, kind):
|
||
|
|
from app.wallet.hd import derive_user_address
|
||
|
|
|
||
|
|
async with session_factory() as session:
|
||
|
|
user = User(username="alice", password_hash="x", derivation_index=3, address=derive_user_address(3))
|
||
|
|
session.add(user)
|
||
|
|
await session.flush()
|
||
|
|
|
||
|
|
address = await own_address_for(session, kind, user.id)
|
||
|
|
|
||
|
|
assert address == derive_user_address(3)
|