439 lines
14 KiB
Python
439 lines
14 KiB
Python
from __future__ import annotations
|
||
|
||
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.location import Location
|
||
from app.models.validation import Validation, ValidationStatus
|
||
from app.services.validation_workflow import ValidationWorkflowService
|
||
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 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_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" 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 – Testlauf 1",
|
||
"3.3 Standardbeladung – Testlauf 2",
|
||
"3.4 Standardbeladung – Testlauf 3",
|
||
]:
|
||
assert title in html
|
||
assert "Neutraler Logo-Platzhalter" not in html
|
||
assert "report-header-brand" 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
|