fix(users): repair create delete and duplicate email handling
This commit is contained in:
Schubert Ferenc
parent
155fdbb16a
commit
0bbcaba211
22 changed files with 773 additions and 107 deletions
Binary file not shown.
|
|
@ -1,9 +1,11 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import create_engine, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
|
|
@ -19,8 +21,9 @@ from app.services.validation_workflow import ValidationWorkflowService
|
|||
from app.services.auth_service import AuthService
|
||||
from app.services.reference_masterdata import ReferenceMasterdataImportService
|
||||
from app.api.v1.auth import login as auth_login
|
||||
from app.api.v1 import domain as domain_api
|
||||
from app.schemas.auth import LoginRequest
|
||||
from app.schemas.domain import ValidationCreate
|
||||
from app.schemas.domain import UserCreate, UserUpdate, ValidationCreate
|
||||
from app.modules.orion.service import OrionReportService
|
||||
from app.modules.orion.assets import SCHUBAMED_LOGO_PATH, schubamed_logo_uri
|
||||
from app.modules.orion.template_service import REPORT_SECTIONS, ReportTemplateService
|
||||
|
|
@ -212,6 +215,319 @@ def test_user_model_supports_password_management_fields():
|
|||
assert user.role == UserRole.PRUEFER
|
||||
|
||||
|
||||
def test_admin_appears_in_user_list():
|
||||
db = session()
|
||||
admin = User(
|
||||
email="admin@schubamed.de",
|
||||
first_name="Admin",
|
||||
last_name="User",
|
||||
role=UserRole.ADMIN.value,
|
||||
password_hash="hash",
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
)
|
||||
db.add(admin)
|
||||
db.flush()
|
||||
|
||||
result = domain_api.list_users(session=db, _=admin, page=1, page_size=20, search=None, role=None, active=None, sort_by="created_at", sort_order="desc")
|
||||
|
||||
assert result["total"] == 1
|
||||
assert result["pages"] == 1
|
||||
assert any(item.id == admin.id for item in result["items"])
|
||||
|
||||
|
||||
def test_user_list_pagination_and_filters_work():
|
||||
db = session()
|
||||
admin = User(
|
||||
email="admin@schubamed.de",
|
||||
first_name="Admin",
|
||||
last_name="User",
|
||||
role=UserRole.ADMIN.value,
|
||||
password_hash="hash",
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
)
|
||||
inactive = User(
|
||||
email="inactive@schubamed.de",
|
||||
first_name="Inactive",
|
||||
last_name="User",
|
||||
role=UserRole.LESER.value,
|
||||
password_hash="hash",
|
||||
is_active=False,
|
||||
must_change_password=False,
|
||||
)
|
||||
db.add_all([admin, inactive])
|
||||
db.flush()
|
||||
|
||||
result = domain_api.list_users(session=db, _=admin, page=1, page_size=1, search=None, role=None, active=None, sort_by="created_at", sort_order="desc")
|
||||
assert result["total"] == 2
|
||||
assert result["pages"] == 2
|
||||
assert len(result["items"]) == 1
|
||||
|
||||
active_only = domain_api.list_users(session=db, _=admin, page=1, page_size=20, search=None, role=None, active=True, sort_by="created_at", sort_order="desc")
|
||||
assert active_only["total"] == 1
|
||||
assert all(item.is_active for item in active_only["items"])
|
||||
|
||||
inactive_only = domain_api.list_users(session=db, _=admin, page=1, page_size=20, search=None, role=None, active=False, sort_by="created_at", sort_order="desc")
|
||||
assert inactive_only["total"] == 1
|
||||
assert all(not item.is_active for item in inactive_only["items"])
|
||||
|
||||
role_filtered = domain_api.list_users(session=db, _=admin, page=1, page_size=20, search=None, role=UserRole.ADMIN.value, active=None, sort_by="created_at", sort_order="desc")
|
||||
assert role_filtered["total"] == 1
|
||||
assert role_filtered["items"][0].role == UserRole.ADMIN.value
|
||||
|
||||
|
||||
def test_admin_can_create_users_with_supported_roles():
|
||||
db = session()
|
||||
admin = User(
|
||||
email="admin@schubamed.de",
|
||||
first_name="Admin",
|
||||
last_name="User",
|
||||
role=UserRole.ADMIN.value,
|
||||
password_hash="hash",
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
)
|
||||
db.add(admin)
|
||||
db.flush()
|
||||
|
||||
for role in (UserRole.MITARBEITER, UserRole.PRUEFER, UserRole.LESER):
|
||||
response = domain_api.create_user(
|
||||
UserCreate(
|
||||
email=f"{role.value}@schubamed.de",
|
||||
first_name="Test",
|
||||
last_name="User",
|
||||
role=role,
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
temporary_password="TempPassword123!",
|
||||
),
|
||||
session=db,
|
||||
_=admin,
|
||||
)
|
||||
payload = json.loads(response.body)
|
||||
assert response.status_code == 201
|
||||
assert payload["user"]["role"] == role.value
|
||||
assert db.scalar(select(User).where(User.email == f"{role.value}@schubamed.de")) is not None
|
||||
|
||||
|
||||
def test_create_user_trims_and_normalizes_email():
|
||||
db = session()
|
||||
admin = User(
|
||||
email="admin@schubamed.de",
|
||||
first_name="Admin",
|
||||
last_name="User",
|
||||
role=UserRole.ADMIN.value,
|
||||
password_hash="hash",
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
)
|
||||
db.add(admin)
|
||||
db.flush()
|
||||
|
||||
response = domain_api.create_user(
|
||||
UserCreate(
|
||||
email=" New.User@Schubamed.DE ",
|
||||
first_name="New",
|
||||
last_name="User",
|
||||
role=UserRole.MITARBEITER,
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
temporary_password="TempPassword123!",
|
||||
),
|
||||
session=db,
|
||||
_=admin,
|
||||
)
|
||||
|
||||
payload = json.loads(response.body)
|
||||
assert response.status_code == 201
|
||||
assert payload["user"]["email"] == "new.user@schubamed.de"
|
||||
assert db.scalar(select(User).where(User.email == "new.user@schubamed.de")) is not None
|
||||
|
||||
|
||||
def test_duplicate_email_returns_409_and_rolls_back_session():
|
||||
from fastapi import HTTPException
|
||||
|
||||
db = session()
|
||||
admin = User(
|
||||
email="admin@schubamed.de",
|
||||
first_name="Admin",
|
||||
last_name="User",
|
||||
role=UserRole.ADMIN.value,
|
||||
password_hash="hash",
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
)
|
||||
db.add(admin)
|
||||
db.flush()
|
||||
|
||||
first = domain_api.create_user(
|
||||
UserCreate(
|
||||
email="duplicate@schubamed.de",
|
||||
first_name="First",
|
||||
last_name="User",
|
||||
role=UserRole.MITARBEITER,
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
temporary_password="TempPassword123!",
|
||||
),
|
||||
session=db,
|
||||
_=admin,
|
||||
)
|
||||
assert first.status_code == 201
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
domain_api.create_user(
|
||||
UserCreate(
|
||||
email="duplicate@schubamed.de",
|
||||
first_name="Second",
|
||||
last_name="User",
|
||||
role=UserRole.PRUEFER,
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
temporary_password="TempPassword123!",
|
||||
),
|
||||
session=db,
|
||||
_=admin,
|
||||
)
|
||||
assert exc_info.value.status_code == 409
|
||||
|
||||
follow_up = domain_api.create_user(
|
||||
UserCreate(
|
||||
email="fresh@schubamed.de",
|
||||
first_name="Fresh",
|
||||
last_name="User",
|
||||
role=UserRole.LESER,
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
temporary_password="TempPassword123!",
|
||||
),
|
||||
session=db,
|
||||
_=admin,
|
||||
)
|
||||
assert follow_up.status_code == 201
|
||||
|
||||
|
||||
def test_update_user_keeps_same_email_without_false_duplicate():
|
||||
db = session()
|
||||
admin = User(
|
||||
email="admin@schubamed.de",
|
||||
first_name="Admin",
|
||||
last_name="User",
|
||||
role=UserRole.ADMIN.value,
|
||||
password_hash="hash",
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
)
|
||||
target = User(
|
||||
email="target@schubamed.de",
|
||||
first_name="Target",
|
||||
last_name="User",
|
||||
role=UserRole.MITARBEITER.value,
|
||||
password_hash="hash",
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
)
|
||||
db.add_all([admin, target])
|
||||
db.flush()
|
||||
|
||||
response = domain_api.update_user(
|
||||
target.id,
|
||||
UserUpdate(
|
||||
first_name="Target",
|
||||
last_name="User",
|
||||
email="target@schubamed.de",
|
||||
role=UserRole.MITARBEITER,
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
),
|
||||
session=db,
|
||||
me=admin,
|
||||
)
|
||||
|
||||
assert response.email == "target@schubamed.de"
|
||||
assert response.id == target.id
|
||||
|
||||
|
||||
def test_delete_user_without_references_removes_row():
|
||||
db = session()
|
||||
admin = User(
|
||||
email="admin@schubamed.de",
|
||||
first_name="Admin",
|
||||
last_name="User",
|
||||
role=UserRole.ADMIN.value,
|
||||
password_hash="hash",
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
)
|
||||
user = User(
|
||||
email="delete-me@schubamed.de",
|
||||
first_name="Delete",
|
||||
last_name="Me",
|
||||
role=UserRole.MITARBEITER.value,
|
||||
password_hash="hash",
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
)
|
||||
db.add_all([admin, user])
|
||||
db.flush()
|
||||
|
||||
response = domain_api.delete_user(user.id, session=db, me=admin)
|
||||
|
||||
assert response.status_code == 204
|
||||
assert db.get(User, user.id) is None
|
||||
|
||||
|
||||
def test_delete_user_with_validation_reference_soft_deletes_row():
|
||||
db = session()
|
||||
customer, location, device = seed(db)
|
||||
admin = User(
|
||||
email="admin@schubamed.de",
|
||||
first_name="Admin",
|
||||
last_name="User",
|
||||
role=UserRole.ADMIN.value,
|
||||
password_hash="hash",
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
)
|
||||
examiner = User(
|
||||
email="examiner@schubamed.de",
|
||||
first_name="Examiner",
|
||||
last_name="User",
|
||||
role=UserRole.PRUEFER.value,
|
||||
password_hash="hash",
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
)
|
||||
validation = valid_validation(customer, location, device)
|
||||
db.add_all([admin, examiner, validation])
|
||||
db.flush()
|
||||
validation.examiner_id = examiner.id
|
||||
db.flush()
|
||||
|
||||
response = domain_api.delete_user(examiner.id, session=db, me=admin)
|
||||
|
||||
assert response.status_code == 204
|
||||
db.refresh(examiner)
|
||||
assert examiner.deleted_at is not None
|
||||
assert examiner.is_active is False
|
||||
listed = domain_api.list_users(session=db, _=admin, page=1, page_size=20, search=None, role=None, active=None, sort_by="created_at", sort_order="desc")
|
||||
assert all(item.id != examiner.id for item in listed["items"])
|
||||
|
||||
|
||||
def test_invalid_user_role_is_rejected_by_schema():
|
||||
with pytest.raises(Exception):
|
||||
UserCreate(
|
||||
email="bad-role@schubamed.de",
|
||||
first_name="Bad",
|
||||
last_name="Role",
|
||||
role="invalid-role", # type: ignore[arg-type]
|
||||
is_active=True,
|
||||
must_change_password=False,
|
||||
temporary_password="TempPassword123!",
|
||||
)
|
||||
|
||||
|
||||
def test_login_updates_last_login_at():
|
||||
from app.core.security import hash_password
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue