python-murfey/tests/workflows/clem/test_align_and_merge.py at fdda3fe2bbbdbd076fd3734175c99a58155a952e · DiamondLightSource/python-murfey · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
from pathlib import Path
from unittest.mock import MagicMock

import pytest
from pytest_mock import MockerFixture

from murfey.server.ispyb import TransportManager
from murfey.util.config import MachineConfig
from murfey.workflows.clem.align_and_merge import run

# Folder and file settings
session_id = 0
instrument_name = "clem"
raw_folder = "images"
processed_folder = "processed"
visit_name = "cm12345-6"
area_name = "test_area"
series_name = "test_series"
colors = [
    "gray",
    "green",
    "red",
]
feedback_queue = "murfey_feedback"


@pytest.fixture
def processed_dir(tmp_path: Path):
    processed_dir = tmp_path / visit_name / processed_folder
    processed_dir.mkdir(parents=True, exist_ok=True)
    return processed_dir


@pytest.fixture
def image_stacks(processed_dir: Path):
    image_dir = processed_dir / area_name / series_name
    image_dir.mkdir(parents=True, exist_ok=True)

    images = [image_dir / f"{color}.tiff" for color in colors]
    for image in images:
        if not image.exists():
            image.touch()

    return images


@pytest.fixture
def metadata(processed_dir: Path):
    metadata_dir = processed_dir / area_name / series_name / "metadata"
    metadata_dir.mkdir(parents=True, exist_ok=True)

    metadata = metadata_dir / f"{series_name}.xml"
    if not metadata.exists():
        metadata.touch()

    return metadata


def test_run(
    mocker: MockerFixture,
    image_stacks: list[Path],
    metadata: Path,
    processed_dir: Path,
):
    # Construct the long series name
    series_name_long = "--".join(
        image_stacks[0].parent.relative_to(processed_dir).parts
    )

    # Create a mock tranpsort object
    mock_transport = MagicMock(spec=TransportManager)
    mock_transport.feedback_queue = feedback_queue

    # Construct a mock MachineConfig object for use within the function
    mock_machine_config = MagicMock(spec=MachineConfig)
    mock_machine_config.processed_directory_name = processed_folder
    mock_get_machine_config = mocker.patch(
        "murfey.workflows.clem.align_and_merge.get_machine_config"
    )
    mock_get_machine_config.return_value = {
        instrument_name: mock_machine_config,
    }

    # Run the function
    run(
        session_id=session_id,
        instrument_name=instrument_name,
        series_name=series_name_long,
        images=image_stacks,
        metadata=metadata,
        messenger=mock_transport,
    )

    # Construct expected recipe to be sent
    sent_recipe = {
        "recipes": ["clem-align-and-merge"],
        "parameters": {
            # Job parameters
            "series_name": series_name_long,
            "images": [str(file) for file in image_stacks],
            "metadata": str(metadata),
            # Other recipe parameters
            "session_dir": str(processed_dir.parent),
            "session_id": session_id,
            "job_name": series_name_long,
            "feedback_queue": feedback_queue,
        },
    }

    # Check that it sends the expected recipe
    mock_transport.send.assert_called_once_with(
        "processing_recipe",
        sent_recipe,
        new_connection=True,
    )