davide
f27fe6243c
Session hardening: / and /admin now respond with Cache-Control: no-store, and
both pages re-derive their auth state on pageshow (event.persisted) as a
safety net against bfcache showing a stale logged-in/out view across
back/forward navigation. The user page also syncs logout across tabs via the
storage event, since localStorage is shared but in-memory JS state isn't.
Password recovery: admin gets a "Reset" button per user (POST
/admin/users/{id}/reset-password) that generates and sets a new password,
shown once — passwords are Argon2-hashed and can never be recovered, only
replaced. Users get self-service password change (POST
/users/me/change-password, requires the current password) under a new
Profilo tab, alongside read-only account info (username, address, balance,
join date).
Round display robustness: the user dashboard now refreshes immediately on
tab visibility change (background tabs get their timers throttled hard),
shows an explicit "connessione persa" state after repeated failed polls
instead of silently freezing on stale data, and polls faster both right when
the countdown hits zero and through the gap where the round is past its
deadline but still waiting for in-flight bets to confirm before the server
actually closes it.
111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
import pytest
|
|
from cryptography.fernet import Fernet
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.config import settings
|
|
|
|
|
|
@pytest.fixture
|
|
async def client(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(settings, "database_url", f"sqlite+aiosqlite:///{tmp_path}/test.db")
|
|
monkeypatch.setattr(settings, "jwt_secret", "test-jwt-secret")
|
|
monkeypatch.setattr(settings, "xprv_encryption_key", Fernet.generate_key().decode())
|
|
monkeypatch.setattr(settings, "master_key_path", str(tmp_path / "master.xprv.enc"))
|
|
|
|
import app.wallet.hd as hd
|
|
|
|
hd._account_key = None
|
|
hd.generate_master_key()
|
|
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.db import base as db_base
|
|
|
|
# Import app.db.models before create_all — declarative models only register
|
|
# themselves into Base.metadata when their module is first imported, and
|
|
# nothing else in this fixture happens to trigger that import beforehand.
|
|
import app.db.models # noqa: F401
|
|
|
|
db_base.engine = create_async_engine(settings.database_url)
|
|
db_base.AsyncSessionLocal = async_sessionmaker(db_base.engine, expire_on_commit=False)
|
|
|
|
from app.db import session as db_session
|
|
|
|
db_session.AsyncSessionLocal = db_base.AsyncSessionLocal
|
|
|
|
async with db_base.engine.begin() as conn:
|
|
await conn.run_sync(db_base.Base.metadata.create_all)
|
|
|
|
from fastapi import FastAPI
|
|
|
|
from app.api.routes.users import router as users_router
|
|
from app.auth.routes import router as auth_router
|
|
from app.electrum.listener import ElectrumListener
|
|
|
|
app = FastAPI()
|
|
app.include_router(auth_router)
|
|
app.include_router(users_router)
|
|
app.state.electrum_listener = ElectrumListener(lambda: None, db_base.AsyncSessionLocal)
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|
|
|
|
await db_base.engine.dispose()
|
|
|
|
|
|
async def _register(client, username="alice", password="original-password"):
|
|
resp = await client.post("/auth/register", json={"username": username, "password": password})
|
|
assert resp.status_code == 201
|
|
return resp.json()["access_token"]
|
|
|
|
|
|
async def test_change_password_requires_current_password(client):
|
|
token = await _register(client)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
resp = await client.post(
|
|
"/users/me/change-password",
|
|
headers=headers,
|
|
json={"current_password": "wrong-password", "new_password": "brand-new-password"},
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
|
|
async def test_change_password_updates_login(client):
|
|
token = await _register(client)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
resp = await client.post(
|
|
"/users/me/change-password",
|
|
headers=headers,
|
|
json={"current_password": "original-password", "new_password": "brand-new-password"},
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
resp = await client.post("/auth/login", json={"username": "alice", "password": "original-password"})
|
|
assert resp.status_code == 401
|
|
|
|
resp = await client.post("/auth/login", json={"username": "alice", "password": "brand-new-password"})
|
|
assert resp.status_code == 200
|
|
|
|
|
|
async def test_change_password_rejects_too_short(client):
|
|
token = await _register(client)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
resp = await client.post(
|
|
"/users/me/change-password",
|
|
headers=headers,
|
|
json={"current_password": "original-password", "new_password": "short"},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
async def test_change_password_requires_auth(client):
|
|
resp = await client.post(
|
|
"/users/me/change-password",
|
|
json={"current_password": "x", "new_password": "brand-new-password"},
|
|
)
|
|
assert resp.status_code in (401, 403)
|