From 46bf217b40668802a61ce521fcb494d71af9dcb0 Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Wed, 12 Aug 2026 03:53:55 +0100 Subject: [PATCH 1/9] Update logger name --- src/murfey/server/api/workflow_clem.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/murfey/server/api/workflow_clem.py b/src/murfey/server/api/workflow_clem.py index d094b287a..44649f681 100644 --- a/src/murfey/server/api/workflow_clem.py +++ b/src/murfey/server/api/workflow_clem.py @@ -18,7 +18,7 @@ from murfey.server.murfey_db import murfey_db # Set up logger -logger = getLogger("murfey.server.api.clem") +logger = getLogger("murfey.server.api.workflow_clem") # Create APIRouter class object router = APIRouter( From 35c62ae6eeb0c60f3f4cd016be8aed324de10ba1 Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Wed, 12 Aug 2026 05:05:31 +0100 Subject: [PATCH 2/9] Updated the 'process_raw_lifs' API endpoint so that it constructs and sends the recipe directly instead of loading the entry point --- src/murfey/server/api/workflow_clem.py | 90 +++++++++++++++++++------- 1 file changed, 66 insertions(+), 24 deletions(-) diff --git a/src/murfey/server/api/workflow_clem.py b/src/murfey/server/api/workflow_clem.py index 44649f681..2f68118a3 100644 --- a/src/murfey/server/api/workflow_clem.py +++ b/src/murfey/server/api/workflow_clem.py @@ -16,6 +16,7 @@ import murfey.util.db as MurfeyDB from murfey.server import _transport_object from murfey.server.murfey_db import murfey_db +from murfey.util import sanitise_path # Set up logger logger = getLogger("murfey.server.api.workflow_clem") @@ -27,38 +28,79 @@ ) -class LifInfo(BaseModel): +class LifFileInfo(BaseModel): lif_file: Path @router.post("/sessions/{session_id}/process_raw_lifs") # API posts to this URL def process_raw_lifs( session_id: int, - lif_file: LifInfo, - db: Session = murfey_db, + lif_file: LifFileInfo, + murfey_db: Session = murfey_db, ): - try: - # Try and load relevant Murfey workflow - workflow: EntryPoint = list( - entry_points(group="murfey.workflows", name="clem.process_raw_lifs") - )[0] - except IndexError: - raise RuntimeError("The relevant Murfey workflow was not found") + if _transport_object is None: + logger.error("No TransportManager object was set up") + return False - # Get instrument name from the database to load the correct config file - session_row: MurfeyDB.Session = db.exec( - select(MurfeyDB.Session).where(MurfeyDB.Session.id == session_id) - ).one() - instrument_name = session_row.instrument_name - - # Pass arguments along to the correct workflow - workflow.load()( - # Match the arguments found in murfey.workflows.clem.process_raw_lifs - file=lif_file.lif_file, - root_folder="images", - session_id=session_id, - instrument_name=instrument_name, - messenger=_transport_object, + # Load the visit name from the database + try: + murfey_session = murfey_db.exec( + select(MurfeyDB.Session).where(MurfeyDB.Session.id == session_id) + ).one() + visit_name = murfey_session.visit + except Exception as e: + logger.error("Error querying session information from database", exc_info=True) + print(e) + return False + + # Find the visit directory, the raw directory name, and the job name + try: + visit_idx = lif_file.lif_file.parts.index(visit_name) + visit_dir = Path( + "/".join( + "" + if part == "/" # Replace root "/" with "" for Linux paths + else part + for part in lif_file.lif_file.parts[: visit_idx + 1] + ) + ) + raw_dir = lif_file.lif_file.parts[visit_idx + 1] + job_name = str( + (lif_file.lif_file.parent / lif_file.lif_file.stem).relative_to( + visit_dir.parent + ) + ) + except Exception: + logger.error( + "Could not determine the visit directory from LIF file " + f"{sanitise_path(lif_file.lif_file)}", + exc_info=True, + ) + return False + + # Construct recipe and submit it for processing + recipe = { + "recipes": ["clem-process-raw-lifs"], + "parameters": { + # Job parameters + "lif_file": f"{str(lif_file.lif_file)}", + "root_folder": raw_dir, + # Other recipe parameters + "session_dir": f"{str(visit_dir)}", + "session_id": session_id, + "job_name": job_name, + "feedback_queue": _transport_object.feedback_queue, + }, + } + logger.debug( + f"Submitting LIF processing request to {_transport_object.feedback_queue!r} " + "with the following recipe: \n" + f"{recipe}" + ) + _transport_object.send( + queue="processing_recipe", + message=recipe, + new_connection=True, ) return True From e7b346c48b494e10e379c82814301a8903810475 Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Wed, 12 Aug 2026 05:11:07 +0100 Subject: [PATCH 3/9] Added test for the updated 'process_raw_lifs' API endpoint --- tests/server/api/test_workflow_clem.py | 100 +++++++++++++++++++++++++ 1 file changed, 100 insertions(+) create mode 100644 tests/server/api/test_workflow_clem.py diff --git a/tests/server/api/test_workflow_clem.py b/tests/server/api/test_workflow_clem.py new file mode 100644 index 000000000..56140cec0 --- /dev/null +++ b/tests/server/api/test_workflow_clem.py @@ -0,0 +1,100 @@ +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +from pytest_mock import MockerFixture + +from murfey.server.api.workflow_clem import LifFileInfo, process_raw_lifs +from murfey.util import sanitise_path + +session_id = 1 +visit_name = "cm12345-6" + + +@pytest.mark.parametrize( + "test_params", + ( # Has transport object | DB query success | Visits match + # Successful case + (True, True, True), + # Fail cases (one False at a time) + (False, True, True), + (True, False, True), + (True, True, False), + ), +) +def test_process_raw_lifs( + mocker: MockerFixture, + tmp_path: Path, + test_params: tuple[bool, bool, bool], +): + # Unpack test params + has_transport, query_successful, visits_match = test_params + + # Mock the transport object + mock_transport = MagicMock(feedback_queue="clem") + mocker.patch( + "murfey.server.api.workflow_clem._transport_object", + mock_transport if has_transport else None, + ) + + # Mock the Murfey DB + mock_murfey_session = MagicMock( + visit=visit_name, + ) + mock_db = MagicMock() + if query_successful: + mock_db.exec.return_value.one.return_value = mock_murfey_session + else: + mock_db.exec.return_value.one.side_effect = Exception("Something went wrong") + + # Create the test LIF file + visit_dir = ( + tmp_path / "data" / "some_year" / (visit_name if visits_match else "cm12345-5") + ) + test_file = visit_dir / "images" / "SomeLifProject.lif" + lif_file = LifFileInfo(**{"lif_file": str(test_file)}) + + # Mock the logger (check what the final logs are) + mock_logger = mocker.patch("murfey.server.api.workflow_clem.logger") + + # Run the function and check that the outputs are as expected + process_raw_lifs( + session_id=session_id, + lif_file=lif_file, + murfey_db=mock_db, + ) + + if not has_transport: + mock_logger.error.assert_called_with("No TransportManager object was set up") + elif not query_successful: + mock_logger.error.assert_called_with( + "Error querying session information from database", exc_info=True + ) + mock_transport.send.assert_not_called() + elif not visits_match: + mock_logger.error.assert_called_with( + "Could not determine the visit directory from LIF file " + f"{sanitise_path(lif_file.lif_file)}", + exc_info=True, + ) + mock_transport.send.assert_not_called() + else: + # Construct the expected recipe + recipe = { + "recipes": ["clem-process-raw-lifs"], + "parameters": { + # Job parameters + "lif_file": f"{str(lif_file.lif_file)}", + "root_folder": "images", + # Other recipe parameters + "session_dir": f"{str(visit_dir)}", + "session_id": session_id, + "job_name": f"{visit_name}/images/SomeLifProject", + "feedback_queue": "clem", + }, + } + mock_transport.send.assert_called_once_with( + queue="processing_recipe", + message=recipe, + new_connection=True, + ) From 6748eefcb2d8c85ece61622c400351c31c374c0c Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Wed, 12 Aug 2026 06:42:45 +0100 Subject: [PATCH 4/9] Updated the 'process_raw_tiffs' API endpoint so that it constructs and sends the recipe directly instead of loading the entry point --- src/murfey/server/api/workflow_clem.py | 85 +++++++++++++++++++------- 1 file changed, 64 insertions(+), 21 deletions(-) diff --git a/src/murfey/server/api/workflow_clem.py b/src/murfey/server/api/workflow_clem.py index 2f68118a3..86576f414 100644 --- a/src/murfey/server/api/workflow_clem.py +++ b/src/murfey/server/api/workflow_clem.py @@ -115,31 +115,74 @@ class TIFFSeriesInfo(BaseModel): def process_raw_tiffs( session_id: int, tiff_info: TIFFSeriesInfo, - db: Session = murfey_db, + murfey_db: Session = murfey_db, ): + if _transport_object is None: + logger.error("No TransportManager object was set up") + return False + + # Load the visit name from the database try: - # Try and load relevant Murfey workflow - workflow: EntryPoint = list( - entry_points(group="murfey.workflows", name="clem.process_raw_tiffs") - )[0] - except IndexError: - raise RuntimeError("The relevant Murfey workflow was not found") + murfey_session = murfey_db.exec( + select(MurfeyDB.Session).where(MurfeyDB.Session.id == session_id) + ).one() + visit_name = murfey_session.visit + except Exception as e: + logger.error("Error querying session information from database", exc_info=True) + print(e) + return False - # Get instrument name from the database to load the correct config file - session_row: MurfeyDB.Session = db.exec( - select(MurfeyDB.Session).where(MurfeyDB.Session.id == session_id) - ).one() - instrument_name = session_row.instrument_name + # Find the visit directory, the raw directory name, and the job name + try: + tiff_file = tiff_info.tiff_files[0] + visit_idx = tiff_file.parts.index(visit_name) + visit_dir = Path( + "/".join( + "" + if part == "/" # Replace root "/" with "" for Linux paths + else part + for part in tiff_file.parts[: visit_idx + 1] + ) + ) + raw_dir = tiff_file.parts[visit_idx + 1] + job_name = str( + (tiff_file.parent / tiff_file.stem.split("--")[0]).relative_to( + visit_dir.parent + ) + ) + except Exception: + logger.error( + "Could not determine the visit directory from TIFF file " + f"{sanitise_path(tiff_file)}", + exc_info=True, + ) + return False - # Pass arguments to correct workflow - workflow.load()( - # Match the arguments found in murfey.workflows.clem.process_raw_tiffs - tiff_list=tiff_info.tiff_files, - root_folder="images", - session_id=session_id, - instrument_name=instrument_name, - metadata=tiff_info.series_metadata, - messenger=_transport_object, + # Construct recipe and submit it for processing + recipe = { + "recipes": ["clem-process-raw-tiffs"], + "parameters": { + # Job parameters + "tiff_list": "null", + "tiff_file": f"{str(tiff_file)}", + "root_folder": raw_dir, + "metadata": f"{str(tiff_info.series_metadata)}", + # Other recipe parameters + "session_dir": f"{str(visit_dir)}", + "session_id": session_id, + "job_name": job_name, + "feedback_queue": _transport_object.feedback_queue, + }, + } + logger.debug( + f"Submitting TIFF processing request to {_transport_object.feedback_queue!r} " + "with the following recipe: \n" + f"{recipe}" + ) + _transport_object.send( + queue="processing_recipe", + message=recipe, + new_connection=True, ) return True From 9e5c058bb5e9bc9cd31cd03cb5f1a9dfeff1687d Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Wed, 12 Aug 2026 06:43:02 +0100 Subject: [PATCH 5/9] Added test for the updated 'process_raw_tiffs' API endpoint --- tests/server/api/test_workflow_clem.py | 115 ++++++++++++++++++++++++- 1 file changed, 114 insertions(+), 1 deletion(-) diff --git a/tests/server/api/test_workflow_clem.py b/tests/server/api/test_workflow_clem.py index 56140cec0..9d1696c52 100644 --- a/tests/server/api/test_workflow_clem.py +++ b/tests/server/api/test_workflow_clem.py @@ -4,7 +4,12 @@ import pytest from pytest_mock import MockerFixture -from murfey.server.api.workflow_clem import LifFileInfo, process_raw_lifs +from murfey.server.api.workflow_clem import ( + LifFileInfo, + TIFFSeriesInfo, + process_raw_lifs, + process_raw_tiffs, +) from murfey.util import sanitise_path session_id = 1 @@ -98,3 +103,111 @@ def test_process_raw_lifs( message=recipe, new_connection=True, ) + + +@pytest.mark.parametrize( + "test_params", + ( # Has transport object | DB query success | Visits match + # Successful case + (True, True, True), + # Fail cases (one False at a time) + (False, True, True), + (True, False, True), + (True, True, False), + ), +) +def test_process_raw_tiffs( + mocker: MockerFixture, + tmp_path: Path, + test_params: tuple[bool, bool, bool], +): + # Unpack test params + has_transport, query_successful, visits_match = test_params + + # Mock the transport object + mock_transport = MagicMock(feedback_queue="clem") + mocker.patch( + "murfey.server.api.workflow_clem._transport_object", + mock_transport if has_transport else None, + ) + + # Mock the Murfey DB + mock_murfey_session = MagicMock( + visit=visit_name, + ) + mock_db = MagicMock() + if query_successful: + mock_db.exec.return_value.one.return_value = mock_murfey_session + else: + mock_db.exec.return_value.one.side_effect = Exception("Something went wrong") + + # Create the test LIF file + visit_dir = ( + tmp_path / "data" / "some_year" / (visit_name if visits_match else "cm12345-5") + ) + series_name = f"{visit_name}/images/grid_1/TileScan 1/Position 1" + tiff_files = [ + visit_dir + / "images" + / "grid_1" + / "TileScan 1" + / f"Position 1--Z{str(i).zfill(2)}.tif" + for i in range(10) + ] + metadata_file = ( + visit_dir / "images" / "grid_1" / "TileScan 1" / "Metadata" / "Position 1.xlif" + ) + tiff_info = TIFFSeriesInfo( + **{ + "series_name": series_name, + "tiff_files": tiff_files, + "series_metadata": metadata_file, + } + ) + + # Mock the logger (check what the final logs are) + mock_logger = mocker.patch("murfey.server.api.workflow_clem.logger") + + # Run the function and check that the outputs are as expected + process_raw_tiffs( + session_id=session_id, + tiff_info=tiff_info, + murfey_db=mock_db, + ) + + if not has_transport: + mock_logger.error.assert_called_with("No TransportManager object was set up") + elif not query_successful: + mock_logger.error.assert_called_with( + "Error querying session information from database", exc_info=True + ) + mock_transport.send.assert_not_called() + elif not visits_match: + mock_logger.error.assert_called_with( + "Could not determine the visit directory from TIFF file " + f"{sanitise_path(tiff_files[0])}", + exc_info=True, + ) + mock_transport.send.assert_not_called() + else: + # Construct the expected recipe + recipe = { + "recipes": ["clem-process-raw-tiffs"], + "parameters": { + # Job parameters + "tiff_list": "null", + "tiff_file": f"{str(tiff_files[0])}", + "root_folder": "images", + "metadata": f"{str(metadata_file)}", + # Other recipe parameters + "session_dir": f"{str(visit_dir)}", + "session_id": session_id, + "job_name": series_name, + "feedback_queue": "clem", + }, + } + mock_transport.send.assert_called_once_with( + queue="processing_recipe", + message=recipe, + new_connection=True, + ) From 0385124ddd7a162cf74648fe69294c2a6cab1047 Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Wed, 12 Aug 2026 07:47:08 +0100 Subject: [PATCH 6/9] Updated parameter names in the LIF endpoint, and added a check for empty list of TIFF files in the TIFF endpoint --- src/murfey/server/api/workflow_clem.py | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/src/murfey/server/api/workflow_clem.py b/src/murfey/server/api/workflow_clem.py index 86576f414..fd0d2e2bf 100644 --- a/src/murfey/server/api/workflow_clem.py +++ b/src/murfey/server/api/workflow_clem.py @@ -35,7 +35,7 @@ class LifFileInfo(BaseModel): @router.post("/sessions/{session_id}/process_raw_lifs") # API posts to this URL def process_raw_lifs( session_id: int, - lif_file: LifFileInfo, + lif_info: LifFileInfo, murfey_db: Session = murfey_db, ): if _transport_object is None: @@ -55,25 +55,25 @@ def process_raw_lifs( # Find the visit directory, the raw directory name, and the job name try: - visit_idx = lif_file.lif_file.parts.index(visit_name) + visit_idx = lif_info.lif_file.parts.index(visit_name) visit_dir = Path( "/".join( "" if part == "/" # Replace root "/" with "" for Linux paths else part - for part in lif_file.lif_file.parts[: visit_idx + 1] + for part in lif_info.lif_file.parts[: visit_idx + 1] ) ) - raw_dir = lif_file.lif_file.parts[visit_idx + 1] + raw_dir = lif_info.lif_file.parts[visit_idx + 1] job_name = str( - (lif_file.lif_file.parent / lif_file.lif_file.stem).relative_to( + (lif_info.lif_file.parent / lif_info.lif_file.stem).relative_to( visit_dir.parent ) ) except Exception: logger.error( "Could not determine the visit directory from LIF file " - f"{sanitise_path(lif_file.lif_file)}", + f"{sanitise_path(lif_info.lif_file)}", exc_info=True, ) return False @@ -83,7 +83,7 @@ def process_raw_lifs( "recipes": ["clem-process-raw-lifs"], "parameters": { # Job parameters - "lif_file": f"{str(lif_file.lif_file)}", + "lif_file": f"{str(lif_info.lif_file)}", "root_folder": raw_dir, # Other recipe parameters "session_dir": f"{str(visit_dir)}", @@ -120,6 +120,9 @@ def process_raw_tiffs( if _transport_object is None: logger.error("No TransportManager object was set up") return False + if not tiff_info.tiff_files: + logger.error("No TIFF files were included in the incoming message") + return False # Load the visit name from the database try: From 3c6c031e8eac3551b415ea05bf79555f5e69df1f Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Wed, 12 Aug 2026 07:47:48 +0100 Subject: [PATCH 7/9] Updated tests --- tests/server/api/test_workflow_clem.py | 44 ++++++++++++++++---------- 1 file changed, 27 insertions(+), 17 deletions(-) diff --git a/tests/server/api/test_workflow_clem.py b/tests/server/api/test_workflow_clem.py index 9d1696c52..1552ee245 100644 --- a/tests/server/api/test_workflow_clem.py +++ b/tests/server/api/test_workflow_clem.py @@ -65,7 +65,7 @@ def test_process_raw_lifs( # Run the function and check that the outputs are as expected process_raw_lifs( session_id=session_id, - lif_file=lif_file, + lif_info=lif_file, murfey_db=mock_db, ) @@ -107,22 +107,23 @@ def test_process_raw_lifs( @pytest.mark.parametrize( "test_params", - ( # Has transport object | DB query success | Visits match + ( # Has transport object | Has TIFF files | DB query success | Visits match # Successful case - (True, True, True), + (True, True, True, True), # Fail cases (one False at a time) - (False, True, True), - (True, False, True), - (True, True, False), + (False, True, True, True), + (True, False, True, True), + (True, True, False, True), + (True, True, True, False), ), ) def test_process_raw_tiffs( mocker: MockerFixture, tmp_path: Path, - test_params: tuple[bool, bool, bool], + test_params: tuple[bool, bool, bool, bool], ): # Unpack test params - has_transport, query_successful, visits_match = test_params + has_transport, has_tiffs, query_successful, visits_match = test_params # Mock the transport object mock_transport = MagicMock(feedback_queue="clem") @@ -141,19 +142,23 @@ def test_process_raw_tiffs( else: mock_db.exec.return_value.one.side_effect = Exception("Something went wrong") - # Create the test LIF file + # Create the test TIFF files visit_dir = ( tmp_path / "data" / "some_year" / (visit_name if visits_match else "cm12345-5") ) series_name = f"{visit_name}/images/grid_1/TileScan 1/Position 1" - tiff_files = [ - visit_dir - / "images" - / "grid_1" - / "TileScan 1" - / f"Position 1--Z{str(i).zfill(2)}.tif" - for i in range(10) - ] + tiff_files = ( + [ + visit_dir + / "images" + / "grid_1" + / "TileScan 1" + / f"Position 1--Z{str(i).zfill(2)}.tif" + for i in range(10) + ] + if has_tiffs + else [] + ) metadata_file = ( visit_dir / "images" / "grid_1" / "TileScan 1" / "Metadata" / "Position 1.xlif" ) @@ -177,6 +182,11 @@ def test_process_raw_tiffs( if not has_transport: mock_logger.error.assert_called_with("No TransportManager object was set up") + elif not has_tiffs: + mock_logger.error.assert_called_with( + "No TIFF files were included in the incoming message" + ) + mock_transport.send.assert_not_called() elif not query_successful: mock_logger.error.assert_called_with( "Error querying session information from database", exc_info=True From b13d3671107f7ddd0efcc8e5f9d309e010c5c4f9 Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Wed, 12 Aug 2026 08:48:12 +0100 Subject: [PATCH 8/9] Updated 'register_preprocessing_result' workflow so that it constructs and sends the 'align_and_merge' message directly instead of loading a new workflow --- .../clem/register_preprocessing_results.py | 61 ++++++++++++++----- 1 file changed, 46 insertions(+), 15 deletions(-) diff --git a/src/murfey/workflows/clem/register_preprocessing_results.py b/src/murfey/workflows/clem/register_preprocessing_results.py index 92e505174..a9506e96a 100644 --- a/src/murfey/workflows/clem/register_preprocessing_results.py +++ b/src/murfey/workflows/clem/register_preprocessing_results.py @@ -25,7 +25,6 @@ from murfey.util.processing_params import ( default_clem_processing_parameters as processing_params, ) -from murfey.workflows.clem.align_and_merge import run as run_align_and_merge logger = logging.getLogger("murfey.workflows.clem.register_preprocessing_results") @@ -556,12 +555,14 @@ def _register_grid_square( def run(message: dict, murfey_db: Session) -> dict[str, bool]: - session_id: int = ( - int(message["session_id"]) - if not isinstance(message["session_id"], int) - else message["session_id"] - ) + # Early exit if no TransportManager object is configured + if not _transport_object: + logger.error("No TransportManager object was set up") + return {"success": False, "requeue": False} + + # Parse the incoming message try: + session_id = int(message["session_id"]) if isinstance(message["result"], str): json_obj: dict = json.loads(message["result"]) result = CLEMPreprocessingResult(**json_obj) @@ -577,6 +578,10 @@ def run(message: dict, murfey_db: Session) -> dict[str, bool]: "Exception encountered when parsing TIFF preprocessing result: \n" f"{traceback.format_exc()}" ) + + # Check that output files were included + if not result.output_files: + logger.error("No files were provided in the incoming message") return {"success": False, "requeue": False} # Outer try-finally block for tidying up database-related section of function @@ -586,6 +591,8 @@ def run(message: dict, murfey_db: Session) -> dict[str, bool]: murfey_session = murfey_db.exec( select(MurfeyDB.Session).where(MurfeyDB.Session.id == session_id) ).one() + instrument_name = murfey_session.instrument_name + visit_name = murfey_session.visit except Exception: logger.error( "Exception encountered when loading Murfey session information: \n", @@ -610,8 +617,8 @@ def run(message: dict, murfey_db: Session) -> dict[str, bool]: # Register data collection group and atlas in ISPyB _register_dcg_and_atlas( session_id=session_id, - instrument_name=murfey_session.instrument_name, - visit_name=murfey_session.visit, + instrument_name=instrument_name, + visit_name=visit_name, imaging_site=clem_img_site, murfey_db=murfey_db, ) @@ -660,15 +667,39 @@ def run(message: dict, murfey_db: Session) -> dict[str, bool]: ) # Request for image alignment and processing for the requested combinations + try: + ref_file = list(result.output_files.values())[0] + visit_idx = ref_file.parts.index(visit_name) + visit_dir = Path( + "/".join( + "" + if part == "/" # Replace root "/" with "" for Linux paths + else part + for part in ref_file.parts[: visit_idx + 1] + ) + ) + except Exception: + logger.error("Could not construct visit directory", exc_info=True) + return {"success": False, "requeue": False} for image_combo in image_combos_to_process: try: - run_align_and_merge( - session_id=session_id, - instrument_name=murfey_session.instrument_name, - series_name=result.series_name, - images=image_combo, - metadata=result.metadata, - messenger=_transport_object, + _transport_object.send( + "processing_recipe", + { + "recipes": ["clem-align-and-merge"], + "parameters": { + # Job parameters + "series_name": result.series_name, + "images": [str(file) for file in image_combo], + "metadata": str(result.metadata), + # Other recipe parameters + "session_dir": str(visit_dir), + "session_id": session_id, + "job_name": result.series_name, + "feedback_queue": _transport_object.feedback_queue, + }, + }, + new_connection=True, ) except Exception: logger.error( From 3a0c13b816d3ec46898181e46f5c109fc1627a19 Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Wed, 12 Aug 2026 08:48:26 +0100 Subject: [PATCH 9/9] Updated tests --- .../test_register_preprocessing_results.py | 94 ++++++++++++------- 1 file changed, 61 insertions(+), 33 deletions(-) diff --git a/tests/workflows/clem/test_register_preprocessing_results.py b/tests/workflows/clem/test_register_preprocessing_results.py index b17d3d1b5..1c6d8bd2b 100644 --- a/tests/workflows/clem/test_register_preprocessing_results.py +++ b/tests/workflows/clem/test_register_preprocessing_results.py @@ -245,31 +245,62 @@ def test_register_grid_square(): @pytest.mark.parametrize( "test_params", - ( # Colors - (["gray"],), - (["gray", "green"],), - (["red", "green", "blue"],), - (["cyan", "magenta", "blue"],), + ( # Has transport | Colors + # Success cases + ( + True, + ["gray"], + ), + ( + True, + ["gray", "green"], + ), + ( + True, + ["red", "green", "blue"], + ), + ( + True, + ["cyan", "magenta", "blue"], + ), + # Fail cases + ( + False, + ["gray"], + ), + ( + True, + [], + ), ), ) def test_run( mocker: MockerFixture, rsync_basepath: Path, - test_params: tuple[list[str]], + test_params: tuple[bool, list[str]], ): # Unpack test params - (colors,) = test_params + has_transport, colors = test_params + + # Mock the transport object + mock_transport = MagicMock() + mocker.patch( + "murfey.workflows.clem.register_preprocessing_results._transport_object", + mock_transport if has_transport else None, + ) # Mock the MurfeyDB connection mock_murfey_session_entry = MagicMock() mock_murfey_session_entry.instrument_name = ExampleVisit.instrument_name mock_murfey_session_entry.visit = visit_name mock_murfey_db = MagicMock() - mock_murfey_db.exec().return_value.one.return_value = mock_murfey_session_entry + mock_murfey_db.exec.return_value.one.return_value = mock_murfey_session_entry # Mock the registration helper functions - mock_register_clem_series = mocker.patch( - "murfey.workflows.clem.register_preprocessing_results._register_clem_imaging_site" + mock_clem_img_site = MagicMock() + mock_register_clem_imaging_site = mocker.patch( + "murfey.workflows.clem.register_preprocessing_results._register_clem_imaging_site", + return_value=mock_clem_img_site, ) mock_register_dcg_and_atlas = mocker.patch( "murfey.workflows.clem.register_preprocessing_results._register_dcg_and_atlas" @@ -278,11 +309,6 @@ def test_run( "murfey.workflows.clem.register_preprocessing_results._register_grid_square" ) - # Mock the align and merge workflow call - mock_align_and_merge_call = mocker.patch( - "murfey.workflows.clem.register_preprocessing_results.run_align_and_merge" - ) - preprocessing_messages = generate_preprocessing_messages( rsync_basepath=rsync_basepath, session_id=ExampleVisit.murfey_session_id, @@ -290,18 +316,22 @@ def test_run( denoising_suffix="_Lng_LVCC", ) for message in preprocessing_messages: - result = run( + run( message=message, murfey_db=mock_murfey_db, ) - assert result == {"success": True} - assert mock_register_clem_series.call_count == len(preprocessing_messages) - assert mock_register_dcg_and_atlas.call_count == len(preprocessing_messages) - assert mock_register_grid_square.call_count == len(preprocessing_messages) - if ("gray" not in colors) or ("gray" in colors and len(colors) == 1): - assert mock_align_and_merge_call.call_count == len(preprocessing_messages) + if not has_transport or not colors: + mock_register_clem_imaging_site.assert_not_called() + mock_register_dcg_and_atlas.assert_not_called() + mock_register_grid_square.assert_not_called() else: - assert mock_align_and_merge_call.call_count == len(preprocessing_messages) * 3 + assert mock_register_clem_imaging_site.call_count == len(preprocessing_messages) + assert mock_register_dcg_and_atlas.call_count == len(preprocessing_messages) + assert mock_register_grid_square.call_count == len(preprocessing_messages) + if ("gray" not in colors) or ("gray" in colors and len(colors) == 1): + assert mock_transport.send.call_count == len(preprocessing_messages) + else: + assert mock_transport.send.call_count == len(preprocessing_messages) * 3 @pytest.mark.parametrize( @@ -357,25 +387,23 @@ def test_run_with_db( return_value=ispyb_db_session, ) - # Mock the align and merge workflow call - mock_align_and_merge_call = mocker.patch( - "murfey.workflows.clem.register_preprocessing_results.run_align_and_merge" - ) - # Patch the TransportManager object in the workflows called from murfey.server.ispyb import TransportManager + mock_send = mocker.patch.object(TransportManager, "send") + transport_object = TransportManager("PikaTransport") + transport_object.feedback_queue = "murfey_feedback" mocker.patch( "murfey.workflows.clem.register_preprocessing_results._transport_object", - new=TransportManager("PikaTransport"), + new=transport_object, ) mocker.patch( "murfey.workflows.register_data_collection_group._transport_object", - new=TransportManager("PikaTransport"), + new=transport_object, ) mocker.patch( "murfey.workflows.register_atlas_update._transport_object", - new=TransportManager("PikaTransport"), + new=transport_object, ) # Run the function @@ -397,9 +425,9 @@ def test_run_with_db( # Each message should call the align-and-merge workflow thrice # if gray and colour channels are both present if ("gray" not in colors) or ("gray" in colors and len(colors) == 1): - assert mock_align_and_merge_call.call_count == len(preprocessing_messages) + assert mock_send.call_count == len(preprocessing_messages) else: - assert mock_align_and_merge_call.call_count == len(preprocessing_messages) * 3 + assert mock_send.call_count == len(preprocessing_messages) * 3 # Murfey's DataCollectionGroup should have an entry murfey_dcg_search = murfey_db_session.exec(