diff --git a/src/murfey/server/feedback.py b/src/murfey/server/feedback.py index f4c4ddd98..46520f7b9 100644 --- a/src/murfey/server/feedback.py +++ b/src/murfey/server/feedback.py @@ -21,7 +21,7 @@ import mrcfile import numpy as np -from gemmi import cif +from gemmi import cif # type: ignore from pipeliner.project_graph import ProjectGraph from pipeliner.star_keys import GENERAL_BLOCK, JOB_COUNTER from sqlalchemy import func @@ -2390,7 +2390,9 @@ def feedback_callback( ) # should be True except Exception: logger.warning( - "Exception encountered in server RabbitMQ callback", exc_info=True + "Exception encountered in server-RabbitMQ callback for workflow " + f"'{message.get('register')}'", + exc_info=True, ) if murfey.server._transport_object: murfey.server._transport_object.transport.nack(header, requeue=False) diff --git a/src/murfey/workflows/clem/register_preprocessing_results.py b/src/murfey/workflows/clem/register_preprocessing_results.py index d1e4049f6..9cbef23e7 100644 --- a/src/murfey/workflows/clem/register_preprocessing_results.py +++ b/src/murfey/workflows/clem/register_preprocessing_results.py @@ -9,12 +9,10 @@ import json import logging -import traceback from collections.abc import Collection from functools import cached_property -from importlib.metadata import entry_points from pathlib import Path -from typing import Literal, Optional, TypeAlias +from typing import TYPE_CHECKING, Literal, Optional, TypeAlias from pydantic import BaseModel, computed_field from sqlmodel import Session, select @@ -25,6 +23,10 @@ from murfey.util.processing_params import ( default_clem_processing_parameters as processing_params, ) +from murfey.workflows.register_data_collection_group import register_dcg + +if TYPE_CHECKING: + from murfey.server.ispyb import TransportManager logger = logging.getLogger("murfey.workflows.clem.register_preprocessing_results") @@ -241,6 +243,7 @@ def _register_dcg_and_atlas( instrument_name: str, visit_name: str, imaging_site: MurfeyDB.ImagingSite, + transport_object: TransportManager, murfey_db: Session, ): """ @@ -261,17 +264,19 @@ def _register_dcg_and_atlas( # Determine values for atlas if is_atlas := imaging_site.data_type == "atlas": - # Register using thumbnail values if they are provided - if ( + # Register using thumbnail values + if not ( imaging_site.thumbnail_path is not None and imaging_site.thumbnail_pixel_size is not None ): - atlas_name: str | None = imaging_site.thumbnail_path - atlas_pixel_size: float | None = imaging_site.thumbnail_pixel_size - # Otherwise, register the TIFF files themselves - else: - atlas_name = imaging_site.image_path - atlas_pixel_size = imaging_site.image_pixel_size + raise ValueError( + f"ImagingSIte entry for {imaging_site.image_path} missing one or more required values:\n" + " 'thumbnail_path'\n" + " 'thumbnail_pixel_size'\n" + ) + atlas_path = imaging_site.thumbnail_path + atlas_pixel_size = imaging_site.thumbnail_pixel_size + # Translate colour flags into ISPyB convention color_flags = { COLOR_FLAGS_MURFEY_TO_ISPYB[key]: getattr(imaging_site, key, 0) @@ -279,7 +284,7 @@ def _register_dcg_and_atlas( } collection_mode = imaging_site.collection_mode else: - atlas_name = "" + atlas_path = "" atlas_pixel_size = 0.0 color_flags = None collection_mode = None @@ -292,26 +297,17 @@ def _register_dcg_and_atlas( dcg_entry = dcg_search[0] # Update if current dataset is atlas and data collection group exists if is_atlas: - atlas_message = { - "session_id": session_id, - "dcgid": dcg_entry.id, - "atlas_id": dcg_entry.atlas_id, - "atlas": atlas_name, - "atlas_pixel_size": atlas_pixel_size, - "sample": dcg_entry.sample, - "color_flags": color_flags, - "collection_mode": collection_mode, - } - if entry_point_result := entry_points( - group="murfey.workflows", name="atlas_update" - ): - (workflow,) = entry_point_result - _ = workflow.load()( - message=atlas_message, - murfey_db=murfey_db, - ) - else: - logger.warning("No workflow found for 'atlas_update'") + transport_object.do_update_atlas( + atlas_id=dcg_entry.atlas_id, + atlas_image=atlas_path, + pixel_size=atlas_pixel_size, + slot=dcg_entry.sample, + # Extract optional parameters + collection_mode=collection_mode, + color_flags=color_flags, + ) + # Update atlas path on Murfey DataCollectionGroup record as well + dcg_entry.atlas = atlas_path else: # Register data collection group and placeholder for the atlas dcg_message = { @@ -322,49 +318,39 @@ def _register_dcg_and_atlas( "session_id": session_id, "tag": dcg_name, "experiment_type_id": 45, - "atlas": atlas_name, + "atlas": atlas_path, "atlas_pixel_size": atlas_pixel_size, "sample": None, "color_flags": color_flags, "collection_mode": collection_mode, } - if entry_point_result := entry_points( - group="murfey.workflows", name="data_collection_group" - ): - (workflow,) = entry_point_result - # Register grid square - _ = workflow.load()( - message=dcg_message, - murfey_db=murfey_db, + dcg_entry = register_dcg( + message=dcg_message, + murfey_db=murfey_db, + ) + if not dcg_entry: + raise RuntimeError( + f"Could not register DataCollectionGroup entry for {imaging_site.image_path}" ) - else: - logger.warning("No workflow found for 'data_collection_group'") - - # Store data collection group id in CLEM image series table - dcg_entry = murfey_db.exec( - select(MurfeyDB.DataCollectionGroup) - .where(MurfeyDB.DataCollectionGroup.session_id == session_id) - .where(MurfeyDB.DataCollectionGroup.tag == dcg_name) - ).one() - + # Update imaging site with new information imaging_site.dcg_id = dcg_entry.id imaging_site.dcg_name = dcg_entry.tag murfey_db.add(imaging_site) murfey_db.commit() + return imaging_site + def _register_grid_square( session_id: int, imaging_site: MurfeyDB.ImagingSite, + transport_object: TransportManager, murfey_db: Session, ): - # Skip this step if no transport manager object is configured - if murfey.server._transport_object is None: - logger.error("Unable to find transport manager") - return if (dcg_name := imaging_site.dcg_name) is None: - logger.warning("Current imaging site has no data collection group name") - return + raise ValueError( + f"ImagingSite entry for {imaging_site.image_path} has no data collection group name" + ) # Check if an atlas has been registered if not ( @@ -380,7 +366,7 @@ def _register_grid_square( logger.info( f"No atlas has been registered for data collection group {dcg_name!r} yet" ) - return + return imaging_site atlas_entry = atlas_results[-1] # Use the latest registered atlas # Check if there are CLEM entries to register @@ -390,7 +376,7 @@ def _register_grid_square( .where(MurfeyDB.ImagingSite.dcg_name == dcg_name) .where(MurfeyDB.ImagingSite.data_type == "grid_square") ).all(): - if ( + if not ( atlas_entry.x0 is not None and atlas_entry.x1 is not None and atlas_entry.y0 is not None @@ -398,20 +384,24 @@ def _register_grid_square( and atlas_entry.thumbnail_pixels_x is not None and atlas_entry.thumbnail_pixels_y is not None ): - atlas_width_real = atlas_entry.x1 - atlas_entry.x0 - atlas_height_real = atlas_entry.y1 - atlas_entry.y0 - else: logger.warning("Atlas entry not populated with required values") - return + return imaging_site + atlas_width_real = atlas_entry.x1 - atlas_entry.x0 + atlas_height_real = atlas_entry.y1 - atlas_entry.y0 for clem_img_site in clem_img_site_to_register: - # Register datasets using thumbnail sizes and scales - if ( - clem_img_site.x0 is not None - and clem_img_site.x1 is not None - and clem_img_site.y0 is not None - and clem_img_site.y1 is not None - ): + try: + # Register datasets using thumbnail sizes and scales + if not ( + clem_img_site.x0 is not None + and clem_img_site.x1 is not None + and clem_img_site.y0 is not None + and clem_img_site.y1 is not None + ): + logger.warning( + f"Image series {clem_img_site.site_name!r} not populated with required values" + ) + continue # Find the real coordinates of the image midpoint x_mid_real = 0.5 * (clem_img_site.x0 + clem_img_site.x1) y_mid_real = 0.5 * (clem_img_site.y0 + clem_img_site.y1) @@ -449,101 +439,112 @@ def _register_grid_square( ) or 1 ) - else: - logger.warning( - f"Image series {clem_img_site.site_name!r} not populated with required values" - ) - continue - # Populate grid square Pydantic model - grid_square_params = GridSquareParameters( - tag=dcg_name, - x_location=clem_img_site.x0, - x_location_scaled=x_mid_px, - y_location=clem_img_site.y0, - y_location_scaled=y_mid_px, - readout_area_x=clem_img_site.image_pixels_x, - readout_area_y=clem_img_site.image_pixels_y, - thumbnail_size_x=clem_img_site.thumbnail_pixels_x, - thumbnail_size_y=clem_img_site.thumbnail_pixels_y, - width=clem_img_site.image_pixels_x, - width_scaled=width_scaled, - height=clem_img_site.image_pixels_y, - height_scaled=height_scaled, - x_stage_position=0.5 * (clem_img_site.x0 + clem_img_site.x1), - y_stage_position=0.5 * (clem_img_site.y0 + clem_img_site.y1), - pixel_size=clem_img_site.image_pixel_size, - image=clem_img_site.thumbnail_path, - collection_mode=clem_img_site.collection_mode, - ) - # Construct colour flags for ISPyB - color_flags = { - ispyb_color_flags: int(getattr(clem_img_site, murfey_color_flags, 0)) - for murfey_color_flags, ispyb_color_flags in COLOR_FLAGS_MURFEY_TO_ISPYB.items() - } - # Register or update the grid square entry as required - if grid_square_entry := murfey_db.exec( - select(MurfeyDB.GridSquare) - .where(MurfeyDB.GridSquare.name == clem_img_site.id) - .where(MurfeyDB.GridSquare.tag == grid_square_params.tag) - .where(MurfeyDB.GridSquare.session_id == session_id) - ).one_or_none(): - # Update existing grid square entry on Murfey - grid_square_entry.x_location = grid_square_params.x_location - grid_square_entry.y_location = grid_square_params.y_location - grid_square_entry.x_stage_position = grid_square_params.x_stage_position - grid_square_entry.y_stage_position = grid_square_params.y_stage_position - grid_square_entry.readout_area_x = grid_square_params.readout_area_x - grid_square_entry.readout_area_y = grid_square_params.readout_area_y - grid_square_entry.thumbnail_size_x = grid_square_params.thumbnail_size_x - grid_square_entry.thumbnail_size_y = grid_square_params.thumbnail_size_y - grid_square_entry.pixel_size = grid_square_params.pixel_size - grid_square_entry.image = grid_square_params.image - - # Update existing entry on ISPyB - murfey.server._transport_object.do_update_grid_square( - grid_square_id=grid_square_entry.id, - grid_square_parameters=grid_square_params, - color_flags=color_flags, + # Populate grid square Pydantic model + grid_square_params = GridSquareParameters( + tag=dcg_name, + x_location=clem_img_site.x0, + x_location_scaled=x_mid_px, + y_location=clem_img_site.y0, + y_location_scaled=y_mid_px, + readout_area_x=clem_img_site.image_pixels_x, + readout_area_y=clem_img_site.image_pixels_y, + thumbnail_size_x=clem_img_site.thumbnail_pixels_x, + thumbnail_size_y=clem_img_site.thumbnail_pixels_y, + width=clem_img_site.image_pixels_x, + width_scaled=width_scaled, + height=clem_img_site.image_pixels_y, + height_scaled=height_scaled, + x_stage_position=0.5 * (clem_img_site.x0 + clem_img_site.x1), + y_stage_position=0.5 * (clem_img_site.y0 + clem_img_site.y1), + pixel_size=clem_img_site.image_pixel_size, + image=clem_img_site.thumbnail_path, + collection_mode=clem_img_site.collection_mode, ) - else: - # Look up data collection group for current series - dcg_entry = murfey_db.exec( - select(MurfeyDB.DataCollectionGroup) - .where(MurfeyDB.DataCollectionGroup.session_id == session_id) - .where(MurfeyDB.DataCollectionGroup.tag == grid_square_params.tag) - ).one() - # Register to ISPyB - grid_square_ispyb_result = ( - murfey.server._transport_object.do_insert_grid_square( + # Construct colour flags for ISPyB + color_flags = { + ispyb_color_flags: int( + getattr(clem_img_site, murfey_color_flags, 0) + ) + for murfey_color_flags, ispyb_color_flags in COLOR_FLAGS_MURFEY_TO_ISPYB.items() + } + # Register or update the grid square entry as required + if grid_square_entry := murfey_db.exec( + select(MurfeyDB.GridSquare) + .where(MurfeyDB.GridSquare.name == clem_img_site.id) + .where(MurfeyDB.GridSquare.tag == grid_square_params.tag) + .where(MurfeyDB.GridSquare.session_id == session_id) + ).one_or_none(): + # Update existing grid square entry on Murfey + grid_square_entry.x_location = grid_square_params.x_location + grid_square_entry.y_location = grid_square_params.y_location + grid_square_entry.x_stage_position = ( + grid_square_params.x_stage_position + ) + grid_square_entry.y_stage_position = ( + grid_square_params.y_stage_position + ) + grid_square_entry.readout_area_x = grid_square_params.readout_area_x + grid_square_entry.readout_area_y = grid_square_params.readout_area_y + grid_square_entry.thumbnail_size_x = ( + grid_square_params.thumbnail_size_x + ) + grid_square_entry.thumbnail_size_y = ( + grid_square_params.thumbnail_size_y + ) + grid_square_entry.pixel_size = grid_square_params.pixel_size + grid_square_entry.image = grid_square_params.image + + # Update existing entry on ISPyB + transport_object.do_update_grid_square( + grid_square_id=grid_square_entry.id, + grid_square_parameters=grid_square_params, + color_flags=color_flags, + ) + else: + # Look up data collection group for current series + dcg_entry = murfey_db.exec( + select(MurfeyDB.DataCollectionGroup) + .where(MurfeyDB.DataCollectionGroup.session_id == session_id) + .where( + MurfeyDB.DataCollectionGroup.tag == grid_square_params.tag + ) + ).one() + # Register to ISPyB + grid_square_ispyb_result = transport_object.do_insert_grid_square( atlas_id=dcg_entry.atlas_id, grid_square_id=clem_img_site.id, grid_square_parameters=grid_square_params, color_flags=color_flags, ) - ) - # Register to Murfey - grid_square_entry = MurfeyDB.GridSquare( - id=grid_square_ispyb_result.get("return_value", None), - name=clem_img_site.id, - session_id=session_id, - tag=grid_square_params.tag, - x_location=grid_square_params.x_location, - y_location=grid_square_params.y_location, - x_stage_position=grid_square_params.x_stage_position, - y_stage_position=grid_square_params.y_stage_position, - readout_area_x=grid_square_params.readout_area_x, - readout_area_y=grid_square_params.readout_area_y, - thumbnail_size_x=grid_square_params.thumbnail_size_x, - thumbnail_size_y=grid_square_params.thumbnail_size_y, - pixel_size=grid_square_params.pixel_size, - image=grid_square_params.image, - ) - murfey_db.add(grid_square_entry) + # Register to Murfey + grid_square_entry = MurfeyDB.GridSquare( + id=grid_square_ispyb_result.get("return_value", None), + name=clem_img_site.id, + session_id=session_id, + tag=grid_square_params.tag, + x_location=grid_square_params.x_location, + y_location=grid_square_params.y_location, + x_stage_position=grid_square_params.x_stage_position, + y_stage_position=grid_square_params.y_stage_position, + readout_area_x=grid_square_params.readout_area_x, + readout_area_y=grid_square_params.readout_area_y, + thumbnail_size_x=grid_square_params.thumbnail_size_x, + thumbnail_size_y=grid_square_params.thumbnail_size_y, + pixel_size=grid_square_params.pixel_size, + image=grid_square_params.image, + ) + murfey_db.add(grid_square_entry) - # Add grid square ID to existing CLEM image series entry - clem_img_site.grid_square_id = grid_square_entry.id - murfey_db.add(clem_img_site) + # Add grid square ID to existing CLEM image series entry + clem_img_site.grid_square_id = grid_square_entry.id + murfey_db.add(clem_img_site) + except Exception: + logger.warning( + f"Could not add/update GridSquare entry for ImagingSite {clem_img_site.site_name}", + exc_info=True, + ) + continue # Do one commit at the end murfey_db.commit() @@ -551,168 +552,108 @@ def _register_grid_square( logger.info( f"No grid squares to register for data collection group {dcg_name!r} yet" ) - return + return imaging_site def run(message: dict, murfey_db: Session) -> dict[str, bool]: # Early exit if no TransportManager object is configured if not murfey.server._transport_object: - logger.error("No TransportManager object was set up") - return {"success": False, "requeue": False} + raise RuntimeError("No TransportManager object was set up") # 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) - elif isinstance(message["result"], dict): - result = CLEMPreprocessingResult(**message["result"]) - else: - logger.error( - f"Invalid type for TIFF preprocessing result: {type(message['result'])}" - ) - return {"success": False, "requeue": False} - except Exception: - logger.error( - "Exception encountered when parsing TIFF preprocessing result: \n" - f"{traceback.format_exc()}" + session_id = int(message["session_id"]) + if isinstance(message["result"], str): + json_obj: dict = json.loads(message["result"]) + result = CLEMPreprocessingResult(**json_obj) + elif isinstance(message["result"], dict): + result = CLEMPreprocessingResult(**message["result"]) + else: + raise ValueError( + f"Invalid type for TIFF preprocessing result: {type(message['result'])}" ) # 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 - try: - try: - # Load current session from database - 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", - f"{traceback.format_exc()}", - ) - return {"success": False, "requeue": False} - try: - # Register items in Murfey database - clem_img_site = _register_clem_imaging_site( - session_id=session_id, - result=result, - murfey_db=murfey_db, - ) - except Exception: - logger.error( - "Exception encountered when registering CLEM preprocessing result for " - f"{result.series_name!r}: \n" - f"{traceback.format_exc()}" - ) - return {"success": False, "requeue": False} - try: - # Register data collection group and atlas in ISPyB - _register_dcg_and_atlas( - session_id=session_id, - instrument_name=instrument_name, - visit_name=visit_name, - imaging_site=clem_img_site, - murfey_db=murfey_db, - ) - except Exception: - # Log error but allow workflow to proceed - logger.error( - "Exception encountered when registering data collection group for CLEM workflow " - f"using {result.series_name!r}: \n" - f"{traceback.format_exc()}" - ) + raise ValueError("No files were provided in the incoming message") - try: - # Register CLEM image series as grid squares - _register_grid_square( - session_id=session_id, - imaging_site=clem_img_site, - murfey_db=murfey_db, - ) - except Exception: - # Log error but allow workflow to proceed - logger.error( - f"Exception encountered when registering grid square for {result.series_name}: \n" - f"{traceback.format_exc()}" - ) + # Load current session from database + 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 + + # Register items in Murfey database + clem_img_site = _register_clem_imaging_site( + session_id=session_id, + result=result, + murfey_db=murfey_db, + ) - # Construct list of files to use for image alignment and merging steps - image_combos_to_process = [ - list(result.output_files.values()) # Composite image of all channels - ] - if ("gray" in result.output_files.keys()) and len(result.output_files) > 1: - # Create additional fluorescent-only composite image - image_combos_to_process.append( - [ - file - for channel, file in result.output_files.items() - if channel != "gray" - ] - ) - # Create additional bright field-only image - image_combos_to_process.append( - [ - file - for channel, file in result.output_files.items() - if channel == "gray" - ] - ) + # Register data collection group and atlas in ISPyB + clem_img_site = _register_dcg_and_atlas( + session_id=session_id, + instrument_name=instrument_name, + visit_name=visit_name, + imaging_site=clem_img_site, + transport_object=murfey.server._transport_object, + murfey_db=murfey_db, + ) - # 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: - murfey.server._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": murfey.server._transport_object.feedback_queue, - }, - }, - new_connection=True, - ) - except Exception: - logger.error( - "Error requesting image alignment and merging job for " - f"{result.series_name!r} series", - exc_info=True, - ) - return {"success": False, "requeue": False} - logger.info( - "Successfully requested image alignment and merging job for " - f"{result.series_name!r} series" + # Register CLEM image series as grid squares + clem_img_site = _register_grid_square( + session_id=session_id, + imaging_site=clem_img_site, + transport_object=murfey.server._transport_object, + murfey_db=murfey_db, + ) + + # Construct list of files to use for image alignment and merging steps + image_combos_to_process = [ + list(result.output_files.values()) # Composite image of all channels + ] + if ("gray" in result.output_files.keys()) and len(result.output_files) > 1: + # Create additional fluorescent-only composite image + image_combos_to_process.append( + [file for channel, file in result.output_files.items() if channel != "gray"] + ) + # Create additional bright field-only image + image_combos_to_process.append( + [file for channel, file in result.output_files.items() if channel == "gray"] ) - return {"success": True} - finally: - murfey_db.close() + # Request for image alignment and processing for the requested combinations + 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] + ) + ) + for image_combo in image_combos_to_process: + murfey.server._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": murfey.server._transport_object.feedback_queue, + }, + }, + new_connection=True, + ) + logger.info( + "Successfully requested image alignment and merging job for " + f"{result.series_name!r} series" + ) + return {"success": True} diff --git a/tests/workflows/clem/test_register_preprocessing_results.py b/tests/workflows/clem/test_register_preprocessing_results.py index f2b6057bc..a792c8397 100644 --- a/tests/workflows/clem/test_register_preprocessing_results.py +++ b/tests/workflows/clem/test_register_preprocessing_results.py @@ -243,31 +243,13 @@ def test_register_grid_square(): "test_params", ( # Has transport | Colors # Success cases - ( - True, - ["gray"], - ), - ( - True, - ["gray", "green"], - ), - ( - True, - ["red", "green", "blue"], - ), - ( - True, - ["cyan", "magenta", "blue"], - ), + (True, ["gray"]), + (True, ["gray", "green"]), + (True, ["red", "green", "blue"]), + (True, ["cyan", "magenta", "blue"]), # Fail cases - ( - False, - ["gray"], - ), - ( - True, - [], - ), + (False, ["gray"]), + (True, []), ), ) def test_run( @@ -311,16 +293,12 @@ def test_run( colors=colors, denoising_suffix="_Lng_LVCC", ) - for message in preprocessing_messages: - run( - message=message, - murfey_db=mock_murfey_db, - ) - 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: + if has_transport and colors: + for message in preprocessing_messages: + run( + message=message, + murfey_db=mock_murfey_db, + ) 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) @@ -328,6 +306,18 @@ def test_run( assert mock_transport.send.call_count == len(preprocessing_messages) else: assert mock_transport.send.call_count == len(preprocessing_messages) * 3 + else: + errors: list[type[Exception]] = [] + if not has_transport: + errors.append(RuntimeError) + elif not colors: + errors.append(ValueError) + with pytest.raises(tuple(errors)): + for message in preprocessing_messages: + run( + message=message, + murfey_db=mock_murfey_db, + ) @pytest.mark.parametrize(