WIP: Test callback structure

em_workflows/config.py

@@ -62,6 +62,10 @@ class Config:
     mrc2tif_loc = os.environ.get("MRC2TIF_LOC", "/opt/rml/imod/bin/mrc2tif")
     newstack_loc = os.environ.get("NEWSTACK_LOC", "/opt/rml/imod/bin/newstack")
 
+    # The number of max workers can to be tuned. This value is chosen through
+    # hit-and-trial based on performance gains.
+    BIOFORMATS_MAX_WORKERS = 20
+
     # environment where the app gets run - used for share selection
     env_to_share = {
         "dev": "RMLEMHedwigDev",

em_workflows/czi/flow.py

@@ -34,7 +34,7 @@ def gen_thumb(image: HedwigZarrImage, file_path: FilePath, image_name: str) -> d
 
 def gen_imageSet(file_path: FilePath) -> List:
     zarr_fp = f"{file_path.assets_dir}/{file_path.base}.zarr"
-    imageSet = list()
+    image_set = list()
     zarr_images = HedwigZarrImages(Path(zarr_fp))
     # for image_name, image in zarr_images.series():
     for k_idx, image_name in enumerate(zarr_images.get_series_keys()):
@@ -71,12 +71,12 @@ def gen_imageSet(file_path: FilePath) -> List:
             )
             assets.append(ng_asset)
         image_elt["assets"] = assets
-        imageSet.append(image_elt)
-    return imageSet
+        image_set.append(image_elt)
+    return image_set
 
 
 @task
-def bioformats_gen_zarr(file_path: FilePath):
+def generate_czi_imageset(file_path: FilePath):
     """
     TODO sort ot the num workers
     TODO, refactor this into ng.gen_zarr
@@ -91,7 +91,7 @@ def bioformats_gen_zarr(file_path: FilePath):
     log_fp = f"{file_path.working_dir}/{file_path.base}_as_zarr.log"
     cmd = [
         CZIConfig.bioformats2raw,
-        "--max_workers=19",
+        f"--max_workers={CZIConfig.BIOFORMATS_MAX_WORKERS}",
         "--overwrite",
         "--downsample-type",
         "AREA",
@@ -145,7 +145,7 @@ def find_thumb_idx(callback: List[Dict]) -> List[Dict]:
     )
     fps = utils.gen_fps(input_dir=input_dir_fp, fps_in=input_fps)
     prim_fps = utils.gen_prim_fps.map(fp_in=fps)
-    imageSets = bioformats_gen_zarr.map(file_path=fps)
+    imageSets = generate_czi_imageset.map(file_path=fps)
     callback_with_zarrs = utils.add_imageSet.map(prim_fp=prim_fps, imageSet=imageSets)
     callback_with_zarrs = find_thumb_idx(callback=callback_with_zarrs)
     filtered_callback = utils.filter_results(callback_with_zarrs)

test/conftest.py

@@ -4,8 +4,10 @@
 will be made available to all tests within this directory and do not need to be imported
 in the test files.
 """
-import pytest
+import json
 import os
+from pathlib import Path
+import pytest
 from prefect.executors import LocalExecutor
 from em_workflows.config import command_loc
 
@@ -40,4 +42,27 @@ def _mock_assets_dir(env: str) -> str:
     monkeypatch.setattr(SEMConfig, "xfalign_loc", command_loc("xfalign"))
     monkeypatch.setattr(SEMConfig, "xftoxg_loc", command_loc("xftoxg"))
     monkeypatch.setattr(SEMConfig, "convert_loc", command_loc("convert"))
-    monkeypatch.setattr(Config, "bioformats2raw", command_loc("bioformats2raw"))
+    # monkeypatch.setattr(Config, "bioformats2raw", command_loc("bioformats2raw"))
+
+
+@pytest.fixture
+def mock_callback_data(monkeypatch, tmp_path):
+    """
+    In order to assert that callback is appropriate, the callback output
+    is dumped into a temporary file. This file can be json.loaded to get
+    the callback output for tests.
+    """
+    from prefect import task
+    from em_workflows.utils import utils
+
+    filename = Path(tmp_path) / "temporary.json"
+
+    @task
+    def _mock_send_callback_body(files_elts, token, callback_url):
+        data = {"files": files_elts}
+        with open(filename, "w") as outfile:
+            json_data = json.dumps(data)
+            outfile.write(json_data)
+
+    monkeypatch.setattr(utils, "send_callback_body", _mock_send_callback_body)
+    return filename

test/test_czi.py

@@ -1,3 +1,4 @@
+import json
 import shutil
 from pathlib import Path
 from em_workflows.file_path import FilePath
@@ -17,13 +18,30 @@ def test_rechunk(mock_nfs_mount, caplog):
     from em_workflows.czi.flow import gen_imageSet
 
     input_dir = Path(
-        "/gs1/home/hedwig_dev/image_portal_workflows/image_portal_workflows/test/input_files/IF_czi/Projects/zarr_dir_2/"
+        "/gs1/home/hedwig_dev/image_portal_workflows/image_portal_workflows/test/input_files/IF_czi/Projects/zarr_dir_2/"  # noqa: E501
     )
     fp_in = Path(
-        "/gs1/home/hedwig_dev/image_portal_workflows/image_portal_workflows/test/input_files/IF_czi/Projects/zarr_dir_2/"
+        "/gs1/home/hedwig_dev/image_portal_workflows/image_portal_workflows/test/input_files/IF_czi/Projects/zarr_dir_2/"  # noqa: E501
     )
     fp = FilePath(input_dir=input_dir, fp_in=fp_in)
     d = shutil.copytree(fp_in.as_posix(), f"{fp.working_dir.as_posix()}/{fp_in.name}")
     print(d)
     zars = gen_imageSet(fp)
     print(zars)
+
+
+def test_czi_workflow_callback_structure(mock_callback_data):
+    """
+    Tests that appropriate structure exists in the callback output
+    Tests that there is no duplication in the callback output
+    """
+    from em_workflows.czi.flow import flow
+
+    state = flow.run(
+        input_dir="test/input_files/IF_czi/Projects/smaller",
+        no_api=True,
+    )
+    assert state.is_successful()
+
+    callback_output = json.loads(mock_callback_data)
+    assert "files" in callback_output