Merge branch 'proper-integration-tests' into 'v3'

Proper integration tests

Closes #480

See merge request openflexure/openflexure-microscope-server!456
This commit is contained in:
Julian Stirling 2026-01-12 21:50:38 +00:00
commit 795d145ae6
4 changed files with 150 additions and 101 deletions

View file

@ -209,14 +209,24 @@ build-docs:
- !reference [.python, rules]
# Only runs when a .py file is edited
server_integration_tests:
server_lifecycle_test:
extends: .python
stage: integration
needs:
- job: build
artifacts: true
script:
- tests/integration_tests/testfile.py
- tests/lifecycle_test/testfile.py
# Pytest integration tests, seperate from the lifecycle test
integration_tests:
stage: integration
extends: .python
script:
- pytest tests/integration_tests
# Don't report this coverage to in the Merge Request or combine with other
# coverage.
pages:
needs:

View file

@ -0,0 +1,132 @@
"""Test the server without creating a full HTTP server and socket connection.
Rather than spinning up a full uvicorn webserver for each test these tests use
the FastAPI ``TestClient`` or directly communicate with the underlying
LabThings-FastAPI code. This increases speed of testing significantly.
For tests that require a full running server see the ``lifecycle_test``
directory in the tests directory.
This file used to be called test_dummy_server when it was in the unit_test suite, but
it has been moved as it artificaially inflated coverage.
"""
import json
import os
import numpy as np
import piexif
import pytest
from PIL import Image
import labthings_fastapi as lt
from ..shared_utils.lt_test_utils import LabThingsTestEnv
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
REPO_ROOT = os.path.dirname(os.path.dirname(THIS_DIR))
SIM_CONFIG = os.path.join(REPO_ROOT, "ofm_config_simulation.json")
@pytest.fixture
def test_env():
"""Yield a server with a very basic configuration."""
with open(SIM_CONFIG, "r", encoding="utf-8") as f_obj:
config_dict = json.load(f_obj)
with LabThingsTestEnv(things=config_dict["things"]) as env:
yield env
def test_autofocus(test_env):
"""Test Fast Autofocus can run doesn't raise an exception."""
stage = test_env.get_thing_client("stage")
autofocus = test_env.get_thing_client("autofocus")
assert stage.position["z"] == 0
# Autofocus 5 times and check each ends within 500 steps
for i in range(5):
autofocus.fast_autofocus()
assert abs(stage.position["z"]) < 500, f"Autofocus failed on iteration {i}"
def test_grab_jpeg(test_env):
"""Check that grab_jpeg returns a blob that can be opened."""
camera = test_env.get_thing_client("camera")
blob = camera.grab_jpeg()
image = Image.open(blob.open())
assert image.size == (820, 616)
def test_capture_jpeg_metadata(test_env):
"""Check that the position is encoded into the image metadata."""
camera = test_env.get_thing_client("camera")
blob = camera.capture_jpeg()
image = Image.open(blob.open())
exif_dict = piexif.load(image.info["exif"])
encoded_metadata = exif_dict["Exif"][piexif.ExifIFD.UserComment]
metadata = json.loads(encoded_metadata)
assert "position" in metadata["stage"]
assert metadata["stage"]["position"] == {"x": 0, "y": 0, "z": 0}
assert image.size == (820, 616)
def test_stage(test_env):
"""Test moving the stage forwards and backwards."""
stage = test_env.get_thing_client("stage")
start = stage.position
move = {"x": 1, "y": 2, "z": 3}
move_back = {"x": -1, "y": -2, "z": -3}
stage.move_relative(**move)
pos = stage.position
assert start["x"] + move["x"] == pos["x"]
assert start["y"] + move["y"] == pos["y"]
assert start["z"] + move["z"] == pos["z"]
# Move back
stage.move_relative(**move_back)
pos = stage.position
# Check nack at start.
assert start["x"] == pos["x"]
assert start["y"] == pos["y"]
assert start["z"] == pos["z"]
def test_capture_array(test_env):
"""Capture array from simulation and check the size is as expected."""
camera = test_env.get_thing_client("camera")
array = np.asarray(camera.capture_array())
assert array.shape == (616, 820, 3)
def test_camera_stage_mapping_calibration(test_env):
"""Check that camera stage mapping runs and returns the expected result."""
camera = test_env.get_thing_client("camera")
# Remove camera settling time for speed.
camera.settling_time = 0
csm = test_env.get_thing_client("camera_stage_mapping")
stage = test_env.get_thing_client("stage")
# Check it starts uncalibrated
assert csm.calibration_required
assert csm.image_to_stage_displacement_matrix is None
assert csm.last_calibration is None
# And therefore that actions error
with pytest.raises(lt.exceptions.FailedToInvokeActionError):
csm.move_in_image_coordinates(x=10)
# Calibrate
csm.calibrate_xy()
assert not csm.calibration_required
assert csm.image_to_stage_displacement_matrix is not None
assert isinstance(csm.last_calibration, dict)
# Check it returned home
assert stage.position == {"x": 0, "y": 0, "z": 0}
csm.move_in_image_coordinates(x=10, y=0)
assert stage.position == {"x": -5, "y": 0, "z": 0}
csm_matrix = csm.image_to_stage_displacement_matrix
expected_matrix = [[0, 0.5], [-0.5, 0]]
assert isinstance(csm_matrix, list)
# Check CSM is as expected to an absolute tolerance of 1e-3
assert np.allclose(csm_matrix, expected_matrix, atol=1e-3)

View file

@ -1,8 +1,12 @@
#! /usr/bin/env python3
"""Start a server subprocess for integration tests.
"""Start a server subprocess for testing the server's life cycle.
This test spools up a server in a sub process. Connects to an MJPEG stream, runs a
couple of other tests. Most importantly it checks the server shuts down gracefully
despite the MJPEG stream being connected.
These tests are separated from unit tests to avoid inflating test coverage.
For now, this file should be run directly rather than through a test framework.
This file should be run directly rather than through a test framework.
They are designed to run on CI and should work on Linux or WSL for local debugging.
"""

View file

@ -1,97 +0,0 @@
"""Test the server without creating a full HTTP server and socket connection.
Rather than spinning up a full uvicorn webserver for each test these tests use
the FastAPI ``TestClient`` or directly communicate with the underlying
LabThings-FastAPI code. This increases speed of testing significantly.
For tests that require a full running server see the ``integration_tests``
directory in the tests directory.
"""
import json
import numpy as np
import piexif
import pytest
from PIL import Image
from ..shared_utils.lt_test_utils import LabThingsTestEnv
@pytest.fixture
def test_env():
"""Yield a server with a very basic configuration."""
thing_conf = {
"camera": {
"class": "openflexure_microscope_server.things.camera.simulation:SimulatedCamera",
"kwargs": {
"shape": (240, 320, 3),
"canvas_shape": (1000, 1500, 3),
"frame_interval": 0.01,
},
},
"stage": {
"class": "openflexure_microscope_server.things.stage.dummy:DummyStage",
"kwargs": {"step_time": 0.000001},
},
"autofocus": "openflexure_microscope_server.things.autofocus:AutofocusThing",
"camera_stage_mapping": "openflexure_microscope_server.things.camera_stage_mapping:CameraStageMapper",
}
with LabThingsTestEnv(things=thing_conf) as env:
yield env
def test_autofocus(test_env):
"""Test Fast Autofocus can run doesn't raise an exception."""
# Adjust the time for stage is 100 microseconds rather than 1 microsecond.
test_env.get_thing_by_name("stage").step_time = 0.0001
autofocus = test_env.get_thing_client("autofocus")
_ = autofocus.fast_autofocus()
def test_grab_jpeg(test_env):
"""Check that grab_jpeg returns a blob that can be opened."""
camera = test_env.get_thing_client("camera")
blob = camera.grab_jpeg()
_image = Image.open(blob.open())
def test_capture_jpeg_metadata(test_env):
"""Check that the position is encoded into the image metadata."""
camera = test_env.get_thing_client("camera")
blob = camera.capture_jpeg()
image = Image.open(blob.open())
exif_dict = piexif.load(image.info["exif"])
encoded_metadata = exif_dict["Exif"][piexif.ExifIFD.UserComment]
metadata = json.loads(encoded_metadata)
assert "position" in metadata["stage"]
def test_stage(test_env):
"""Test moving th stage forwards and backwards."""
stage = test_env.get_thing_client("stage")
start = stage.position
move = {"x": 1, "y": 2, "z": 3}
stage.move_relative(**move)
pos = stage.position
for s, m, p in zip(start.values(), move.values(), pos.values(), strict=True):
assert s + m == p
stage.move_relative(**{k: -v for k, v in move.items()})
pos = stage.position
for s, p in zip(start.values(), pos.values(), strict=True):
assert s == p
def test_capture_array(test_env):
"""Capture array from simulation and check the size is as expected."""
camera = test_env.get_thing_client("camera")
array = np.asarray(camera.capture_array())
assert array.shape == (240, 320, 3)
def test_camera_stage_mapping_calibration(test_env):
"""Check that camera stage mapping can run without an exception."""
camera = test_env.get_thing_client("camera")
camera.settling_time = 0
camera_stage_mapping = test_env.get_thing_client("camera_stage_mapping")
camera_stage_mapping.calibrate_xy()