tests: added tests for daq_server, error_codes, exception_handler and tell_client

tests/integration/daq/test_daq_server.py

@@ -0,0 +1,30 @@
+import pytest
+import os
+from fastapi.testclient import TestClient
+
+# We need to set the environment variable before importing the app
+os.environ["JWT_AAREDAQ_KEY"] = "test_key_for_integration_testing"
+
+from aare.daq.server import app
+
+@pytest.fixture
+def client():
+    # We use a TestClient to interact with the FastAPI app
+    # Note: Many endpoints require authentication and a running backend (daq, bl, etc.)
+    # For a simple integration test, we can check public endpoints.
+    return TestClient(app)
+
+@pytest.mark.integration
+def test_read_error_codes(client):
+    response = client.get("/meta/error-codes")
+    assert response.status_code == 200
+    data = response.json()
+    assert "AuthErrorCode" in data
+    assert "DAQErrorCode" in data
+
+@pytest.mark.integration
+def test_login_unauthorized(client):
+    # Testing login with invalid credentials. 
+    # The current implementation raises KeyError if user not found.
+    with pytest.raises(Exception):
+        client.post("/token", data={"username": "non_existent_user_123", "password": "bad"})

tests/unit/common/test_error_codes.py

@@ -0,0 +1,14 @@
+from aare.common.error_codes import AuthErrorCode, DAQErrorCode, error_code_help, export_error_codes
+
+def test_error_code_help_returns_string():
+    help_text = error_code_help(AuthErrorCode.AUTHENTICATION_FAILED)
+    assert isinstance(help_text, str)
+    assert len(help_text) > 0
+
+def test_error_code_help_unknown_code():
+    help_text = error_code_help("UNKNOWN_CODE")
+    assert help_text is None
+
+def test_export_error_codes_contains_known_codes():
+    exported = export_error_codes()
+    assert AuthErrorCode.AUTHENTICATION_FAILED.name in exported

tests/unit/common/test_exception_handler.py

@@ -0,0 +1,137 @@
+import pytest
+from aare.common.exception_handler import (
+    DataCollectionException, 
+    AuthenticationException, 
+    AuthErrorCode,
+    TellCommunicationError,
+    TransformationInvalidException,
+    RasterScanException,
+    LoopCenteringFailed,
+    UnmountingFailed,
+    MountingFailed,
+    ManualMountException,
+    SmartMagnetFaultException,
+    TellMountFailedException,
+    TellCommandWhileBusyException,
+    TellConnectionException,
+    WarningTellException,
+    CriticalTellException,
+    AXCFailed,
+    BeamlineBusyException,
+    SampleException,
+    UserRightsException,
+    SmargonCommunicationError,
+    JFJochCommunicationError,
+    AareDBCommunicationError,
+    AerotechCommunicationError,
+    MagnetPositionSensorErorr
+)
+
+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_mount_failed_exception():
+    exc = TellMountFailedException()
+    assert "Tell mount failed" 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)

tests/unit/devices/test_tell_client.py

@@ -0,0 +1,30 @@
+import pytest
+from unittest.mock import MagicMock
+from aare.devices.tell_client import TellClient
+from aare.devices.tell_backend import TellBackend
+
+@pytest.fixture
+def mock_backend():
+    backend = MagicMock(spec=TellBackend)
+    backend.get_state.return_value = "Ready"
+    return backend
+
+@pytest.fixture
+def mock_beamline():
+    return MagicMock()
+
+def test_get_state(mock_beamline, mock_backend):
+    client = TellClient(mock_beamline, backend=mock_backend)
+    assert client.get_state() == "Ready"
+    mock_backend.get_state.assert_called()
+
+def test_is_in_mount_position_true(mock_beamline, mock_backend):
+    mock_backend.eval.return_value = "True"
+    client = TellClient(mock_beamline, backend=mock_backend)
+    assert client.is_in_mount_position() is True
+    mock_backend.eval.assert_called_with("in_mount_position&")
+
+def test_is_in_mount_position_false(mock_beamline, mock_backend):
+    mock_backend.eval.return_value = "False"
+    client = TellClient(mock_beamline, backend=mock_backend)
+    assert client.is_in_mount_position() is False