AareCommon/tests/test_exception_handler.py

import pytest
from aarecommon.errors.exception_handler import (
    AareDBCommunicationError,
    AerotechCommunicationError,
    AuthenticationException,
    AuthErrorCode,
    AXCFailed,
    BeamlineBusyException,
    CriticalTellException,
    DataCollectionException,
    JFJochCommunicationError,
    LoopCenteringFailed,
    MagnetPositionSensorErorr,
    ManualMountException,
    MountingFailed,
    RasterScanException,
    SampleException,
    SmargonCommunicationError,
    SmartMagnetFaultException,
    TellCommandWhileBusyException,
    TellCommunicationError,
    TellConnectionException,
    TransformationInvalidException,
    UnmountingFailed,
    UserRightsException,
    WarningTellException,
)


def test_data_collection_exception_message():
    exc = DataCollectionException("Custom error")
    assert str(exc) == "Custom error"


def test_data_collection_exception_default_message():
    exc = DataCollectionException()
    assert str(exc) == "Data collection failed"


def test_authentication_exception_properties():
    exc = AuthenticationException("Failed", status_code=403, code=AuthErrorCode.FORBIDDEN)
    assert exc.status_code == 403
    assert exc.code == AuthErrorCode.FORBIDDEN
    assert "Failed" in str(exc)


def test_tell_communication_error_str():
    exc = TellCommunicationError("Timeout", endpoint="/state", operation="GET")
    assert str(exc) == "Timeout"
    assert exc.endpoint == "/state"
    assert exc.operation == "GET"


def test_transformation_invalid_exception():
    exc = TransformationInvalidException()
    assert "Transformation is not implemented" in str(exc)


def test_raster_scan_exception():
    exc = RasterScanException("Raster failed")
    assert "Raster failed" in str(exc)


def test_loop_centering_failed():
    exc = LoopCenteringFailed()
    assert "Loop Centering did not detect a sample" in str(exc)


def test_unmounting_failed():
    exc = UnmountingFailed()
    assert "A sample was not unmounted" in str(exc)


def test_mounting_failed():
    exc = MountingFailed()
    assert "A sample was not mounted" in str(exc)


def test_manual_mount_exception():
    exc = ManualMountException()
    assert "Manual mounting failed" in str(exc)


def test_smart_magnet_fault_exception():
    exc = SmartMagnetFaultException()
    assert "Smart magnet fault" in str(exc)


def test_tell_command_while_busy_exception():
    exc = TellCommandWhileBusyException()
    assert "Tell is busy" in str(exc)


def test_tell_connection_exception():
    exc = TellConnectionException()
    assert "Lost connection to Tell" in str(exc)


def test_warning_tell_exception():
    exc = WarningTellException()
    assert "Warning error in TELL" in str(exc)


def test_critical_tell_exception():
    exc = CriticalTellException()
    assert "Critical error in TELL" in str(exc)


def test_axc_failed():
    exc = AXCFailed()
    assert "Auto X-ray centering failed" in str(exc)


def test_beamline_busy_exception():
    exc = BeamlineBusyException()
    assert "Beamline is in busy state" in str(exc)


def test_sample_exception():
    exc = SampleException()
    assert "Sample not found" in str(exc)


def test_user_rights_exception():
    exc = UserRightsException(code=AuthErrorCode.NOT_STAFF)
    assert exc.status_code == 403
    assert exc.code == AuthErrorCode.NOT_STAFF
    assert "User does not have rights" in str(exc)


def test_smargon_communication_error():
    exc = SmargonCommunicationError("Conn error", endpoint="/move", status_code=500)
    assert str(exc) == "Conn error"
    assert exc.endpoint == "/move"
    assert exc.status_code == 500


def test_jfjoch_communication_error():
    exc = JFJochCommunicationError("JFJoch error", operation="POST")
    assert str(exc) == "JFJoch error"
    assert exc.operation == "POST"


def test_aaredb_communication_error():
    exc = AareDBCommunicationError("DB error")
    assert "DB error" in str(exc)


def test_aerotech_communication_error():
    exc = AerotechCommunicationError("Aerotech error")
    assert "Aerotech error" in str(exc)


def test_magnet_position_sensor_error():
    exc = MagnetPositionSensorErorr()
    assert "Magnet position sensor error" in str(exc)


@pytest.mark.parametrize(
    "factory",
    [
        lambda: TransformationInvalidException("t"),
        lambda: LoopCenteringFailed("lc"),
        lambda: UnmountingFailed("un"),
        lambda: MountingFailed("mt"),
        lambda: ManualMountException("mm"),
        lambda: SmartMagnetFaultException("sm"),
        lambda: TellCommandWhileBusyException("tb"),
        lambda: TellConnectionException("tc"),
        lambda: WarningTellException("wt"),
        lambda: CriticalTellException("ct"),
        lambda: AXCFailed("ax"),
        lambda: BeamlineBusyException("bb"),
        lambda: SampleException("se"),
        lambda: AuthenticationException("ae"),
        lambda: TellCommunicationError("tce", endpoint="/x", operation="GET"),
        lambda: SmargonCommunicationError("sce", endpoint="/m", status_code=500),
        lambda: JFJochCommunicationError("jce", operation="POST"),
        lambda: AareDBCommunicationError("dbce"),
        lambda: AerotechCommunicationError("ace"),
        lambda: MagnetPositionSensorErorr(),
    ],
)
def test_exception_construction_emits_no_logs(factory, caplog):
    """Constructing an Aare exception must not emit log records.

    Logging is the responsibility of the catch site (server handler,
    best-effort wrapper, or callsite that decides to swallow). Emitting
    on construction double-logs and pollutes the WARNING-vs-ERROR split
    for best-effort flows."""
    caplog.clear()
    with caplog.at_level("DEBUG"):
        factory()
    assert caplog.records == []