feat: add demo measurements to seed and user management to setup page
Seed now generates 180 realistic measurements (30 per subtask) with Gaussian distribution for SPC testing. Setup page gains full user CRUD with list, create/edit, password change, and activate/deactivate. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Adriano
@@ -0,0 +1,294 @@
|
||||
"""Tests for setup router - demo measurements and user management."""
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from config import settings
|
||||
from 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 routers.setup as setup_mod
|
||||
monkeypatch.setattr(setup_mod, "engine", test_engine)
|
||||
monkeypatch.setattr(setup_mod, "async_session_factory", TestSessionFactory)
|
||||
|
||||
|
||||
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 models.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 models.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 models.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):
|
||||
"""Creating a user with a duplicate username should return 409."""
|
||||
await _seed(client)
|
||||
|
||||
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):
|
||||
"""Update an existing user via manage-user endpoint."""
|
||||
await _seed(client)
|
||||
|
||||
# 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):
|
||||
"""Changing password should succeed and invalidate api_key."""
|
||||
await _seed(client)
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user