Merge branch 'Windows-Testing' into 'v3'

Fix (most) unit tests on Windows

See merge request openflexure/openflexure-microscope-server!457

src/openflexure_microscope_server/stitching.py

@@ -18,6 +18,8 @@ import labthings_fastapi as lt
 
 from openflexure_microscope_server.utilities import make_path_safe
 
+IS_WINDOWS = os.name == "nt"
+
 STITCHING_CMD = "openflexure-stitch"
 STITCHING_RESOLUTION = (820, 616)
 STITCH_TILE_SIZE = 8192
@@ -95,7 +97,8 @@ class BaseStitcher:
         self.validate_path()
 
         # The command, and the mode
-        initial_args = shlex.split(STITCHING_CMD) + ["--stitching_mode", self._mode]
+        base_args = shlex.split(STITCHING_CMD, posix=not IS_WINDOWS)
+        initial_args = base_args + ["--stitching_mode", self._mode]
         # Use float() just to really ensure that anything input is a float
         setting_args = [
             "--minimum_overlap",
@@ -181,7 +184,11 @@ class PreviewStitcher(BaseStitcher):
             except lt.exceptions.InvocationCancelledError as e:
                 with self._popen_lock:
                     if self._popen_obj is not None:
-                        self._popen_obj.send_signal(signal.SIGKILL)
+                        if IS_WINDOWS:
+                            # Windows has no SIGKILL
+                            self._popen_obj.kill()
+                        else:
+                            self._popen_obj.send_signal(signal.SIGKILL)
                 raise (e)
 
 

tests/unit_tests/test_config_utilities.py

@@ -53,13 +53,15 @@ def test_merge_patch(target, patch, outcome, err_if_enforce):
 @pytest.mark.parametrize(
     ("base_conf", "resolves_to"),
     [
-        ("/var/base.json", "/var/base.json"),
+        ("/var/base.json", os.path.normpath("/var/base.json")),
-        ("base.json", "/var/openflexure/settings/base.json"),
+        ("base.json", os.path.normpath("/var/openflexure/settings/base.json")),
-        ("./base.json", "/var/openflexure/settings/base.json"),
+        ("./base.json", os.path.normpath("/var/openflexure/settings/base.json")),
-        ("../base.json", "/var/openflexure/base.json"),
+        ("../base.json", os.path.normpath("/var/openflexure/base.json")),
         (
             "$OFM_LIB/base.json",
-            "/var/openflexure/application/openflexure-microscope-server/base.json",
+            os.path.normpath(
+                "/var/openflexure/application/openflexure-microscope-server/base.json"
+            ),
         ),
     ],
 )

tests/unit_tests/test_logging.py

@@ -3,6 +3,8 @@
 import logging
 import os
 import tempfile
+from contextlib import contextmanager
+from logging.handlers import RotatingFileHandler
 
 import pytest
 from fastapi import HTTPException
@@ -11,6 +13,25 @@ from fastapi.responses import PlainTextResponse
 from openflexure_microscope_server import logging as ofm_logging
 
 
+@contextmanager
+def tmp_logging_dir():
+    """Yield a temporary logging dir, and delete rotating file handlers before closing.
+
+    This is needed for unit tests on Windows for the temp dir to cleanup without an
+    error.
+    """
+    with tempfile.TemporaryDirectory() as tmpdir:
+        yield tmpdir
+        root_logger = logging.getLogger()
+        # Get the rotating file handlers. There should only be 1 (or none if mocked)
+        file_handlers = [
+            h for h in root_logger.handlers if isinstance(h, RotatingFileHandler)
+        ]
+        for handler in file_handlers:
+            handler.close()
+            root_logger.removeHandler(handler)
+
+
 def test_no_warnings_if_correct_permissions(caplog):
     """Check that configure_logging no warnings on normal operation.
 
@@ -22,7 +43,7 @@ def test_no_warnings_if_correct_permissions(caplog):
     """
     # Reset handler at start of test
     ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler()
-    with caplog.at_level(logging.WARNING), tempfile.TemporaryDirectory() as tmpdir:
+    with caplog.at_level(logging.WARNING), tmp_logging_dir() as tmpdir:
         ofm_logging.configure_logging(tmpdir)
         assert len(caplog.records) == 0
         with open(ofm_logging.OFM_LOG_FILE, "r", encoding="utf-8") as log_file:
@@ -40,7 +61,7 @@ def test_permission_error_raises_warning(mocker, caplog):
     )
     # Reset handler at start of test
     ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler()
-    with caplog.at_level(logging.WARNING), tempfile.TemporaryDirectory() as tmpdir:
+    with caplog.at_level(logging.WARNING), tmp_logging_dir() as tmpdir:
         ofm_logging.configure_logging(tmpdir)
         assert len(caplog.records) == 1
         # Check OFM logger is added even if the file logger couldn't be.
@@ -52,7 +73,7 @@ def test_making_log_dir(caplog):
     """Check that configure_logging will make a dir if needed."""
     # Reset handler at start of test
     ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler()
-    with caplog.at_level(logging.WARNING), tempfile.TemporaryDirectory() as tmpdir:
+    with caplog.at_level(logging.WARNING), tmp_logging_dir() as tmpdir:
         log_dir = os.path.join(tmpdir, "new_dir")
         assert not os.path.isdir(log_dir)
         ofm_logging.configure_logging(log_dir)
@@ -64,7 +85,7 @@ def test_max_logs():
     """Proclaim that only the most recent 250 logs are stored in memory."""
     # Reset handler at start of test
     ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler()
-    with tempfile.TemporaryDirectory() as tmpdir:
+    with tmp_logging_dir() as tmpdir:
         ofm_logging.configure_logging(tmpdir)
         # But I would log five hundred times.
         for i in range(500):
@@ -90,7 +111,7 @@ def test_ofm_handler_ignores_uvicorn_access():
     """
     # Reset handler at start of test
     ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler()
-    with tempfile.TemporaryDirectory() as tmpdir:
+    with tmp_logging_dir() as tmpdir:
         ofm_logging.configure_logging(tmpdir)
         logs = ofm_logging.OFM_HANDLER.log_history.split("\n")
         starting_len = len(logs)
@@ -114,7 +135,7 @@ def test_server_responses():
     """
     # Reset handler at start of test
     ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler()
-    with tempfile.TemporaryDirectory() as tmpdir:
+    with tmp_logging_dir() as tmpdir:
         ofm_logging.configure_logging(tmpdir)
         for i in range(500):
             logging.info("log %s", i)
@@ -142,7 +163,7 @@ def test_server_response_with_no_log_file():
 def test_server_response_with_no_log_dir():
     """Check that an HTTP exception is raised if the log file cannot be accessed."""
     ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler()
-    with tempfile.TemporaryDirectory() as tmpdir:
+    with tmp_logging_dir() as tmpdir:
         ofm_logging.configure_logging(tmpdir)
         for i in range(500):
             logging.info("log %s", i)
@@ -176,7 +197,7 @@ def test_uvicorn_error_only_says_error_on_error(
     error logs as <uvicorn.error>.
     """
     ofm_logging.OFM_HANDLER = ofm_logging.OFMHandler()
-    with tempfile.TemporaryDirectory() as tmpdir:
+    with tmp_logging_dir() as tmpdir:
         ofm_logging.configure_logging(tmpdir)
         log_command("Mockety mock mock!")
         with open(ofm_logging.OFM_LOG_FILE, "r", encoding="utf-8") as log_file:

tests/unit_tests/test_stitching.py

@@ -235,6 +235,7 @@ class StitchingTestThing(lt.Thing):
         """Run the final stitcher."""
         stitcher = FinalStitcher(FAKE_DIR, logger=self.logger)
         # Send in the argument HANG to mock-stitch and it just hang for 10s
+        stitcher._extra_args = ["HANG"]
         stitcher.run()
 
 

tests/unit_tests/test_system_thing.py

@@ -97,12 +97,14 @@ def test_pi_shutdown(mocker):
     # Check the shutdown command is as expected.
     assert system.SHUTDOWN_CMD == ["sudo", "shutdown", "-h", "now"]
     # Mock the shutdown command as we don't want to shutdown when running tests.
-    mocker.patch.object(system, "SHUTDOWN_CMD", new=["echo", "shutdown"])
+    mocker.patch.object(
+        system, "SHUTDOWN_CMD", new=["python", "-c", "print('shutdown')"]
+    )
     # Call shutdown on a MockPiSystem
     system_thing = create_thing_without_server(MockPiSystem)
     result = system_thing.shutdown()
 
-    # Check the result of the echo mock command was returned
+    # Check the result of the mock command was returned
     assert result.output.strip() == "shutdown"
     assert result.error.strip() == ""
 
@@ -112,12 +114,12 @@ def test_pi_reboot(mocker):
     # Check the reboot command is as expected.
     assert system.REBOOT_CMD == ["sudo", "shutdown", "-r", "now"]
     # Mock the reboot command as we don't want to reboot when running tests.
-    mocker.patch.object(system, "REBOOT_CMD", new=["echo", "restart"])
+    mocker.patch.object(system, "REBOOT_CMD", new=["python", "-c", "print('restart')"])
     # Call reboot on a MockPiSystem
     system_thing = create_thing_without_server(MockPiSystem)
     result = system_thing.reboot()
 
-    # Check the result of the echo mock command was returned
+    # Check the result of the mock command was returned
     assert result.output.strip() == "restart"
     assert result.error.strip() == ""
 

tests/unit_tests/test_version_strings.py

@@ -59,13 +59,13 @@ def _git(command: str) -> str:
 def temp_dir():
     """Return the path of a temporary directory (set as working dir)."""
     working_dir = os.getcwd()
-    try:
+    with tempfile.TemporaryDirectory() as tmpdir:
-        with tempfile.TemporaryDirectory() as tmpdir:
+        try:
             os.chdir(tmpdir)
             yield tmpdir
-    finally:
+        finally:
-        # Return to original working dir
+            # Change back to working dir before closing context manager or Windows will error
-        os.chdir(working_dir)
+            os.chdir(working_dir)
 
 
 @pytest.fixture
@@ -191,8 +191,10 @@ def test_reading_hash_from_git():
             # 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
+            # Change back to working dir before closing context manager or Windows will error
+            os.chdir(working_dir)
     finally:
-        # Return to original working dir
+        # Ensure changed to working dir
         os.chdir(working_dir)