Validation_Suite/validation-suite/backend/mercury/tests/test_validation_workflow.py

from __future__ import annotations

import json
import re
from datetime import date
from pathlib import Path

import pytest
from sqlalchemy import create_engine, func, 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.services.quickstart_service import QuickStartService
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 QuickStartCreateRequest, UserCreate, UserUpdate, ValidationCreate
from app.schemas.domain import OrionReportSettingsUpdate
from app.modules.orion.service import OrionReportService
from app.modules.orion.assets import SCHUBAMED_LOGO_PATH, schubamed_logo_uri
from app.modules.orion.result import validation_result_box, validation_result_presentation
from app.modules.orion.template_service import REPORT_SECTIONS, ReportTemplateService
from app.modules.helios.service import HeliosImportService
from app.services.report_settings import OrionReportSettingsService
from app.services.customer_import import CustomerImportService


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_result_filter_accepts_canonical_values_for_legacy_rows():
    db = session()
    customer, location, device = seed(db)
    validation = valid_validation(customer, location, device)
    validation.result = "mit_auflagen"
    db.add(validation)
    db.commit()

    result = ValidationWorkflowService(db).query_validations(
        search=None,
        page=1,
        page_size=20,
        sort_by="updated_at",
        sort_order="desc",
        filters={"result": "BESTANDEN_MIT_AUFLAGEN"},
    )

    assert result["total"] == 1
    assert result["items"][0].id == validation.id


def test_dashboard_exposes_validation_result_counts():
    db = session()
    customer, location, device = seed(db)
    values = ["BESTANDEN", "mit_auflagen", "nicht_bestanden", "offen"]
    for index, result_value in enumerate(values, start=1):
        validation = valid_validation(customer, location, device)
        validation.report_number = f"VAL-RESULT-{index}"
        validation.result = result_value
        db.add(validation)
    db.commit()

    payload = domain_api.dashboard(session=db)

    assert payload["validation_result_passed"] == 1
    assert payload["validation_result_conditional"] == 1
    assert payload["validation_result_failed"] == 1
    assert payload["validation_result_open"] == 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_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_paginated_lists_expose_pages_for_masterdata_and_validations():
    db = session()
    customer, location, device = seed(db)
    db.add(valid_validation(customer, location, device))
    db.flush()

    customers = domain_api.list_customers(params={"page": 1, "page_size": 20, "search": None}, session=db)
    devices = domain_api.list_devices(params={"page": 1, "page_size": 20, "search": None}, session=db)
    validations = domain_api.list_validations(
        search=None,
        page=1,
        page_size=20,
        sort_by="created_at",
        sort_order="desc",
        status=None,
        customer_id=None,
        device_id=None,
        validation_type=None,
        result=None,
        date_from=None,
        date_to=None,
        overdue_only=False,
        session=db,
    )

    assert customers["pages"] == 1
    assert devices["pages"] == 1
    assert validations["pages"] == 1


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

    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_quickstart_customer_data_and_validation_creation():
    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()
    )
    ReferenceMasterdataImportService(db).import_reference_docx(reference, create_validation=True)
    customer = db.scalar(select(Customer).where(Customer.name == "Urologische Praxis Dr. Durmaz"))
    assert customer is not None
    customer_data = QuickStartService(db).customer_data(customer.id)
    assert customer_data.locations
    assert customer_data.contacts
    assert customer_data.devices
    assert customer_data.validations

    admin = User(
        email="quickstart-admin@schubamed.de",
        first_name="Quickstart",
        last_name="Admin",
        role=UserRole.ADMIN.value,
        password_hash="hash",
        is_active=True,
        must_change_password=False,
    )
    db.add(admin)
    db.flush()
    created = QuickStartService(db).create_validation(
        QuickStartCreateRequest(
            customer_id=customer.id,
            location_id=customer_data.locations[0].id,
            contact_id=customer_data.contacts[0].id,
            device_id=customer_data.devices[0].id,
            validation_type="Erstvalidierung",
        ),
        examiner=admin,
    )
    db.commit()
    assert created.id is not None
    assert created.status == ValidationStatus.draft.value
    assert created.customer_id == customer.id
    assert created.examiner_id == admin.id
    assert created.documentation_checklist
    assert created.programs
    assert created.loading_patterns


def test_quickstart_auto_selects_single_customer_masterdata():
    db = session()
    customer, location, device = seed(db)
    contact = seed_contact(db, customer)
    admin = User(
        email="single-data-admin@schubamed.de",
        first_name="Quickstart",
        last_name="Admin",
        role=UserRole.ADMIN.value,
        password_hash="hash",
        is_active=True,
        must_change_password=False,
    )
    db.add(admin)
    db.flush()

    created = QuickStartService(db).create_validation(
        QuickStartCreateRequest(
            customer_id=customer.id,
            validation_type="Erstvalidierung",
        ),
        examiner=admin,
    )

    assert created.customer_id == customer.id
    assert created.location_id == location.id
    assert created.contact_id == contact.id
    assert created.device_id == device.id


def test_quickstart_requires_location_when_customer_has_multiple_locations():
    db = session()
    customer, location, device = seed(db)
    db.add(Location(customer_id=customer.id, name="Zweiter Standort"))
    admin = User(
        email="multi-location-admin@schubamed.de",
        first_name="Quickstart",
        last_name="Admin",
        role=UserRole.ADMIN.value,
        password_hash="hash",
        is_active=True,
        must_change_password=False,
    )
    db.add(admin)
    db.flush()

    with pytest.raises(ValueError, match="Bitte Standort auswählen"):
        QuickStartService(db).create_validation(
            QuickStartCreateRequest(
                customer_id=customer.id,
                device_id=device.id,
                validation_type="Erstvalidierung",
            ),
            examiner=admin,
        )

    created = QuickStartService(db).create_validation(
        QuickStartCreateRequest(
            customer_id=customer.id,
            location_id=location.id,
            device_id=device.id,
            validation_type="Erstvalidierung",
        ),
        examiner=admin,
    )
    assert created.location_id == location.id


def test_quickstart_rejects_foreign_location_and_contact_ids():
    db = session()
    customer, location, device = seed(db)
    other_customer = Customer(customer_type=CustomerType.practice, name="Andere Praxis")
    db.add(other_customer)
    db.flush()
    other_location = Location(customer_id=other_customer.id, name="Fremder Standort")
    other_contact = Contact(customer_id=other_customer.id, full_name="Fremder Kontakt")
    admin = User(
        email="foreign-masterdata-admin@schubamed.de",
        first_name="Quickstart",
        last_name="Admin",
        role=UserRole.ADMIN.value,
        password_hash="hash",
        is_active=True,
        must_change_password=False,
    )
    db.add_all([other_location, other_contact, admin])
    db.flush()

    with pytest.raises(ValueError, match="Standort gehört nicht"):
        QuickStartService(db).create_validation(
            QuickStartCreateRequest(
                customer_id=customer.id,
                location_id=other_location.id,
                device_id=device.id,
                validation_type="Erstvalidierung",
            ),
            examiner=admin,
        )

    with pytest.raises(ValueError, match="Ansprechpartner gehört nicht"):
        QuickStartService(db).create_validation(
            QuickStartCreateRequest(
                customer_id=customer.id,
                location_id=location.id,
                contact_id=other_contact.id,
                device_id=device.id,
                validation_type="Erstvalidierung",
            ),
            examiner=admin,
        )


def test_quickstart_allows_multiple_contacts_without_required_selection():
    db = session()
    customer, location, device = seed(db)
    db.add_all([
        Contact(customer_id=customer.id, full_name="Kontakt A"),
        Contact(customer_id=customer.id, full_name="Kontakt B"),
    ])
    admin = User(
        email="optional-contact-admin@schubamed.de",
        first_name="Quickstart",
        last_name="Admin",
        role=UserRole.ADMIN.value,
        password_hash="hash",
        is_active=True,
        must_change_password=False,
    )
    db.add(admin)
    db.flush()

    created = QuickStartService(db).create_validation(
        QuickStartCreateRequest(
            customer_id=customer.id,
            location_id=location.id,
            device_id=device.id,
            validation_type="Erstvalidierung",
        ),
        examiner=admin,
    )

    assert created.contact_id is None


def test_quickstart_device_last_validation_returns_latest_completed_validation():
    db = session()
    customer, location, device = seed(db)
    draft = valid_validation(customer, location, device)
    draft.status = ValidationStatus.draft.value
    completed = valid_validation(customer, location, device)
    completed.report_number = "VAL-PAST"
    completed.status = ValidationStatus.completed.value
    completed.performed_on = date(2026, 1, 1)
    db.add_all([draft, completed])
    db.flush()

    summary = QuickStartService(db).device_last_validation(device.id)

    assert summary is not None
    assert summary.report_number == "VAL-PAST"
    assert summary.device_id == device.id


def test_quickstart_revalidation_copies_structured_data_but_not_measurements():
    db = session()
    customer, location, device = seed(db)
    admin = User(
        email="revalidation-admin@schubamed.de",
        first_name="Quickstart",
        last_name="Admin",
        role=UserRole.ADMIN.value,
        password_hash="hash",
        is_active=True,
        must_change_password=False,
    )
    base = valid_validation(customer, location, device)
    base.status = ValidationStatus.approved.value
    base.documentation_checklist = [{"text": "Prüfung A", "value": "yes"}]
    base.performance_checklist = [{"text": "Prüfung B", "value": "no"}]
    base.programs = [{"name": "Vakuumtest", "selected": True}]
    base.loading_patterns = [{"run": 1, "pattern": "Standard", "description": "Basis", "images": ["x"]}]
    base.measurement_data = [{"field": "temperatur"}]
    base.attachments = [{"filename": "bild.png"}]
    db.add_all([admin, base])
    db.flush()

    created = QuickStartService(db).create_validation(
        QuickStartCreateRequest(
            customer_id=customer.id,
            device_id=device.id,
            validation_type="Revalidierung",
        ),
        examiner=admin,
    )

    assert created.previous_validation_id == base.id
    assert created.version == 2
    assert created.documentation_checklist == base.documentation_checklist
    assert created.performance_checklist == base.performance_checklist
    assert created.programs == base.programs
    assert created.loading_patterns == base.loading_patterns
    assert created.measurement_data == []
    assert created.attachments == []


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)
    validation.result = "BESTANDEN"
    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_validation_result_presentations_cover_supported_values():
    expected = {
        "BESTANDEN": ("Bestanden", "result-box--passed"),
        "BESTANDEN_MIT_AUFLAGEN": ("Bestanden mit Auflagen", "result-box--conditional"),
        "NICHT_BESTANDEN": ("Nicht bestanden", "result-box--failed"),
        "OFFEN": ("Noch nicht bewertet", "result-box--open"),
    }

    for value, (label, css_class) in expected.items():
        presentation = validation_result_presentation(value)
        assert presentation.label == label
        assert presentation.css_class == css_class
        assert css_class in validation_result_box(value)

    assert validation_result_presentation("mit_auflagen").label == "Bestanden mit Auflagen"


def test_orion_result_boxes_render_on_cover_summary_and_results(tmp_path):
    db = session()
    customer, location, device = seed(db)
    validation = valid_validation(customer, location, device)
    validation.result = "BESTANDEN_MIT_AUFLAGEN"
    db.add(validation)
    db.flush()

    html = OrionReportService(db, tmp_path).render_html(validation.id)

    assert html.count("VALIDIERUNGSERGEBNIS") >= 2
    assert 'class="result-box result-box--cover' not in html
    assert html.count('class="result-box result-box--compact') >= 2
    assert "Ergebnis der Validierung" in html
    assert "Die technische Validierung wurde mit Auflagen bestanden." in html
    assert "BESTANDEN MIT AUFLAGEN" not in html
    assert "Bestanden mit Auflagen" in html
    assert 'class="cover-closing"' in html
    assert 'class="cover-result-text"' in html
    assert 'class="signature-block"' in html
    assert "Unterschrift technische Validierung" in html
    assert "Unterschrift Auftraggeber" not in html
    assert html.index('id="summary"') < html.index("Bestanden mit Auflagen")
    assert html.index('id="results"') < html.rindex("Bestanden mit Auflagen")


def test_orion_cover_result_text_for_supported_statuses(tmp_path):
    expected = {
        "BESTANDEN": "Die technische Validierung wurde erfolgreich bestanden.",
        "BESTANDEN_MIT_AUFLAGEN": "Die technische Validierung wurde mit Auflagen bestanden.",
        "NICHT_BESTANDEN": "Die technische Validierung wurde nicht bestanden.",
    }

    for value, sentence in expected.items():
        db = session()
        customer, location, device = seed(db)
        validation = valid_validation(customer, location, device)
        validation.report_number = f"VAL-{value}"
        validation.result = value
        db.add(validation)
        db.flush()

        html = OrionReportService(db, tmp_path / value).render_html(validation.id)

        assert "Ergebnis der Validierung" in html
        assert sentence in html
        assert 'class="result-box result-box--cover' not in html
        assert html.count('class="result-box result-box--compact') >= 2


def test_orion_result_box_css_prevents_page_breaks():
    css = Path("app/modules/orion/templates/report.css").read_text(encoding="utf-8")

    assert ".result-box" in css
    assert ".cover-closing" in css
    assert ".cover-result-text" in css
    assert ".signature-block" in css
    assert "margin-top: 9mm" in css
    assert "break-inside: avoid" in css
    assert "page-break-inside: avoid" in css
    assert "#EAF5EE" in css
    assert "#FFF7DD" in css
    assert "#FCEBEC" in css


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)
    validation.result = "BESTANDEN"
    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
    assert "ERGEBNISDERVALIDIERUNG" in normalized_text
    assert "DIETECHNISCHEVALIDIERUNGWURDEERFOLGREICHBESTANDEN" 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


def seed_admin(db: Session) -> User:
    user = 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(user)
    db.flush()
    return user


def report_settings_payload(**overrides) -> OrionReportSettingsUpdate:
    values = {
        "layout_profile": "STANDARD",
        "show_cover_result_text": True,
        "signature_mode": "TECHNICAL_ONLY",
        "image_size": "MEDIUM",
        "page_break_before_main_chapters": True,
        "page_margin": "STANDARD",
        "section_spacing": "STANDARD",
        "table_layout": "STANDARD",
        "table_font_size": "STANDARD",
        "image_position": "SIDE_BY_SIDE",
        "max_images_per_page": 2,
        "show_image_captions": True,
        "show_header": True,
        "show_footer": True,
        "logo_size": "MEDIUM",
        "compact_cover": False,
    }
    values.update(overrides)
    return OrionReportSettingsUpdate(**values)


def test_report_settings_defaults_and_update_persist():
    db = session()
    admin = seed_admin(db)

    settings = OrionReportSettingsService(db).get_global()

    assert settings.scope == "GLOBAL"
    assert settings.layout_profile == "STANDARD"
    assert settings.show_cover_result_text is True
    assert settings.signature_mode == "TECHNICAL_ONLY"
    assert settings.image_size == "MEDIUM"
    assert settings.page_break_before_main_chapters is True
    assert settings.page_margin == "STANDARD"
    assert settings.section_spacing == "STANDARD"
    assert settings.table_layout == "STANDARD"
    assert settings.table_font_size == "STANDARD"
    assert settings.image_position == "SIDE_BY_SIDE"
    assert settings.max_images_per_page == 2
    assert settings.show_image_captions is True
    assert settings.show_header is True
    assert settings.show_footer is True
    assert settings.logo_size == "MEDIUM"
    assert settings.compact_cover is False

    updated = OrionReportSettingsService(db).update_global(
        report_settings_payload(
            layout_profile="COMPACT",
            show_cover_result_text=False,
            signature_mode="TECHNICAL_AND_CLIENT",
            image_size="LARGE",
            page_break_before_main_chapters=False,
            page_margin="WIDE",
            section_spacing="COMPACT",
            table_layout="COMPACT",
            table_font_size="SMALL",
            image_position="STACKED",
            max_images_per_page=4,
            show_image_captions=False,
            show_header=False,
            show_footer=False,
            logo_size="LARGE",
            compact_cover=True,
        ),
        admin,
    )
    db.commit()

    assert updated.layout_profile == "COMPACT"
    assert updated.show_cover_result_text is False
    assert updated.signature_mode == "TECHNICAL_AND_CLIENT"
    assert updated.image_size == "LARGE"
    assert updated.page_break_before_main_chapters is False
    assert updated.page_margin == "WIDE"
    assert updated.section_spacing == "COMPACT"
    assert updated.table_layout == "COMPACT"
    assert updated.table_font_size == "SMALL"
    assert updated.image_position == "STACKED"
    assert updated.max_images_per_page == 4
    assert updated.show_image_captions is False
    assert updated.show_header is False
    assert updated.show_footer is False
    assert updated.logo_size == "LARGE"
    assert updated.compact_cover is True
    assert updated.updated_by_user_id == admin.id


def test_report_settings_invalid_enum_is_rejected():
    with pytest.raises(Exception):
        report_settings_payload(
            layout_profile="WILD",
        )


def test_orion_settings_preview_pdf_contains_pdf_bytes(tmp_path):
    db = session()
    seed_admin(db)

    pdf = OrionReportService(db, tmp_path).render_settings_preview_pdf(
        report_settings_payload(
            layout_profile="COMPACT",
            show_cover_result_text=True,
            signature_mode="TECHNICAL_AND_CLIENT",
            image_size="SMALL",
            page_break_before_main_chapters=False,
            page_margin="NARROW",
            section_spacing="COMPACT",
            table_layout="COMPACT",
            table_font_size="SMALL",
            image_position="STACKED",
            max_images_per_page=1,
            show_image_captions=False,
            show_header=False,
            show_footer=False,
            logo_size="SMALL",
            compact_cover=True,
        )
    )

    assert pdf.startswith(b"%PDF")
    assert len(pdf) > 1024


def test_orion_report_settings_affect_html_without_removing_summary_result(tmp_path):
    db = session()
    customer, location, device = seed(db)
    validation = valid_validation(customer, location, device)
    validation.result = "BESTANDEN"
    db.add(validation)
    admin = seed_admin(db)
    OrionReportSettingsService(db).update_global(
        report_settings_payload(
            layout_profile="COMPACT",
            show_cover_result_text=False,
            signature_mode="TECHNICAL_AND_CLIENT",
            image_size="SMALL",
            page_break_before_main_chapters=False,
            page_margin="NARROW",
            section_spacing="COMPACT",
            table_layout="COMPACT",
            table_font_size="SMALL",
            image_position="STACKED",
            max_images_per_page=1,
            show_image_captions=False,
            show_header=False,
            show_footer=False,
            logo_size="SMALL",
            compact_cover=True,
        ),
        admin,
    )
    db.commit()

    html = OrionReportService(db, tmp_path).render_html(validation.id)

    assert "layout-compact" in html
    assert "page-margin-narrow" in html
    assert "section-spacing-compact" in html
    assert "table-layout-compact" in html
    assert "table-font-small" in html
    assert "image-size-small" in html
    assert "image-position-stacked" in html
    assert "image-page-max-1" in html
    assert "image-captions-off" in html
    assert "logo-size-small" in html
    assert "cover-compact" in html
    assert "main-chapter-flow" in html
    assert "Ergebnis der Validierung" not in html.split('<section class="report-content">')[0]
    assert "Unterschrift Auftraggeber" in html
    assert '<header class="report-header"' not in html
    assert '<footer class="report-footer"' not in html
    assert html.count("result-box") >= 1


def xlsx_bytes(headers: list[str], rows: list[list[str]]) -> bytes:
    from zipfile import ZIP_DEFLATED, ZipFile
    import io

    def cell_ref(index: int, row_number: int) -> str:
        return f"{chr(65 + index)}{row_number}"

    sheet_rows = []
    for row_number, row in enumerate([headers, *rows], start=1):
        cells = "".join(
            f'<c r="{cell_ref(index, row_number)}" t="inlineStr"><is><t>{value}</t></is></c>'
            for index, value in enumerate(row)
        )
        sheet_rows.append(f'<row r="{row_number}">{cells}</row>')
    buffer = io.BytesIO()
    with ZipFile(buffer, "w", ZIP_DEFLATED) as archive:
        archive.writestr("[Content_Types].xml", '<Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types"><Default Extension="xml" ContentType="application/xml"/></Types>')
        archive.writestr("xl/workbook.xml", '<workbook xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main"><sheets><sheet name="Tabelle1" sheetId="1" r:id="rId1" xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships"/></sheets></workbook>')
        archive.writestr("xl/worksheets/sheet1.xml", f'<worksheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main"><sheetData>{"".join(sheet_rows)}</sheetData></worksheet>')
    return buffer.getvalue()


def test_customer_import_csv_semicolon_preview_and_confirm():
    db = session()
    admin = seed_admin(db)
    content = "Kundennr;Firma;PLZ;Ort;E-Mail;Ansprechpartner\nD4-1;Praxis Müller;10115;Berlin;info@praxis.de;Max Mustermann\n".encode("utf-8-sig")

    service = CustomerImportService(db)
    preview = service.preview("desk4.csv", content)

    assert preview["columns"] == ["Kundennr", "Firma", "PLZ", "Ort", "E-Mail", "Ansprechpartner"]
    assert preview["mapping"]["Kundennr"] == "customer_number"
    assert preview["rows"][0]["action"] == "NEU_ANLEGEN"
    assert db.scalar(select(func.count()).select_from(Customer)) == 0

    result = service.confirm("desk4.csv", content, preview["mapping"], {}, True, admin)

    assert result["summary"]["created_customers"] == 1
    assert result["summary"]["created_locations"] == 1
    assert result["summary"]["created_contacts"] == 1
    customer = db.scalar(select(Customer).where(Customer.external_id == "D4-1"))
    assert customer is not None
    assert customer.source_system == "desk4"
    assert customer.name == "Praxis Müller"


def test_customer_import_csv_comma_existing_customer_and_ignore_empty_values():
    db = session()
    admin = seed_admin(db)
    existing = Customer(
        customer_type=CustomerType.practice,
        source_system="desk4",
        external_id="D4-2",
        name="Alt",
        email="alt@example.de",
    )
    db.add(existing)
    db.commit()
    content = "Kundennr,Firma,E-Mail\nD4-2,Neu,\n".encode()
    service = CustomerImportService(db)
    preview = service.preview("desk4.csv", content)

    assert preview["rows"][0]["action"] == "BESTEHENDEN_AKTUALISIEREN"
    service.confirm("desk4.csv", content, preview["mapping"], {}, True, admin)
    db.refresh(existing)

    assert existing.name == "Neu"
    assert existing.email == "alt@example.de"


def test_customer_import_detects_possible_duplicate_and_invalid_email():
    db = session()
    db.add(Customer(customer_type=CustomerType.practice, name="Praxis A", postal_code="12345"))
    db.commit()
    service = CustomerImportService(db)

    duplicate = service.preview("desk4.csv", "Firma;PLZ\nPraxis A;12345\n".encode())
    invalid = service.preview("desk4.csv", "Firma;E-Mail\nPraxis B;nicht-mail\n".encode())

    assert duplicate["rows"][0]["action"] == "UEBERSPRINGEN"
    assert duplicate["summary"]["duplicates"] == 1
    assert invalid["rows"][0]["action"] == "ERROR"
    assert "ungültig" in invalid["rows"][0]["errors"][0]


def test_customer_import_xlsx_and_invalid_files():
    db = session()
    service = CustomerImportService(db)
    preview = service.preview(
        "desk4.xlsx",
        xlsx_bytes(["Kundennr", "Firma", "Ort"], [["X-1", "Praxis XLSX", "Hamburg"]]),
    )

    assert preview["rows"][0]["recognized"]["customer"]["name"] == "Praxis XLSX"
    with pytest.raises(ValueError):
        service.preview("desk4.txt", b"Firma\nTest")
    with pytest.raises(ValueError):
        service.preview("desk4.csv", b"")


def test_customer_import_missing_required_mapping_is_error():
    db = session()
    service = CustomerImportService(db)
    content = "Telefon\n123\n".encode()
    preview = service.preview("desk4.csv", content, {"Telefon": "customer_phone"})

    assert preview["rows"][0]["action"] == "ERROR"
    assert "Kundennummer oder Firmenname fehlt." in preview["rows"][0]["errors"]