Add a first unit test

This checks that the server can start
and perform an autofocus.
Autofocus in turn will test the camera and stage dependencies.
This commit is contained in:
Richard Bowman 2024-11-29 13:03:36 +00:00
parent bccb1ee923
commit 58066f1948
11 changed files with 22 additions and 1700 deletions

Binary file not shown.

Before

Width:  |  Height:  |  Size: 906 B

View file

@ -1,19 +0,0 @@
import pytest
from openflexure_microscope.api import utilities
@pytest.mark.parametrize(
"test_input,expected",
[
("true", True),
("True", True),
("1", True),
(True, True),
("false", False),
("somestring", False),
(False, False),
],
)
def test_get_bool(test_input, expected):
assert utilities.get_bool(test_input) == expected

View file

@ -1,43 +0,0 @@
from apispec.utils import validate_spec
import json
import jsonschema
import os
from openflexure_microscope.api.app import api_microscope, app, labthing
from openflexure_microscope.camera.base import BaseCamera
from openflexure_microscope.microscope import Microscope
from openflexure_microscope.stage.base import BaseStage
from labthings.json import encode_json
def test_app_creation():
assert labthing.app is app
def test_microscope_creation():
assert isinstance(api_microscope, Microscope)
assert isinstance(api_microscope.camera, BaseCamera)
assert isinstance(api_microscope.stage, BaseStage)
def test_openapi_valid():
assert validate_spec(labthing.spec)
def test_thing_description_valid():
# TODO: it would be nice to put this into LabThings
# First load the schema from file and check it validates
schema_fname = os.path.join(os.path.dirname(__file__), "w3c_td_schema.json")
schema = json.load(open(schema_fname, "r"), encoding="utf-8")
jsonschema.Draft7Validator.check_schema(schema)
# Build a TD dictionary
with labthing.app.test_request_context():
td_dict = labthing.thing_description.to_dict()
# Allow our LabThingsJSONEncoder to encode the RD
td_json = encode_json(td_dict)
# Decode the JSON back into a primitive dictionary
td_json_dict = json.loads(td_json)
# Validate
jsonschema.validate(instance=td_json_dict, schema=schema)

View file

@ -1,74 +0,0 @@
import os
import threading
import time
from openflexure_microscope.camera.base import FrameStream
def test_framestream_write():
d = b"openflexure"
with FrameStream() as f:
f.write(d)
assert f.getvalue() == d
def test_framestream_overwrite():
d1 = b"openflexure"
d2 = b"microscope"
with FrameStream() as f:
f.write(d1)
assert f.getvalue() == d1
f.write(d2)
assert f.getvalue() == d2
# d1 should be removed, so f.getbuffer() should only contain d2
assert f.getbuffer().nbytes == len(d2)
def test_tracker():
sizes = range(1, 100)
with FrameStream() as f:
for length in sizes:
time.sleep(0.01)
d = b"\x00" + os.urandom(length) + b"\x00"
f.write(d)
assert f.getbuffer().nbytes == len(d)
# Make sure we have one TrackerFrame for each f.write() call
assert len(f.frames) == len(sizes)
# For each tested write size
for i, frame in enumerate(f.frames):
# Make sure the recorded size is what we expect
# Should be size +2 for our padding bytes \x00
assert frame.size == sizes[i] + 2
# Make sure we have a valid time value
assert isinstance(frame.time, float)
# Make sure the recorded time increases for each frame
if i > 0:
assert f.frames[i].time > f.frames[i - 1].time
# Test FrameStream.last
assert f.last.size == sizes[-1] + 2
def test_new_frame_event():
d = b"openflexure"
def target(framestream):
time.sleep(0.01)
framestream.write(d)
with FrameStream() as f:
# Start a thread to write some data
thread = threading.Thread(target=target, args=(f,))
thread.daemon = True
thread.start()
# Make sure the stream is empty before we wait for a frame
assert not f.getvalue()
# Wait for a frame to be written
frame = f.getframe()
assert frame == d
# Make sure getframe() cleared the event
assert not f.new_frame.events[threading.get_ident()][0].is_set()

View file

@ -1,150 +0,0 @@
import os
import pytest
from freezegun import freeze_time
from openflexure_microscope.captures.capture_manager import (
BASE_CAPTURE_PATH,
TEMP_CAPTURE_PATH,
CaptureManager,
)
def test_new_manager():
m = CaptureManager()
assert BASE_CAPTURE_PATH
assert TEMP_CAPTURE_PATH
assert m.paths["default"] == BASE_CAPTURE_PATH
assert m.paths["temp"] == TEMP_CAPTURE_PATH
def test_update_settings():
m = CaptureManager()
assert m.paths["default"] == BASE_CAPTURE_PATH
assert m.paths["temp"] == TEMP_CAPTURE_PATH
new_settings = {
"paths": {"default": "BASE_CAPTURE_PATH", "temp": "TEMP_CAPTURE_PATH"}
}
m.update_settings(new_settings)
assert m.paths["default"] == "BASE_CAPTURE_PATH"
assert m.paths["temp"] == "TEMP_CAPTURE_PATH"
assert m.read_settings() == new_settings
def test__new_output():
m = CaptureManager()
out = m._new_output(True, "filename", "subfolder", "fmt")
assert out.file == os.path.join(TEMP_CAPTURE_PATH, "subfolder", "filename.fmt")
def test__new_output_generate_filename():
m = CaptureManager()
out = m._new_output(True, None, "subfolder", "fmt")
# Assert a filename was actually generated
assert out.name != ".fmt"
def test_new_image():
m = CaptureManager()
out = m.new_image()
capture_key = str(out.id)
assert capture_key in m.images.keys()
assert m.images[capture_key] is out
def test_new_video():
m = CaptureManager()
out = m.new_video()
capture_key = str(out.id)
assert capture_key in m.videos.keys()
assert m.videos[capture_key] is out
def test_image_from_id_str():
m = CaptureManager()
out = m.new_image()
capture_key = str(out.id)
assert m.image_from_id(capture_key) is out
def test_image_from_id_int():
m = CaptureManager()
out = m.new_image()
capture_key = int(out.id)
assert m.image_from_id(capture_key) is out
def test_image_from_id_uuid():
m = CaptureManager()
out = m.new_image()
capture_key = out.id
assert m.image_from_id(capture_key) is out
def test_image_from_id_invalid():
m = CaptureManager()
out = m.new_image()
capture_key = object()
with pytest.raises(TypeError):
m.image_from_id(capture_key) is out
def test_video_from_id_str():
m = CaptureManager()
out = m.new_video()
capture_key = str(out.id)
assert m.video_from_id(capture_key) is out
def test_video_from_id_int():
m = CaptureManager()
out = m.new_video()
capture_key = int(out.id)
assert m.video_from_id(capture_key) is out
def test_video_from_id_uuid():
m = CaptureManager()
out = m.new_video()
capture_key = out.id
assert m.video_from_id(capture_key) is out
def test_video_from_id_invalid():
m = CaptureManager()
out = m.new_video()
capture_key = object()
with pytest.raises(TypeError):
m.video_from_id(capture_key) is out
@freeze_time("2020-11-27 12:00:01")
def test_generate_numbered_basename():
m = CaptureManager()
for _ in range(10):
m.new_image()
generated_filenames = [obj.basename for obj in m.images.values()]
# Assert enough names were generated
assert len(generated_filenames) == 10
# Assert all names are unique
len(generated_filenames) > len(set(generated_filenames))
# Assert all filenames are in the expected form
for fname in generated_filenames:
assert fname.startswith("2020-11-27_12-00-01")
def test_context_manager():
with CaptureManager() as m:
m.new_image()
m.new_video()

View file

@ -1,232 +0,0 @@
import datetime
import io
import os
import uuid
from collections import OrderedDict
import piexif
from PIL import Image, ImageChops
from openflexure_microscope.captures import THUMBNAIL_SIZE, CaptureObject
from openflexure_microscope.captures.capture import (
build_captures_from_exif,
capture_from_path,
)
IN_DIR = "tests/images/"
OUT_DIR = "tests/images/out/"
def generate_small_image():
# Create a dummy image to serve in the stream
image = Image.new("RGB", (20, 20), color=(0, 0, 0))
return image
def generate_small_thumbnail():
# Create a dummy image to serve in the stream
image = Image.new("RGB", (20, 20), color=(0, 0, 0))
image.thumbnail(THUMBNAIL_SIZE)
return image
def generate_small_capture(filename: str):
out = os.path.join(OUT_DIR, filename)
obj = CaptureObject(out)
generate_small_image().save(obj, format="JPEG")
return obj
def check_valid_capture(obj: CaptureObject):
assert isinstance(obj.time, datetime.datetime)
assert isinstance(obj.id, uuid.UUID)
assert isinstance(obj.annotations, dict)
assert isinstance(obj.tags, list)
assert obj.file
assert obj.format
assert obj.basename
assert obj.filefolder
assert obj.name
def check_valid_openflexure_metadata(d: dict):
assert "image" in d
assert d["image"].get("id")
assert d["image"].get("name")
assert d["image"].get("time")
assert d["image"].get("format")
assert isinstance(d["image"].get("tags"), list)
assert isinstance(d["image"].get("annotations"), dict)
def test_new_capture_object():
obj = generate_small_capture("capture0.jpg")
check_valid_capture(obj)
def test_initial_metadata():
obj = generate_small_capture("capture1.jpg")
data = obj.read_full_metadata()
check_valid_openflexure_metadata(data)
def test_tags():
obj = generate_small_capture("capture2.jpg")
obj.put_tags(["foo", "bar"])
assert obj.read_full_metadata()["image"]["tags"] == ["foo", "bar"]
obj.delete_tag("foo")
assert obj.read_full_metadata()["image"]["tags"] == ["bar"]
# Delete nonexistent tag, should change nothing
obj.delete_tag("foo")
assert obj.read_full_metadata()["image"]["tags"] == ["bar"]
def test_annotations():
obj = generate_small_capture("capture3.jpg")
obj.put_annotations({"foo": "zoo", "bar": "zar"})
assert obj.read_full_metadata()["image"]["annotations"] == {
"foo": "zoo",
"bar": "zar",
}
obj.delete_annotation("foo")
assert obj.read_full_metadata()["image"]["annotations"] == {"bar": "zar"}
# Delete nonexistent annotation, should change nothing
obj.delete_annotation("foo")
assert obj.read_full_metadata()["image"]["annotations"] == {"bar": "zar"}
def test_put_metadata():
obj = generate_small_capture("capture4.jpg")
obj.put_metadata({"foo": {"bar": "zar"}})
assert obj.read_full_metadata()["foo"] == {"bar": "zar"}
def test_put_and_save():
obj = generate_small_capture("capture_all.jpg")
obj.put_and_save(
tags=["foo", "bar"],
annotations={"foo": "zoo", "bar": "zar"},
metadata={"foo": {"bar": "zar"}},
)
full_metadata = obj.read_full_metadata()
assert full_metadata["image"]["tags"] == ["foo", "bar"]
assert full_metadata["image"]["annotations"] == {"foo": "zoo", "bar": "zar"}
assert full_metadata["foo"] == {"bar": "zar"}
def test_capture_from_path_this_version():
"""Tests reloading a capture object from a file created in the working server version
"""
def _check_metadata(data: dict):
assert data["image"]["tags"] == ["foo", "bar"]
assert data["image"]["annotations"] == {"foo": "zoo", "bar": "zar"}
assert data["foo"] == {"bar": "zar"}
# Create the capture file
obj_in = generate_small_capture("capture_reload.jpg")
obj_in.put_and_save(
tags=["foo", "bar"],
annotations={"foo": "zoo", "bar": "zar"},
metadata={"foo": {"bar": "zar"}},
)
full_metadata_in = obj_in.read_full_metadata()
check_valid_openflexure_metadata(full_metadata_in)
_check_metadata(full_metadata_in)
# Reload the capture file
obj = capture_from_path(os.path.join(OUT_DIR, "capture_reload.jpg"))
check_valid_capture(obj)
full_metadata = obj.read_full_metadata()
check_valid_openflexure_metadata(full_metadata)
_check_metadata(full_metadata)
def test_capture_from_path_v280():
"""Tests reloading a capture object from a file created in server v2.8.0
"""
obj = capture_from_path(os.path.join(IN_DIR, "capture_v280.jpg"))
check_valid_capture(obj)
full_metadata = obj.read_full_metadata()
check_valid_openflexure_metadata(full_metadata)
assert full_metadata["image"]["tags"] == ["foo", "bar"]
assert full_metadata["image"]["annotations"] == {"foo": "zoo", "bar": "zar"}
assert full_metadata["foo"] == {"bar": "zar"}
def test_build_captures_from_exif():
# Load the whole tests captures directory
objs = build_captures_from_exif(IN_DIR)
# Make sure we have a valid, populated OrderedDict
assert isinstance(objs, OrderedDict)
assert len(objs) > 0
# Check each reloaded capture
for obj in objs.values():
check_valid_capture(obj)
def test_data():
# Get a reference PIL image
ref_data = generate_small_image()
# Generate a capture with the same image data as our reference
obj = generate_small_capture("capture5.jpg")
# Pull the capture data out
obj_data = Image.open(obj.data)
# Compare the images
diff = ImageChops.difference(obj_data, ref_data)
assert not diff.getbbox()
def test_binary():
# Get a reference PIL image
ref_data = generate_small_image()
# Generate a capture with the same image data as our reference
obj = generate_small_capture("capture6.jpg")
# Pull the capture data out
obj_data = Image.open(io.BytesIO(obj.binary))
# Compare the images
diff = ImageChops.difference(obj_data, ref_data)
assert not diff.getbbox()
def test_thumbnail():
# Get a reference PIL image
ref_thumb = generate_small_thumbnail()
# Generate a capture with the same image data as our reference
obj = generate_small_capture("capture7.jpg")
# Pull the generated thumbnail out
obj_thumb = Image.open(obj.thumbnail)
# Compare the images
diff = ImageChops.difference(obj_thumb, ref_thumb)
assert not diff.getbbox()
# Assert thumbnail was saved to capture file EXIF data
exif_dict = piexif.load(obj.file)
thumbnail = exif_dict.pop("thumbnail")
assert thumbnail
# Compare the images
diff = ImageChops.difference(Image.open(io.BytesIO(thumbnail)), ref_thumb)
def test_delete():
# Generate a capture with the same image data as our reference
obj = generate_small_capture("capture_del.jpg")
assert os.path.isfile(obj.file)
obj.delete()
assert not os.path.isfile(obj.file)

View file

@ -0,0 +1,22 @@
import tempfile
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from labthings_fastapi.client import ThingClient
from openflexure_microscope_server.server import ThingServer
from openflexure_microscope_server.things.camera.simulation import SimulatedCamera
from openflexure_microscope_server.things.stage.dummy import DummyStage
from openflexure_microscope_server.things.autofocus import AutofocusThing
temp_folder = tempfile.TemporaryDirectory()
server = ThingServer(temp_folder.name)
server.add_thing(SimulatedCamera(), "/camera/")
server.add_thing(DummyStage(), "/stage/")
server.add_thing(AutofocusThing(), "/autofocus/")
def test_autofocus():
with TestClient(server.app) as client:
autofocus = ThingClient.from_url("/autofocus/", client)
_ = autofocus.fast_autofocus()

View file

@ -1,57 +0,0 @@
from openflexure_microscope.api.default_extensions import scan
def test_construct_grid_raster():
grid = scan.construct_grid((0, 0), (100, 100), (3, 3), style="raster")
assert grid == [
[(0, 0), (0, 100), (0, 200)],
[(100, 0), (100, 100), (100, 200)],
[(200, 0), (200, 100), (200, 200)],
]
def test_construct_grid_snake():
grid = scan.construct_grid((0, 0), (100, 100), (3, 3), style="snake")
assert grid == [
[(0, 0), (0, 100), (0, 200)],
[(100, 200), (100, 100), (100, 0)],
[(200, 0), (200, 100), (200, 200)],
]
def test_construct_grid_spiral():
grid = scan.construct_grid((0, 0), (100, 100), (3, 0), style="spiral")
assert grid == [
[(0, 0)],
[
(0, 100),
(100, 100),
(100, 0),
(100, -100),
(0, -100),
(-100, -100),
(-100, 0),
(-100, 100),
],
[
(-100, 200),
(0, 200),
(100, 200),
(200, 200),
(200, 100),
(200, 0),
(200, -100),
(200, -200),
(100, -200),
(0, -200),
(-100, -200),
(-200, -200),
(-200, -100),
(-200, 0),
(-200, 100),
(-200, 200),
],
]
# Ensure second n_steps value makes no difference
assert grid == scan.construct_grid((0, 0), (100, 100), (3, 3), style="spiral")

View file

@ -1,59 +0,0 @@
import json
import uuid
from fractions import Fraction
import numpy as np
import pytest
from openflexure_microscope.json import JSONEncoder
@pytest.mark.parametrize(
"test_input,expected",
[
(uuid.uuid1(), 1),
(uuid.uuid3(uuid.NAMESPACE_DNS, "openflexure.org"), 3),
(uuid.uuid4(), 4),
(uuid.uuid5(uuid.NAMESPACE_DNS, "openflexure.org"), 5),
],
)
def test_encode_uuid(test_input, expected):
encoded = json.dumps(test_input, cls=JSONEncoder)
decoded = json.loads(encoded)
assert uuid.UUID(decoded).version == expected
@pytest.mark.parametrize(
"test_input",
[
Fraction(1, 2),
Fraction(1, 3),
Fraction(-27, 4),
Fraction(27, -4),
Fraction(10),
Fraction(18, 63),
Fraction(1, 1000000),
],
)
def test_encode_fraction(test_input):
encoded = json.dumps(test_input, cls=JSONEncoder)
decoded = json.loads(encoded)
assert Fraction(decoded).limit_denominator() == test_input
@pytest.mark.parametrize(
"test_input", [np.random.randint(100, dtype=np.int64) for _ in range(10)]
)
def test_encode_np_int(test_input):
encoded = json.dumps(test_input, cls=JSONEncoder)
decoded = json.loads(encoded)
assert np.int64(decoded) == test_input
@pytest.mark.parametrize(
"test_input", [np.float64(np.random.random()) for _ in range(10)]
)
def test_encode_np_float(test_input):
encoded = json.dumps(test_input, cls=JSONEncoder)
decoded = json.loads(encoded)
assert np.float64(decoded) == test_input

View file

@ -1,31 +0,0 @@
import numpy as np
from openflexure_microscope import utilities
def test_serialise_array_b64():
shape_in = (3, 2)
arr_in = np.random.randint(100, size=shape_in, dtype=np.int32)
b64_string, dtype, shape = utilities.serialise_array_b64(arr_in)
assert b64_string
assert dtype == "int32"
assert shape == shape_in
arr_out = utilities.deserialise_array_b64(b64_string, dtype, shape)
assert np.array_equal(arr_out, arr_in)
def test_ndarray_to_json():
shape_in = (3, 2)
arr_in = np.random.randint(100, size=shape_in, dtype=np.int32)
json_out = utilities.ndarray_to_json(arr_in)
arr_out = utilities.json_to_ndarray(json_out)
assert np.array_equal(arr_out, arr_in)
def test_axes_to_array():
dict_in = {"x": 1, "y": 2, "z": 3}
assert utilities.axes_to_array(dict_in) == [1, 2, 3]

File diff suppressed because it is too large Load diff