"""PIN a 4 cifre: login rapido, blocco, capoturno via PIN (#59).""" from datetime import timedelta import pytest from fastapi import HTTPException from httpx import AsyncClient from sqlalchemy import select from src.backend.config import settings from src.backend.models.orm.user import User from src.backend.services import auth_service from src.backend.tests.conftest import _create_user, auth_headers from src.backend.tests.test_production_runs import ( _open, _recipe_with_interval, _station, ) from src.backend.utils.tempo import utcnow @pytest.fixture(autouse=True) def _alza_limite_login(monkeypatch): # #65: login-pin conta nel bucket del login (limite al minuto); i test del blocco PIN # fanno piu tentativi di fila, quindi qui il limite si alza. Il middleware resta com'e. monkeypatch.setattr(settings, "rate_limit_login", 1000) async def _pin_user(db_session, username="operaio", pin="1234", **kw) -> User: user = await _create_user(db_session, username=username, **kw) await auth_service.set_user_pin(db_session, user, pin) await db_session.commit() return user async def _login_pin(client, user_id, pin): return await client.post("/api/auth/login-pin", json={"user_id": user_id, "pin": pin}) async def test_login_pin_ok_same_shape_as_login(client: AsyncClient, db_session): user = await _pin_user(db_session) resp = await _login_pin(client, user.id, "1234") assert resp.status_code == 200 data = resp.json() assert data["user"]["username"] == "operaio" assert data["user"]["has_pin"] is True assert data["api_key"] assert "pin_hash" not in data["user"] # la api_key vale davvero me = await client.get("/api/auth/me", headers={"X-API-Key": data["api_key"]}) assert me.status_code == 200 async def test_login_pin_wrong_is_401(client: AsyncClient, db_session): user = await _pin_user(db_session) assert (await _login_pin(client, user.id, "9999")).status_code == 401 async def test_login_pin_format_422(client: AsyncClient, db_session): user = await _pin_user(db_session) for bad in ("123", "12345", "12a4", ""): assert (await _login_pin(client, user.id, bad)).status_code == 422 async def test_lock_after_five_errors_even_with_right_pin(client: AsyncClient, db_session): user = await _pin_user(db_session) for _ in range(4): assert (await _login_pin(client, user.id, "0000")).status_code == 401 resp = await _login_pin(client, user.id, "0000") # 5 errore: scatta il blocco assert resp.status_code == 423 assert "locked until" in resp.json()["detail"] assert (await _login_pin(client, user.id, "1234")).status_code == 423 async def test_lock_expires_and_counter_resets(client: AsyncClient, db_session): user = await _pin_user(db_session) for _ in range(5): await _login_pin(client, user.id, "0000") row = (await db_session.execute(select(User).where(User.id == user.id))).scalar_one() await db_session.refresh(row) assert row.pin_locked_until > utcnow() + timedelta(minutes=14) # tempo simulato: il blocco e scaduto un secondo fa row.pin_locked_until = utcnow() - timedelta(seconds=1) await db_session.commit() assert (await _login_pin(client, user.id, "1234")).status_code == 200 await db_session.refresh(row) assert row.pin_failed_count == 0 and row.pin_locked_until is None async def test_success_resets_failed_counter(client: AsyncClient, db_session): user = await _pin_user(db_session) for _ in range(3): await _login_pin(client, user.id, "0000") assert (await _login_pin(client, user.id, "1234")).status_code == 200 for _ in range(4): # il contatore e ripartito da zero: 4 errori non bloccano assert (await _login_pin(client, user.id, "0000")).status_code == 401 async def test_admin_cannot_have_pin_or_login_with_it(client: AsyncClient, admin_user, db_session): resp = await client.put( f"/api/users/{admin_user.id}/pin", headers=auth_headers(admin_user), json={"pin": "1234"}, ) assert resp.status_code == 422 # anche un admin con pin_hash scritto a mano non entra da login-pin admin_user.pin_hash = auth_service.hash_password("1234") await db_session.commit() assert (await _login_pin(client, admin_user.id, "1234")).status_code == 401 listed = (await client.get("/api/auth/pin-users")).json() assert all(u["id"] != admin_user.id for u in listed) async def test_put_pin_set_and_remove(client: AsyncClient, admin_user, db_session): user = await _create_user(db_session, username="senzapin") await db_session.commit() h = auth_headers(admin_user) assert (await client.put(f"/api/users/{user.id}/pin", headers=h, json={"pin": "4321"})).status_code == 200 assert (await _login_pin(client, user.id, "4321")).status_code == 200 assert (await client.put(f"/api/users/{user.id}/pin", headers=h, json={"pin": "12"})).status_code == 422 assert (await client.put(f"/api/users/{user.id}/pin", headers=h, json={"pin": None})).status_code == 200 assert (await _login_pin(client, user.id, "4321")).status_code == 401 assert (await client.put("/api/users/99999/pin", headers=h, json={"pin": "1234"})).status_code == 404 # non admin: 403 assert (await client.put( f"/api/users/{user.id}/pin", headers=auth_headers(user), json={"pin": "1234"}, )).status_code == 403 async def test_pin_users_list(client: AsyncClient, db_session): a = await _pin_user(db_session, username="zeta", display_name="Zeta Rossi", pin="1111") b = await _pin_user(db_session, username="alfa", display_name="Alfa Bianchi", pin="2222") await _create_user(db_session, username="nopin") await _pin_user(db_session, username="inattivo", pin="3333", active=False) await db_session.commit() resp = await client.get("/api/auth/pin-users") # senza autenticazione assert resp.status_code == 200 data = resp.json() assert [u["display_name"] for u in data] == ["Alfa Bianchi", "Zeta Rossi"] assert all(set(u) == {"id", "display_name"} for u in data) assert {u["id"] for u in data} == {a.id, b.id} async def test_supervisor_via_pin(client: AsyncClient, measurement_tec_user, admin_user, db_session): station = await _station(db_session, admin_user.id, code="ST-PIN") recipe = await _recipe_with_interval(db_session, admin_user.id, 10, code="REC-PIN") run_id = (await _open(client, measurement_tec_user, station, recipe)).json()["id"] capo = await _pin_user(db_session, username="capo-pin", pin="5555", roles=["Supervisor"]) h = auth_headers(measurement_tec_user) bad = await client.post( f"/api/production-runs/{run_id}/pause", headers=h, json={"supervisor_user_id": capo.id, "supervisor_pin": "0000"}, ) assert bad.status_code == 401 ok = await client.post( f"/api/production-runs/{run_id}/pause", headers=h, json={"supervisor_user_id": capo.id, "supervisor_pin": "5555"}, ) assert ok.status_code == 200 # nessuna coppia di credenziali: 422 none = await client.post(f"/api/production-runs/{run_id}/resume", headers=h, json={}) assert none.status_code == 422 async def test_supervisor_pin_role_still_checked(db_session): plain = await _pin_user(db_session, username="non-capo", pin="6666", roles=["Maker"]) with pytest.raises(HTTPException) as e: await auth_service.authorise_supervisor(db_session, user_id=plain.id, pin="6666") assert e.value.status_code == 403 async def test_supervisor_pin_lockout(db_session): capo = await _pin_user(db_session, username="capo-lock", pin="7777", roles=["Supervisor"]) codes = [] for pin in ["0000"] * 5 + ["7777"]: with pytest.raises(HTTPException) as e: await auth_service.authorise_supervisor(db_session, user_id=capo.id, pin=pin) codes.append(e.value.status_code) assert codes == [401, 401, 401, 401, 423, 423]