Validation_Suite/validation-suite/backend/mercury/tests/test_validation_workflow.py
2026-07-11 15:50:15 +02:00

647 lines
21 KiB
Python

from __future__ import annotations
import re
from datetime import date
from pathlib import Path
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session
from app.db.base import Base
from app.models.customer import Customer, CustomerType
from app.models.contact import Contact
from app.models.device import Device
from app.models.equipment import Equipment
from app.models.user import User, UserRole
from app.models.location import Location
from app.models.validation import Validation, ValidationStatus
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.schemas.auth import LoginRequest
from app.schemas.domain import 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
from app.modules.helios.service import HeliosImportService
def session() -> Session:
engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(engine)
return Session(engine)
def seed(session: Session):
customer = Customer(customer_type=CustomerType.practice, name="Praxis Test")
session.add(customer)
session.flush()
location = Location(customer_id=customer.id, name="OP")
session.add(location)
session.flush()
device = Device(
customer_id=customer.id,
location_id=location.id,
manufacturer="MELAG",
model="Vacuklav",
serial_number="SN-1",
)
session.add(device)
session.flush()
return customer, location, device
def seed_contact(session: Session, customer: Customer) -> Contact:
contact = Contact(customer_id=customer.id, full_name="Kontakt")
session.add(contact)
session.flush()
return contact
def valid_validation(customer: Customer, location: Location, device: Device) -> Validation:
return Validation(
report_number="VAL-TEST-1",
validation_type="Erstvalidierung",
performed_on=date.today(),
customer_id=customer.id,
location_id=location.id,
device_id=device.id,
examiner_name="Pruefer",
status=ValidationStatus.draft.value,
)
def test_required_fields_are_reported():
db = session()
validation = Validation(status=ValidationStatus.draft.value)
review = ValidationWorkflowService(db).review(validation)
assert {item["field"] for item in review["errors"]} >= {
"report_number",
"validation_type",
"performed_on",
"customer_id",
"location_id",
"device_id",
"examiner_name",
}
def test_status_changes_to_ready_when_no_errors():
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
db.add(validation)
db.flush()
review = ValidationWorkflowService(db).mark_ready_for_review(validation)
assert review["errors"] == []
assert validation.status == ValidationStatus.ready_for_review.value
def test_search_finds_report_number():
db = session()
customer, location, device = seed(db)
db.add(valid_validation(customer, location, device))
db.flush()
result = ValidationWorkflowService(db).query_validations(
search="VAL-TEST",
page=1,
page_size=10,
sort_by="updated_at",
sort_order="desc",
filters={},
)
assert result["total"] == 1
def test_csv_import_preview_detects_duplicate_and_missing_fields():
db = session()
customer, location, device = seed(db)
db.add(valid_validation(customer, location, device))
db.flush()
csv_content = (
"report_number,validation_type,test_date,customer_reference,location_reference,"
"device_serial_number,examiner,status,result,notes\n"
"VAL-TEST-1,Erstvalidierung,2026-07-11,Praxis Test,OP,SN-1,Pruefer,ENTWURF,offen,\n"
"VAL-NEW,,,,,,,\n"
).encode()
preview = ValidationWorkflowService(db).preview_csv(csv_content)
assert preview["duplicates"] == 1
assert preview["invalid_rows"] == 1
def test_validation_without_contact_id_can_be_saved():
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
validation.contact_id = None
db.add(validation)
db.flush()
assert validation.contact_id is None
def test_validation_with_valid_contact_id_can_be_saved():
db = session()
customer, location, device = seed(db)
contact = seed_contact(db, customer)
validation = valid_validation(customer, location, device)
validation.contact_id = contact.id
db.add(validation)
db.flush()
assert validation.contact_id == contact.id
def test_empty_contact_id_string_is_normalized_to_none():
payload = ValidationCreate(
report_number="VAL-EMPTY",
validation_type="Erstvalidierung",
performed_on=date.today(),
customer_id=None,
location_id="",
contact_id="",
device_id=None,
examiner_name="Pruefer",
)
assert payload.contact_id is None
assert payload.location_id is None
def test_invalid_uuid_returns_validation_error():
try:
ValidationCreate(
report_number="VAL-BAD",
validation_type="Erstvalidierung",
performed_on=date.today(),
customer_id="not-a-uuid",
device_id=None,
examiner_name="Pruefer",
)
except Exception as exc:
assert "uuid" in str(exc).lower()
else:
raise AssertionError("Invalid UUID was accepted")
def test_user_model_supports_password_management_fields():
db = session()
user = User(
email="user@schubamed.de",
first_name="Max",
last_name="Mustermann",
role=UserRole.PRUEFER.value,
password_hash="hash",
is_active=True,
must_change_password=True,
)
db.add(user)
db.flush()
assert user.full_name == "Max Mustermann"
assert user.must_change_password is True
assert user.role == UserRole.PRUEFER
def test_login_updates_last_login_at():
from app.core.security import hash_password
db = session()
user = User(
email="login@schubamed.de",
first_name="Login",
last_name="Test",
role=UserRole.MITARBEITER.value,
password_hash=hash_password("VerySecretPass123"),
is_active=True,
must_change_password=False,
)
db.add(user)
db.flush()
AuthService(db).login("login@schubamed.de", "VerySecretPass123")
assert user.last_login_at is not None
def test_login_response_contains_user_and_expires_in():
from app.core.security import hash_password
from app.core.config import settings
db = session()
user = User(
email="response@schubamed.de",
first_name="Response",
last_name="Test",
role=UserRole.PRUEFER.value,
password_hash=hash_password("VerySecretPass123"),
is_active=True,
must_change_password=True,
)
db.add(user)
db.flush()
response = auth_login(LoginRequest(email="response@schubamed.de", password="VerySecretPass123"), db)
assert response.expires_in == settings.access_token_minutes * 60
assert response.user.email == "response@schubamed.de"
assert response.user.must_change_password is True
def test_login_rejects_inactive_user_with_403():
from app.core.security import hash_password
from fastapi import HTTPException
db = session()
user = User(
email="inactive@schubamed.de",
first_name="Inactive",
last_name="User",
role=UserRole.PRUEFER.value,
password_hash=hash_password("VerySecretPass123"),
is_active=False,
must_change_password=False,
)
db.add(user)
db.flush()
try:
AuthService(db).login("inactive@schubamed.de", "VerySecretPass123")
except HTTPException as exc:
assert exc.status_code == 403
else:
raise AssertionError("Inactive user was accepted")
def test_reference_masterdata_import_is_idempotent_and_links_entities(tmp_path):
db = session()
reference = next(
candidate
for parent in Path(__file__).resolve().parents
if (candidate := parent / "docs/reference/reports/Erstvalidierung_Dr.Durmaz_Steri_12-25.docx").exists()
)
first = ReferenceMasterdataImportService(db).import_reference_docx(reference)
second = ReferenceMasterdataImportService(db).import_reference_docx(reference)
assert "customer" in first.created
assert "location" in first.created
assert "device" in first.created
assert first.validation_id is None
assert "customer" in second.unchanged
assert "device" in second.unchanged
assert db.scalar(select(Customer).where(Customer.name == "Urologische Praxis Dr. Durmaz")) is not None
assert db.scalar(select(Device).where(Device.serial_number == "EXN250688")) is not None
assert db.scalar(select(Equipment).where(Equipment.serial_number == "15125738")) is not None
def test_reference_masterdata_import_can_create_validation():
db = session()
reference = next(
candidate
for parent in Path(__file__).resolve().parents
if (candidate := parent / "docs/reference/reports/Erstvalidierung_Dr.Durmaz_Steri_12-25.docx").exists()
)
result = ReferenceMasterdataImportService(db).import_reference_docx(reference, create_validation=True)
assert result.validation_id is not None
validation = db.get(Validation, result.validation_id)
assert validation is not None
assert validation.status == ValidationStatus.draft.value
def test_revalidation_date_uses_calendar_months():
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
validation.performed_on = date(2026, 1, 31)
validation.revalidation_interval_months = 1
ValidationWorkflowService(db).apply_revalidation_date(validation)
assert validation.next_validation_on == date(2026, 2, 28)
def test_new_version_keeps_structured_json_and_previous_reference():
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
validation.status = ValidationStatus.approved.value
validation.documentation_checklist = [{"text": "A", "value": "yes"}]
db.add(validation)
db.flush()
clone = ValidationWorkflowService(db).create_new_version(validation)
assert clone.previous_validation_id == validation.id
assert clone.version == 2
assert clone.documentation_checklist == validation.documentation_checklist
def test_orion_renders_reference_main_chapters(tmp_path):
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
db.add(validation)
db.flush()
html = OrionReportService(db, tmp_path).render_html(validation.id)
assert "1 Funktionsqualifikation" in html
assert "2.2 Prüfmittel zur thermoelektrischen Untersuchung" in html
assert "4 Ergebnisse der Validierung" in html
assert "9 Werkskalibrierzertifikate Sensoren" in html
def test_orion_renders_uploaded_images_as_real_images(tmp_path):
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
validation.attachments = [
{
"category": "Beladung",
"filename": "beladung.png",
"content_type": "image/png",
"description": "Beladungsmuster Testlauf 1",
"order": 1,
"url": "/uploads/validations/example/beladung.png",
}
]
db.add(validation)
db.flush()
html = OrionReportService(db, tmp_path).render_html(validation.id)
assert '<img class="report-image"' in html
assert "http://localhost:8000/uploads/validations/example/beladung.png" in html
assert "Abbildung 1: Beladungsmuster Testlauf 1" in html
def test_weasyprint_renders_minimal_pdf_bytes():
from weasyprint import HTML
pdf_bytes = HTML(string="<html><body><h1>Orion PDF Test</h1></body></html>").write_pdf()
assert pdf_bytes.startswith(b"%PDF")
assert len(pdf_bytes) > 1024
def test_orion_renders_real_pdf_with_logo(tmp_path):
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
db.add(validation)
db.flush()
assert SCHUBAMED_LOGO_PATH.exists()
html = OrionReportService(db, tmp_path).render_html(validation.id)
pdf_path = OrionReportService(db, tmp_path).render_pdf(validation.id)
assert schubamed_logo_uri() in html
assert "Neutraler Logo-Platzhalter" not in html
assert pdf_path.read_bytes().startswith(b"%PDF")
assert pdf_path.stat().st_size > 1024
def test_orion_header_footer_layout_markers_are_rendered(tmp_path):
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
validation.report_number = "SV-2026-00005"
db.add(validation)
db.flush()
html = OrionReportService(db, tmp_path).render_html(validation.id)
assert 'class="report-header"' in html
assert 'class="report-header-brand"' in html
assert "SCHUBAMED®" in html
assert "Aufbereitung mit System" in html
assert "SV-2026-00005" in html
assert "Version" in html
assert "Datum" in html
assert 'class="report-footer"' in html
assert "Validation Suite" in html
assert "Seite" in html
assert "page-count" in html
def test_orion_layout_css_reserves_header_footer_space():
css = Path("app/modules/orion/templates/report.css").read_text(encoding="utf-8")
assert "margin: 34mm 18mm 24mm 18mm" in css
assert "position: running(report-header)" in css
assert "position: running(report-footer)" in css
assert "grid-template-columns: 1fr 54mm" in css
assert "height: 24mm" in css
assert "height: 20mm" in css
assert "font-size: 22pt" in css
assert "counter(pages)" in css
def test_reference_template_textblocks_and_checklists_are_loaded():
db = session()
bundle = ReportTemplateService(db).ensure_default_template()
assert bundle.template.template_key == "small_steam_sterilizer_initial_validation"
assert ReportTemplateService(db).reference_path.exists()
assert {"summary", "bq_goal", "legal", "performance"} <= set(bundle.text_blocks)
assert len(bundle.checklists) >= 6
assert any(item.title == "Beschreibung Sterilisator" for item in bundle.checklists)
def test_orion_contains_required_reference_sections_and_three_runs(tmp_path):
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
db.add(validation)
db.flush()
html = OrionReportService(db, tmp_path).render_html(validation.id)
for title in [
"Zusammenfassendes Ergebnis der Validierung",
"1.1 Anlass und Ziel der Prüfung",
"1.2 Gesetzliche Grundlagen",
"3.2 Standardbeladung (1. Durchlauf)",
"3.3 Standardbeladung (2. Durchlauf)",
"3.4 Standardbeladung (3. Durchlauf)",
]:
assert title in html
assert "Neutraler Logo-Platzhalter" not in html
assert "report-header-brand" in html
def test_orion_toc_uses_central_section_numbers_and_titles(tmp_path):
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
db.add(validation)
db.flush()
html = OrionReportService(db, tmp_path).render_html(validation.id)
expected = [
"1.3 Angaben zum Gerät",
"1.8 Umgebungsbedingungen",
"2.3 Prüfkonfiguration",
"3.1.1 Vakuumtest",
"4.2 Empfehlungen und Auflagen",
"9 Werkskalibrierzertifikate Sensoren",
]
for label in expected:
assert label in html
assert "3.1.1 Leistungsqualifikation" not in html
assert "4.2 Trocknung" not in html
assert html.index("1 Funktionsqualifikation") < html.index("9 Werkskalibrierzertifikate")
def test_orion_html_has_single_cover_before_report_content(tmp_path):
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
db.add(validation)
db.flush()
html = OrionReportService(db, tmp_path).render_html(validation.id)
assert html.count('class="cover-page"') == 1
assert html.index('class="cover-page"') < html.index('class="report-content"')
assert html.index('class="cover-page"') < html.index('class="report-meta"')
assert "<h1>PRÜFBERICHT ZUR VALIDIERUNG</h1>" in html
def test_orion_pdf_first_page_is_cover_not_blank(tmp_path):
from pypdf import PdfReader
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
db.add(validation)
db.flush()
pdf_path = OrionReportService(db, tmp_path).render_pdf(validation.id)
first_page_text = PdfReader(str(pdf_path)).pages[0].extract_text() or ""
normalized_text = re.sub(r"\s+", "", first_page_text.upper())
assert "PRÜFBERICHTZURVALIDIERUNG" in normalized_text
assert "FUNKTIONS-UNDLEISTUNGSQUALIFIKATION" in normalized_text
def test_orion_bookmark_labels_are_created_from_central_sections(tmp_path):
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
db.add(validation)
db.flush()
html = OrionReportService(db, tmp_path).render_html(validation.id)
bookmark_count = html.count("data-bookmark-label=")
toc_count = len([item for item in REPORT_SECTIONS if item.get("toc")])
assert bookmark_count == toc_count
for item in REPORT_SECTIONS:
number = item.get("number")
label = f"{number} {item['title']}" if number else item["title"]
assert f'data-bookmark-label="{label}"' in html
def test_orion_appends_pdf_attachments(tmp_path):
from pypdf import PdfReader, PdfWriter
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
attachment_path = tmp_path / "anlage.pdf"
writer = PdfWriter()
writer.add_blank_page(width=200, height=200)
with attachment_path.open("wb") as handle:
writer.write(handle)
validation.attachments = [
{
"category": "Kalibrierschein",
"filename": "anlage.pdf",
"content_type": "application/pdf",
"description": "Anlage",
"order": 1,
"storage_path": str(attachment_path),
}
]
db.add(validation)
db.flush()
pdf_path = OrionReportService(db, tmp_path).render_pdf(validation.id)
assert len(PdfReader(str(pdf_path)).pages) >= 2
def test_winlog_pdf_import_preview_and_confirmation(tmp_path):
from weasyprint import HTML
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
db.add(validation)
db.flush()
pdf_bytes = HTML(
string="<p>Vakuumtest</p><p>Programm: Vakuum</p><p>Leckrate: 0,1</p><p>Ergebnis: bestanden</p>"
).write_pdf()
preview = HeliosImportService(db, tmp_path).save_winlog_pdf(
validation.id, "winlog.pdf", pdf_bytes
)
assert preview["status"] == "VORSCHAU_BEREIT"
assert preview["sha256"]
assert any(value["field_name"] == "leak_rate" for value in preview["values"])
first = preview["values"][0]
confirmed = HeliosImportService(db, tmp_path).confirm_values(
preview["id"], [{"id": first["id"], "confirmed": True, "corrected_value": "korrigiert"}]
)
assert confirmed["status"] == "BESTAETIGT"
def test_unreadable_winlog_pdf_does_not_crash(tmp_path):
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
db.add(validation)
db.flush()
preview = HeliosImportService(db, tmp_path).save_winlog_pdf(
validation.id, "broken.pdf", b"not a pdf"
)
assert preview["status"] in {"FEHLER", "NUR_ANLAGE"}
assert preview["values"] == []
def test_confirmed_measurements_appear_in_orion_report(tmp_path):
from weasyprint import HTML
db = session()
customer, location, device = seed(db)
validation = valid_validation(customer, location, device)
db.add(validation)
db.flush()
pdf_bytes = HTML(string="<p>Testlauf 1</p><p>Leckrate: 0,2</p>").write_pdf()
preview = HeliosImportService(db, tmp_path).save_winlog_pdf(validation.id, "winlog.pdf", pdf_bytes)
leak_rate = next(value for value in preview["values"] if value["field_name"] == "leak_rate")
HeliosImportService(db, tmp_path).confirm_values(
preview["id"], [{"id": leak_rate["id"], "confirmed": True}]
)
html = OrionReportService(db, tmp_path).render_html(validation.id)
assert "leak_rate" in html
assert "0,2" in html