556 lines
18 KiB
Python
556 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import create_engine
|
|
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.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.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_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
|