Adriano/TieMeasureFlow
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases 3 Wiki Activity
Files
accf416cc91c5adaefa30142ccc140d67c0e63c3
TieMeasureFlow/src/backend/tests/test_setup.py
T
Adriano e20299548f fix(migrations): schema base 000 per DB vuoti, stamp head dopo create_all (#6) 
upgrade head partiva dalla 001, che sa solo aggiungere colonne a tabelle
che assume gia' esistenti: su un DB davvero vuoto falliva subito. Nuova
000_base_schema.py (revision "000_base", down_revision None) crea lo
schema pre-001 esplicitamente (create_table, non Base.metadata.create_all,
che porterebbe gia' le colonne di 001-013 e farebbe fallire 001 su
"Duplicate column"). No-op se la tabella recipes esiste gia'
(sa.inspect(bind).has_table("recipes")). 001 ora ha down_revision =
"000_base".

setup.py: init-db e reset-db, dopo create_all/drop_all+create_all,
eseguono l'equivalente di "alembic stamp head" (nuovo _stamp_head(),
command.stamp con la config di src/backend/migrations/alembic.ini) cosi'
un DB nato da create_all non ritenta le migrazioni 001-013 al prossimo
deploy.

Test: catena upgrade head su SQLite vuoto, tabelle/colonne confrontate
con Base.metadata; DB con recipes gia' nello stato pre-001 -> upgrade
head non fallisce (000 no-op). _stamp_head si appoggia a
migrations/env.py, che costruisce una connessione propria da
settings.database_url (non modificabile dai test tramite l'engine
patchato di setup.py): nei test di setup.py e' quindi no-op via
monkeypatch, cosi' init-db/reset-db continuano a girare su SQLite senza
toccare un MySQL vero.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-21 22:49:38 +02:00

461 lines
16 KiB
Python
Raw Blame History

"""Tests for setup router - demo measurements and user management."""
import pytest
from httpx import AsyncClient
from src.backend.config import settings
from src.backend.tests.conftest import test_engine, TestSessionFactory
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
SETUP_PWD = "test-setup-password"
@pytest.fixture(autouse=True)
def enable_setup(monkeypatch):
"""Enable setup endpoints and redirect engine/session to test DB."""
monkeypatch.setattr(settings, "setup_password", SETUP_PWD)
# setup.py imports engine and async_session_factory directly from database
# and uses them instead of get_db dependency injection. Patch them to use
# the test in-memory SQLite engine/session instead of real MySQL.
import src.backend.api.routers.setup as setup_mod
monkeypatch.setattr(setup_mod, "engine", test_engine)
monkeypatch.setattr(setup_mod, "async_session_factory", TestSessionFactory)
# _stamp_head runs `alembic stamp head`, which builds its own connection
# from settings.database_url via migrations/env.py - the real MySQL URL,
# not the patched test engine above. No-op it here so init-db/reset-db
# tests never try to reach a MySQL server that does not exist in CI/dev.
monkeypatch.setattr(setup_mod, "_stamp_head", lambda: None)
async def _seed(client: AsyncClient) -> dict:
"""Run seed and return response data."""
resp = await client.post("/api/setup/init-db", json={"password": SETUP_PWD})
assert resp.status_code == 200
resp = await client.post("/api/setup/seed", json={"password": SETUP_PWD})
assert resp.status_code == 200
return resp.json()
# ---------------------------------------------------------------------------
# Feature 1: Demo Measurements in Seed
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_seed_creates_measurements(client: AsyncClient):
"""Seed should create 180 measurements (30 per subtask x 6 subtasks)."""
data = await _seed(client)
assert data["measurements_created"] == 180
@pytest.mark.asyncio
async def test_seed_measurements_have_correct_distribution(client: AsyncClient, db_session):
"""Measurements should have a mix of pass, warning, and fail."""
await _seed(client)
from sqlalchemy import select, func
from src.backend.models.orm.measurement import Measurement
result = await db_session.execute(
select(Measurement.pass_fail, func.count()).group_by(Measurement.pass_fail)
)
counts = dict(result.all())
# Should have all three statuses
assert "pass" in counts
assert "warning" in counts
assert "fail" in counts
# Pass should be the majority
assert counts["pass"] > counts["warning"]
assert counts["pass"] > counts["fail"]
@pytest.mark.asyncio
async def test_seed_measurements_have_lot_and_serial(client: AsyncClient, db_session):
"""All measurements should have lot and serial numbers."""
await _seed(client)
from sqlalchemy import select, func
from src.backend.models.orm.measurement import Measurement
result = await db_session.execute(
select(func.count()).where(Measurement.lot_number.is_(None))
)
null_lots = result.scalar()
assert null_lots == 0
@pytest.mark.asyncio
async def test_seed_measurements_input_methods(client: AsyncClient, db_session):
"""Measurements should have mixed input methods."""
await _seed(client)
from sqlalchemy import select, func
from src.backend.models.orm.measurement import Measurement
result = await db_session.execute(
select(Measurement.input_method, func.count()).group_by(Measurement.input_method)
)
methods = dict(result.all())
assert "usb_caliper" in methods
assert "manual" in methods
@pytest.mark.asyncio
async def test_seed_is_reproducible(client: AsyncClient):
"""Seed uses random.seed(42), so results should be deterministic."""
data = await _seed(client)
assert data["measurements_created"] == 180
assert data["recipe_created"] == "DEMO-001"
# ---------------------------------------------------------------------------
# Feature 2: User Management - List
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_list_users_after_seed(client: AsyncClient):
"""After seed, list-users should return 4 users."""
await _seed(client)
resp = await client.post("/api/setup/list-users", json={"password": SETUP_PWD})
assert resp.status_code == 200
data = resp.json()
assert len(data["users"]) == 4
@pytest.mark.asyncio
async def test_list_users_wrong_password(client: AsyncClient):
"""list-users with wrong password should return 403."""
resp = await client.post("/api/setup/list-users", json={"password": "wrong"})
assert resp.status_code == 403
# ---------------------------------------------------------------------------
# Feature 2: User Management - Create / Update
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
"""Create a new user via manage-user endpoint."""
resp = await client.post("/api/setup/init-db", json={"password": SETUP_PWD})
assert resp.status_code == 200
resp = await client.post("/api/setup/manage-user", json={
"password": SETUP_PWD,
"user_id": None,
"username": "newuser",
"user_password": "pass123",
"display_name": "New User",
"email": "new@test.com",
"roles": ["Maker"],
"is_admin": False,
})
assert resp.status_code == 200
data = resp.json()
assert data["user_id"] is not None
assert "created" in data["message"]
@pytest.mark.asyncio
async def test_create_user_duplicate_username(client: AsyncClient, monkeypatch):
"""Creating a user with a duplicate username should return 409."""
await _seed(client)
# _seed already created an admin; bypass the "setup already completed"
# guard (#9) so this test can still exercise manage-user's own checks.
monkeypatch.setattr(settings, "setup_allow_after_init", True)
resp = await client.post("/api/setup/manage-user", json={
"password": SETUP_PWD,
"user_id": None,
"username": "admin",
"user_password": "pass123",
"display_name": "Duplicate",
"roles": [],
})
assert resp.status_code == 409
@pytest.mark.asyncio
async def test_create_user_without_password(client: AsyncClient):
"""Creating a user without a password should return 400."""
resp = await client.post("/api/setup/init-db", json={"password": SETUP_PWD})
resp = await client.post("/api/setup/manage-user", json={
"password": SETUP_PWD,
"user_id": None,
"username": "nopassuser",
"display_name": "No Pass",
"roles": [],
})
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_update_user(client: AsyncClient, monkeypatch):
"""Update an existing user via manage-user endpoint."""
await _seed(client)
monkeypatch.setattr(settings, "setup_allow_after_init", True)
# Get users to find the ID
resp = await client.post("/api/setup/list-users", json={"password": SETUP_PWD})
users = resp.json()["users"]
maker = next(u for u in users if u["username"] == "maker1")
resp = await client.post("/api/setup/manage-user", json={
"password": SETUP_PWD,
"user_id": maker["id"],
"username": "maker1",
"display_name": "Updated Name",
"email": "updated@test.com",
"roles": ["Maker", "MeasurementTec"],
"is_admin": False,
})
assert resp.status_code == 200
assert "updated" in resp.json()["message"]
@pytest.mark.asyncio
async def test_create_user_invalid_role(client: AsyncClient):
"""Creating a user with an invalid role should return 400."""
resp = await client.post("/api/setup/init-db", json={"password": SETUP_PWD})
resp = await client.post("/api/setup/manage-user", json={
"password": SETUP_PWD,
"user_id": None,
"username": "badrole",
"user_password": "pass123",
"display_name": "Bad Role",
"roles": ["InvalidRole"],
})
assert resp.status_code == 400
# ---------------------------------------------------------------------------
# Feature 2: User Management - Password Change
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_change_password(client: AsyncClient, monkeypatch):
"""Changing password should succeed and invalidate api_key."""
await _seed(client)
monkeypatch.setattr(settings, "setup_allow_after_init", True)
resp = await client.post("/api/setup/list-users", json={"password": SETUP_PWD})
users = resp.json()["users"]
tec = next(u for u in users if u["username"] == "tec1")
resp = await client.post("/api/setup/change-user-password", json={
"password": SETUP_PWD,
"user_id": tec["id"],
"new_password": "newpass123",
})
assert resp.status_code == 200
assert "changed" in resp.json()["message"].lower()
@pytest.mark.asyncio
async def test_change_password_user_not_found(client: AsyncClient):
"""Changing password for non-existent user should return 404."""
resp = await client.post("/api/setup/init-db", json={"password": SETUP_PWD})
resp = await client.post("/api/setup/change-user-password", json={
"password": SETUP_PWD,
"user_id": 99999,
"new_password": "newpass123",
})
assert resp.status_code == 404
# ---------------------------------------------------------------------------
# Feature 2: User Management - Toggle Active
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_toggle_user_active(client: AsyncClient):
"""Toggle active should flip the user's active status."""
await _seed(client)
resp = await client.post("/api/setup/list-users", json={"password": SETUP_PWD})
users = resp.json()["users"]
maker = next(u for u in users if u["username"] == "maker1")
assert maker["active"] is True
# Deactivate
resp = await client.post("/api/setup/toggle-user-active", json={
"password": SETUP_PWD,
"user_id": maker["id"],
})
assert resp.status_code == 200
assert resp.json()["active"] is False
# Re-activate
resp = await client.post("/api/setup/toggle-user-active", json={
"password": SETUP_PWD,
"user_id": maker["id"],
})
assert resp.status_code == 200
assert resp.json()["active"] is True
# ---------------------------------------------------------------------------
# Feature 3: _check_password is timing-safe
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_wrong_password_still_rejected_with_compare_digest(client: AsyncClient):
"""hmac.compare_digest must not change the pass/fail outcome."""
resp = await client.post("/api/setup/list-users", json={"password": "definitely-wrong"})
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_correct_password_still_accepted_with_compare_digest(client: AsyncClient):
resp = await client.post("/api/setup/init-db", json={"password": SETUP_PWD})
assert resp.status_code == 200
# ---------------------------------------------------------------------------
# Feature 4: setup already completed - privileged endpoints blocked (#9)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_privileged_endpoints_blocked_once_admin_exists(client: AsyncClient, monkeypatch):
"""reset-db, seed, manage-user, change-user-password, init-db all 403
once an admin exists, unless setup_allow_after_init is set."""
# This test makes more than rate_limit_login (5/min) requests to
# /api/setup/* on purpose, to exercise every guarded endpoint - raise the
# limit so it is the #9 guard being tested, not the #8/#9(d) rate limit.
monkeypatch.setattr(settings, "rate_limit_login", 100)
data = await _seed(client) # creates the "admin" user (is_admin=True)
admin_id = None
resp = await client.post("/api/setup/list-users", json={"password": SETUP_PWD})
for u in resp.json()["users"]:
if u["username"] == "admin":
admin_id = u["id"]
assert admin_id is not None
resp = await client.post("/api/setup/init-db", json={"password": SETUP_PWD})
assert resp.status_code == 403
assert "already completed" in resp.json()["detail"]
resp = await client.post("/api/setup/seed", json={"password": SETUP_PWD})
assert resp.status_code == 403
resp = await client.post("/api/setup/manage-user", json={
"password": SETUP_PWD,
"user_id": None,
"username": "blocked",
"user_password": "pass123",
"display_name": "Blocked",
"roles": [],
})
assert resp.status_code == 403
resp = await client.post("/api/setup/change-user-password", json={
"password": SETUP_PWD,
"user_id": admin_id,
"new_password": "whatever123",
})
assert resp.status_code == 403
resp = await client.post("/api/setup/reset-db", json={
"password": SETUP_PWD,
"confirm": True,
})
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_privileged_endpoints_allowed_before_any_admin(client: AsyncClient):
"""On a fresh setup (no admin yet), init-db and seed still work."""
resp = await client.post("/api/setup/init-db", json={"password": SETUP_PWD})
assert resp.status_code == 200
resp = await client.post("/api/setup/seed", json={"password": SETUP_PWD})
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_setup_allow_after_init_overrides_guard(client: AsyncClient, monkeypatch):
"""SETUP_ALLOW_AFTER_INIT=true restores the old (pre-#9) behaviour."""
await _seed(client)
monkeypatch.setattr(settings, "setup_allow_after_init", True)
resp = await client.post("/api/setup/seed", json={"password": SETUP_PWD})
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_toggle_user_active_not_guarded(client: AsyncClient):
"""toggle-user-active is not in the guarded list, admin or not."""
await _seed(client)
resp = await client.post("/api/setup/list-users", json={"password": SETUP_PWD})
maker = next(u for u in resp.json()["users"] if u["username"] == "maker1")
resp = await client.post("/api/setup/toggle-user-active", json={
"password": SETUP_PWD,
"user_id": maker["id"],
})
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_create_admin_not_guarded(client: AsyncClient):
"""create-admin is not in the guarded list either (see analysis of #9)."""
await _seed(client)
resp = await client.post("/api/setup/create-admin", json={
"password": SETUP_PWD,
"admin_username": "second_admin",
"admin_password": "pass12345",
"admin_email": "second@test.com",
"admin_display_name": "Second Admin",
})
assert resp.status_code == 200
# ---------------------------------------------------------------------------
# Feature 5: GET /api/setup/status without/with password (#9)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_status_without_password_is_minimal(client: AsyncClient):
resp = await client.get("/api/setup/status")
assert resp.status_code == 200
data = resp.json()
assert data == {"initialised": False}
@pytest.mark.asyncio
async def test_status_with_wrong_password_is_minimal(client: AsyncClient):
resp = await client.get("/api/setup/status", params={"password": "nope"})
assert resp.status_code == 200
assert resp.json() == {"initialised": False}
@pytest.mark.asyncio
async def test_status_with_valid_password_lists_tables(client: AsyncClient):
await client.post("/api/setup/init-db", json={"password": SETUP_PWD})
resp = await client.get("/api/setup/status", params={"password": SETUP_PWD})
assert resp.status_code == 200
data = resp.json()
assert "tables" in data
assert "table_count" in data
assert data["tables_exist"] is True
@pytest.mark.asyncio
async def test_status_initialised_true_after_seed(client: AsyncClient):
await _seed(client)
resp = await client.get("/api/setup/status")
assert resp.status_code == 200
assert resp.json() == {"initialised": True}
Reference in New Issue View Git Blame Copy Permalink