Merge branch 'update-setting-manager' into 'v3'

Update setting manager

Closes #477, #343, and #475

See merge request openflexure/openflexure-microscope-server!321
This commit is contained in:
Julian Stirling 2025-07-14 16:47:41 +00:00
commit bed59c051b
6 changed files with 493 additions and 96 deletions

View file

@ -125,6 +125,10 @@ pytest:
stage: testing
extends: .python
script:
# Our unit tests create fake Git repositories. This doesn't work unless
# Git is configured with a name and email.
- git config --global user.name "Sir Unit of Test"
- git config --global user.email "fake@email.no"
- pytest
# --gitlab-code-quality-report=pytest-warnings.json # can restore when it installs ok
coverage: '/(?i)total.*? (100(?:\.0+)?\%|[1-9]?\d(?:\.\d+)?\%)$/'

View file

@ -1,5 +1,8 @@
[project]
name = "openflexure-microscope-server"
# Note that version needs to match a string stored in a unit test, and
# the package.json file in the webapp or the CI will fail.
version = "3.0.0-alpha1"
authors = [
{ name="Richard Bowman", email="richard.bowman@cantab.net" },

View file

@ -7,40 +7,12 @@ the server.
from collections.abc import Mapping
from socket import gethostname
from typing import Annotated, Any, MutableMapping, Optional, Sequence
from typing import Optional
from uuid import UUID, uuid4
from fastapi import Depends, HTTPException, Request
import labthings_fastapi as lt
def thing_server_from_request(request: Request) -> lt.ThingServer:
"""Wrap lt.find_thing_server."""
return lt.find_thing_server(request.app)
ThingServerDep = Annotated[lt.ThingServer, Depends(thing_server_from_request)]
def recursive_update(old_dict: MutableMapping, update: Mapping):
"""Update a dictionary recursively."""
for k, v in update.items():
if isinstance(v, Mapping):
if k in old_dict and isinstance(old_dict[k], MutableMapping):
recursive_update(old_dict[k], v)
else:
old_dict[k] = dict(**v)
else:
old_dict[k] = v
def nested_dict_get(data: dict, key: Sequence[str], create=False) -> Any:
"""Index a dict-of-dicts with a sequence of strings."""
subdict = data
for k in key:
if k not in subdict and create:
subdict[k] = {}
subdict = subdict[k]
return subdict
from openflexure_microscope_server.utilities import VersionData, robust_version_strings
class SettingsManager(lt.Thing):
@ -50,59 +22,6 @@ class SettingsManager(lt.Thing):
and the state of other Things.
"""
external_metadata = lt.ThingSetting(
initial_value={},
model=Mapping,
)
"""External metadata stored in the server's settings."""
external_metadata_in_state = lt.ThingSetting(
initial_value=[],
model=Sequence[str],
)
"""Keys from ``external_metadata`` that are included in the "state" metadata."""
@lt.thing_action
def update_external_metadata(
self, data: Mapping, key: Optional[str] = None
) -> None:
"""Add or replace keys in the external metadata.
The data supplied will be merged into the existing dictionary
recursively, i.e. if a key exists, it will be added to rather than
replaced.
If a key is supplied, we will treat ``data`` as being relative to that
key, i.e. calling this action with ``data={"a":1 }, key="foo/bar"`` is
equivalent to calling it with ``data={"foo": {"bar": {"a": 1}}}``.
"""
metadata = self.external_metadata
subdict = metadata
# If we're operating only on a part of the dict, make sure
# that key exists, and find it.
if key:
subdict = nested_dict_get(subdict, key.split("/"), create=True)
recursive_update(subdict, data)
self.external_metadata = metadata
@lt.thing_action
def delete_external_metadata(self, key: str) -> None:
"""Delete a key from the stored metadata.
The key may contain forward slashes, which are understood to separate
levels of the dictionary - i.e. ``'a/c'`` will remove the ``c`` key from a
dictionary that looks like: ``{'a': {'c': 1}, 'b': {'d': 2}}``
"""
metadata = self.external_metadata
try:
subdict = nested_dict_get(metadata, key.split("/")[:-1])
del subdict[key.split("/")[-1]]
except KeyError:
raise HTTPException(
status_code=404, detail="The specified key '{key}' was not found"
)
self.external_metadata = metadata
_microscope_id: Optional[str] = None
@lt.thing_setting
@ -122,6 +41,23 @@ class SettingsManager(lt.Thing):
"""The hostname of the microscope, as reported by its operating system."""
return gethostname()
_version_data: Optional[VersionData] = None
@lt.thing_property
def version_data(self) -> VersionData:
"""The version string and version source for the server.
The source may be a commit hash if installed from git. "TOML" if installed from
source, or "Dist" if installed from a distribution package.
If the version or its source cannot be determined an error will be Logged.
"""
# Don't save as a setting as it may change. Get the version string the first
# time the property is requested this boot.
if self._version_data is None:
self._version_data = robust_version_strings()
return self._version_data
@lt.thing_action
def get_things_state(self, metadata_getter: lt.deps.GetThingStates) -> Mapping:
"""Metadata summarising the current state of all Things in the server."""
@ -130,17 +66,9 @@ class SettingsManager(lt.Thing):
@property
def thing_state(self) -> Mapping:
"""Summary metadata describing the current state of the Thing."""
state = {
return {
"hostname": self.hostname,
"microscope-uuid": str(self.microscope_id),
"version": self.version_data.version,
"version_source": self.version_data.version_source,
}
metadata = self.external_metadata
for k in self.external_metadata_in_state:
try:
kl = k.split("/")
v = nested_dict_get(metadata, kl)
subdict = nested_dict_get(state, kl[:-1], create=True)
subdict[kl[-1]] = v
except KeyError:
continue
return state

View file

@ -1,6 +1,21 @@
"""Utility functions and classes."""
import os
import re
from threading import Thread
import logging
from importlib.metadata import version
import tomllib
from pydantic import BaseModel
LOGGER = logging.getLogger(__name__)
REPO_DIR = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
# Regex for a full commit hash
COMMIT_REGEX = re.compile(r"[0-9a-f]{40}")
# Regex for a reference in the git HEAD file. Group 1 is the path.
REF_REGEX = re.compile(r"^ref:\s(.*)$")
class ErrorCapturingThread(Thread):
@ -74,3 +89,147 @@ def _wrap_and_catch_errors(target, error_buffer, *args, **kwargs):
target(*args, **kwargs)
except BaseException as e:
error_buffer.append(e)
class VersionData(BaseModel):
"""A BaseModel containing the information about the server version.
:param version: The version string for the server. Or "Undefined" if there is an
error.
:param version_source: Either a Git commit hash, "TOML", "Dist" or in the case of
an error - "Undefined".
"""
version: str
version_source: str
def robust_version_strings() -> VersionData:
"""Return a version string and information on its source.
Returning a python version string is not without problems. If the package is
installed in an editable way, often METADATA is never updated. This provides 4
ways to provide a version string:
1. If both a Git directory and a pyproject.toml are available. Return the version
string from the toml file and the git hash from the ``.git`` directory as its
source.
2. If a pyproject.toml is available but no Git directory then this is likely an
installation from source. Return the version string from the toml file and
return the literal string "TOML" as its source.
3. If a Git directory is available but no pyproject.toml then something is very
weird - log an error. Then return "Undefined" as the version string, but still
return the Git hash as the source.
4. Finally if neither a Git directory nor a pyproject.toml are available then the
server has been installed from a distribution package (i.e. a wheel). In this
case use the standard library function ``importlib.metadata.version`` for the
version string (prepending a "v") and return "Dist" as its source.
:returns: a VersionData instance with the version string and source.
"""
project_toml_path = os.path.join(REPO_DIR, "pyproject.toml")
git_dir_path = os.path.join(REPO_DIR, ".git")
has_project_toml = os.path.isfile(project_toml_path)
has_git_dir = os.path.isdir(git_dir_path)
if has_project_toml and has_git_dir:
# .git dir and pyproject.toml available. This is a development
# installation.
ver = "v" + _get_version_from_toml(project_toml_path)
source = _get_hash_from_git_dir(git_dir_path)
elif has_project_toml:
# Only pyproject.toml available. Likely installed from source. Check
# file directly as `importlib.metadata.version` plays badly with editable
# installs.
ver = "v" + _get_version_from_toml(project_toml_path)
source = "TOML"
elif has_git_dir:
# Only .git dir, that is weird.
LOGGER.error(
"Unexpected installation configuration. Version number cannot be verified."
)
ver = "Undefined"
source = _get_hash_from_git_dir(git_dir_path)
else:
# Has neither .git nor pyproject.toml. It is probably installed from wheel.
# This will be the expected method of packaging in the future. This option
# is handled last as ``importlib.metadata.version`` can be unreliable if
# the package is no installed as a distribution package.
ver = "v" + version("openflexure_microscope_server")
source = "Dist"
return VersionData(version=ver, version_source=source)
def _get_hash_from_git_dir(git_dir_path: str) -> str:
"""Get the commit hash from a .git directory directly.
This function doesn't rely on GIT being on the PATH or even installed. It doesn't
require any packages outside the Python standard library. It directly inspects the
.git directory to get the commit hash:
* If the repository is on a detached HEAD then the hash will be in the file
.git/HEAD. A detached HEAD is could be from checking out a tag or checking out
a commit directly.
* In normal operation .git/HEAD points to a reference or "ref". This ref file
contains the hash.
:param git_dir_path: The path to the .git directory.
:returns: The hash, or "Undefined" if there is any error, errors are logged not
raised.
"""
head_path = os.path.join(git_dir_path, "HEAD")
try:
with open(head_path, "r", encoding="utf-8") as head_file:
head = head_file.read()
except IOError:
LOGGER.error("Problem opening .git/HEAD")
return "Undefined"
# The file HEAD should either be a reference or commit ID.
if match := COMMIT_REGEX.match(head):
return match.group(0)
if match := REF_REGEX.match(head):
ref_path = os.path.join(git_dir_path, match.group(1))
return _get_hash_from_git_ref(ref_path)
LOGGER.error("Unexpected format for .git/HEAD")
return "Undefined"
def _get_hash_from_git_ref(git_ref_path: str) -> str:
"""Get the commit hash from a ref in the .git directory.
For more detail see `_get_hash_from_git_dir`
:param git_ref_path: The full path to the ref file in the .git directory.
:returns: The hash, or "Undefined" if there is any error, errors are logged not
raised.
"""
try:
with open(git_ref_path, "r", encoding="utf-8") as ref_file:
ref = ref_file.read()
except IOError:
LOGGER.error("Problem opening Git branch reference.")
return "Undefined"
if ref_match := COMMIT_REGEX.match(ref):
return ref_match.group(0)
LOGGER.error("No hash found in Git ref")
return "Undefined"
def _get_version_from_toml(toml_path: str) -> str:
"""Get the version string from a pyproject.toml file.
:param toml_path: The path to the toml file.
:returns: The version string directly as reported in the toml file, or "Undefined"
if there is any error, errors are logged not raised.
"""
try:
with open(toml_path, "rb") as toml_file:
toml_dict = tomllib.load(toml_file)
return toml_dict["project"]["version"]
except (IOError, ValueError, KeyError):
LOGGER.error("Problem opening pyproject.toml")
return "Undefined"

View file

@ -0,0 +1,273 @@
"""Tests that version strings can be produced reliably."""
import subprocess
import os
import re
import tempfile
import shutil
import warnings
import logging
from hypothesis import strategies as st
from hypothesis.errors import NonInteractiveExampleWarning
import pytest
from openflexure_microscope_server import utilities
# Useful for really finding the repo dir when we mock where it is.
THIS_DIR = os.path.dirname(__file__)
TRUE_REPO_DIR = os.path.dirname(THIS_DIR)
# For explicit version checking.
VER_STRING = "3.0.0-alpha1"
# This is the regex provided by https://semver.org/
SEMVER_REGEX = re.compile(
r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)"
r"(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)"
r"(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?"
r"(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$"
)
# Use hypothesis to generate random filenames
FILE_NAME_GEN = st.from_regex(r"[a-zA-Z_-]{5,20}\.[a-z]{1,5}", fullmatch=True)
# Use hypothesis to generate random data for files
FILE_DATA_GEN = st.from_regex(r"[a-zA-Z_-]{5,20}\.[a-z]{1,5}", fullmatch=True)
def create_fake_file() -> None:
"""Create a file with a random name and random content in the working directory."""
with warnings.catch_warnings():
# Ignore hypothesis warning as we are not using for hypothesis testing but just
# as a convenient way to create random files for Git.
warnings.filterwarnings("ignore", category=NonInteractiveExampleWarning)
fname = FILE_NAME_GEN.example()
with open(fname, "w", encoding="utf-8") as file_obj:
file_obj.write(FILE_DATA_GEN.example())
def _git(command: str) -> str:
"""Run a git command in a subprocess."""
git_command = ["git"] + command.split()
# Use check=True so an error is raised if git has a problem. Only read STDOUT
result = subprocess.run(git_command, check=True, capture_output=True)
return result.stdout.decode().strip()
@pytest.fixture
def temp_dir():
"""Return the path of a temporary directory (set as working dir)."""
working_dir = os.getcwd()
try:
with tempfile.TemporaryDirectory() as tmpdir:
os.chdir(tmpdir)
yield tmpdir
finally:
# Return to original working dir
os.chdir(working_dir)
@pytest.fixture
def git_repo(temp_dir):
"""Return a git repo path (set as working dir) with couple of commits."""
# Initialise, create files and commit a couple of times
_git("init")
create_fake_file()
create_fake_file()
_git("add -A")
_git("commit -m 'message'")
create_fake_file()
create_fake_file()
_git("add -A")
_git("commit -m 'message2'")
yield temp_dir
def test_version_data_git_and_toml(mocker, git_repo):
"""Check expected version reported if git repo and pyproject.toml are available."""
# Patch the variable for this repo to the repo provided by the fixture.
mocker.patch.object(utilities, "REPO_DIR", new=git_repo)
git_hash = _git("rev-parse HEAD")
shutil.copy(os.path.join(TRUE_REPO_DIR, "pyproject.toml"), git_repo)
version_data = utilities.robust_version_strings()
# Check versions are as expected. Prepending the v to the semantic version.
assert version_data.version == "v" + VER_STRING
assert version_data.version_source == git_hash
def test_version_data_toml_only(mocker, temp_dir):
"""In the case of a pyproject.toml but not git dir, check version is reported."""
# Patch the variable for this repo to the temp dir provided by the fixture.
mocker.patch.object(utilities, "REPO_DIR", new=temp_dir)
shutil.copy(os.path.join(TRUE_REPO_DIR, "pyproject.toml"), temp_dir)
version_data = utilities.robust_version_strings()
# Check versions are as expected. Prepending the v to the semantic version.
assert version_data.version == "v" + VER_STRING
assert version_data.version_source == "TOML"
def test_version_data_git_only(mocker, git_repo, caplog):
"""In the odd case of a git repo and no pyproject.toml. Check the hash is reported.
The version string should just be reported as "Undefined". This is weird, so we
expect there to be errors logged.
"""
mocker.patch.object(utilities, "REPO_DIR", new=git_repo)
with caplog.at_level(logging.ERROR):
git_hash = _git("rev-parse HEAD")
version_data = utilities.robust_version_strings()
# Check versions are as expected. Prepending the v to the semantic version.
assert version_data.version == "Undefined"
assert version_data.version_source == git_hash
# There should be 1 error level log
assert len(caplog.records) == 1
def test_version_distribution(mocker, temp_dir):
"""Simulate a distribution package with no pyproject.toml or git directory.
Check the returned result is from importlib.
"""
mocker.patch.object(utilities, "REPO_DIR", new=temp_dir)
# As `openflexure_microscope_server.utilities` imported `version` from
# `importlib.metadata` to mock the already imported:
# `openflexure_microscope_server.utilities.version`, this is `version` from
# `importlib.metadata`
mocker.patch(
"openflexure_microscope_server.utilities.version", return_value="oobar"
)
version_data = utilities.robust_version_strings()
# Expect the version is the output from importlib.metadata.version, prepended by a
# "v". So, voobar, as we mocked the version to be oobar.!
assert version_data.version == "voobar"
assert version_data.version_source == "Dist"
def test_reading_hash_from_git():
"""Use create a Git repo and check that hash can be read.
The performs a series of directory and git operations, commented as they go.
"""
working_dir = os.getcwd()
try:
with tempfile.TemporaryDirectory() as tmpdir:
os.chdir(tmpdir)
git_dir = os.path.join(tmpdir, ".git")
# Git dir not created yet. Returns Undefined
assert utilities._get_hash_from_git_dir(git_dir) == "Undefined"
# Initialised. Should still be undefined.
_git("init")
assert utilities._get_hash_from_git_dir(git_dir) == "Undefined"
# Create some fake files and commit them.
create_fake_file()
create_fake_file()
_git("add -A")
_git("commit -m 'message'")
git_hash1 = _git("rev-parse HEAD")
# Check we read the hash the same as the git executable
assert utilities._get_hash_from_git_dir(git_dir) == git_hash1
# Create more fake files and check again.
create_fake_file()
create_fake_file()
_git("add -A")
_git("commit -m 'message2'")
git_hash2 = _git("rev-parse HEAD")
assert utilities._get_hash_from_git_dir(git_dir) == git_hash2
# Save the branch name (robust to the default branch name changing)
branch_name = _git("rev-parse --abbrev-ref HEAD")
# Check out the old commit in detached HEAD and check it is correct
_git(f"checkout {git_hash1}")
assert utilities._get_hash_from_git_dir(git_dir) == git_hash1
# Check out the branch and check commit changed back
_git(f"checkout {branch_name}")
assert utilities._get_hash_from_git_dir(git_dir) == git_hash2
finally:
# Return to original working dir
os.chdir(working_dir)
def test_reading_hash_from_broken_git_dir(git_repo):
"""Break a git dir and check the hash is "Undefined"."""
git_dir = os.path.join(git_repo, ".git")
# Break the git directory and check the hash is undefined
shutil.move(
os.path.join(git_dir, "refs"),
os.path.join(git_dir, "refs1"),
copy_function=shutil.copytree,
)
assert utilities._get_hash_from_git_dir(git_dir) == "Undefined"
# Fix it and check it works again
shutil.move(
os.path.join(git_dir, "refs1"),
os.path.join(git_dir, "refs"),
copy_function=shutil.copytree,
)
commit_hash = utilities._get_hash_from_git_dir(git_dir)
assert utilities.COMMIT_REGEX.match(commit_hash) is not None
# Break the git directory differently and check the hash is undefined
os.remove(os.path.join(git_dir, "HEAD"))
assert utilities._get_hash_from_git_dir(git_dir) == "Undefined"
def test_reading_hash_from_broken_git_ref_file(git_repo):
"""Break the git ref file and check the hash is "Undefined"."""
git_dir = os.path.join(git_repo, ".git")
branch_name = _git("rev-parse --abbrev-ref HEAD")
ref_path = os.path.join(git_dir, "refs", "heads", branch_name)
# Check commit can be read
commit_hash = utilities._get_hash_from_git_dir(git_dir)
assert utilities.COMMIT_REGEX.match(commit_hash) is not None
# Write something into the ref file
with open(ref_path, "w", encoding="utf-8") as head_file:
head_file.write("This is now broken")
# It is now undefined
assert utilities._get_hash_from_git_dir(git_dir) == "Undefined"
def test_reading_hash_from_broken_git_head_file(git_repo):
"""Break the git HEAD file and check the hash is "Undefined"."""
git_dir = os.path.join(git_repo, ".git")
head_path = os.path.join(git_dir, "HEAD")
# Check commit can be read
commit_hash = utilities._get_hash_from_git_dir(git_dir)
assert utilities.COMMIT_REGEX.match(commit_hash) is not None
# Write something into the ref file
with open(head_path, "w", encoding="utf-8") as ref_file:
ref_file.write("This is now broken")
# It is now undefined
assert utilities._get_hash_from_git_dir(git_dir) == "Undefined"
def test_reading_version_from_toml():
"""Check version string can be read from this project's TOML file.
As we only expect to ever read our own TOML file there is not a need to do
excessive testing on different generated TOML files.
"""
project_toml_path = os.path.join(utilities.REPO_DIR, "pyproject.toml")
ver_string = utilities._get_version_from_toml(project_toml_path)
# Check version does follow semantic versioning.
assert SEMVER_REGEX.match(ver_string)
# Explicit check
assert ver_string == VER_STRING
def test_error_reading_version_from_toml():
"""Check Undefined is returned if error reading TOML file."""
# Just give it the wrong path.
project_toml_path = os.path.join(utilities.REPO_DIR, "pyppproject.toml")
ver_string = utilities._get_version_from_toml(project_toml_path)
# Explicit check
assert ver_string == "Undefined"

View file

@ -25,7 +25,8 @@
<div>
<b>Server Version:</b> <br />
v{{ app_version() }}
{{ version }}<br />
({{ version_source }})
</div>
<hr />
@ -66,11 +67,40 @@ export default {
name: "StatusPane",
components: { ActionButton },
data: function() {
return {
version: undefined,
version_source: undefined
}
},
async mounted() {
let version_data = await this.readThingProperty(
"settings",
"version_data"
);
this.version = version_data.version;
this.version_source = this.truncate(version_data.version_source);
},
computed: {
things: function() {
return this.$store.getters["wot/thingDescriptions"];
}
},
methods: {
truncate(string, max_length=15) {
if (!(typeof string === 'string' || string instanceof String)) {
return string;
}
if (string.length <= max_length) {
return string;
}
return string.slice(0, max_length - 3) + "...";
}
}
};
</script>