Do not allow rsync destination to be empty (#718)

The rsyncer destination determination was very out-of-date and potentially caused a problem on Krios 3.
This updates the logic and adds some tests.

Author: stephen-riggs

src/murfey/client/analyser.py

@@ -21,6 +21,7 @@
 from murfey.client.contexts.spa_metadata import SPAMetadataContext
 from murfey.client.contexts.tomo import TomographyContext
 from murfey.client.contexts.tomo_metadata import TomographyMetadataContext
+from murfey.client.destinations import find_longest_data_directory
 from murfey.client.instance_environment import MurfeyInstanceEnvironment
 from murfey.client.rsync import RSyncerUpdate, TransferResult
 from murfey.util.client import Observer, get_machine_config_client
@@ -394,12 +395,8 @@ def _xml_file(self, data_file: Path) -> Path:
             return data_file.with_suffix(".xml")
         file_name = f"{'_'.join(p for p in data_file.stem.split('_')[:-1])}.xml"
         data_directories = self._murfey_config.get("data_directories", [])
-        for dd in data_directories:
-            if str(data_file).startswith(dd):
-                base_dir = Path(dd).absolute()
-                mid_dir = data_file.relative_to(base_dir).parent
-                break
-        else:
+        base_dir, mid_dir = find_longest_data_directory(data_file, data_directories)
+        if not base_dir:
             return data_file.with_suffix(".xml")
         return base_dir / self._environment.visit / mid_dir / file_name
 

src/murfey/client/contexts/spa.py

@@ -8,6 +8,7 @@
 import xmltodict
 
 from murfey.client.context import Context, ProcessingParameter
+from murfey.client.destinations import find_longest_data_directory
 from murfey.client.instance_environment import (
     MovieTracker,
     MurfeyID,
@@ -51,12 +52,8 @@ def _file_transferred_to(
 def _grid_square_metadata_file(
     f: Path, data_directories: List[Path], visit: str, grid_square: int
 ) -> Path:
-    for dd in data_directories:
-        if str(f).startswith(str(dd)):
-            base_dir = dd.absolute()
-            mid_dir = f.relative_to(base_dir).parent
-            break
-    else:
+    base_dir, mid_dir = find_longest_data_directory(f, data_directories)
+    if not base_dir:
         raise ValueError(f"Could not determine grid square metadata path for {f}")
     metadata_file = (
         base_dir

src/murfey/client/destinations.py

@@ -2,88 +2,92 @@
 
 import logging
 from pathlib import Path
-from typing import Dict
 
-from murfey.client.analyser import Analyser
 from murfey.client.instance_environment import (
     MurfeyInstanceEnvironment,
     global_env_lock,
 )
+from murfey.util import secure_path
 from murfey.util.client import capture_get, capture_post
 
 logger = logging.getLogger("murfey.client.destinations")
 
 
+def find_longest_data_directory(
+    match_path: Path, data_directories: list[str] | list[Path]
+):
+    """
+    Determine the longest path in the data_directories list
+    which the match path is relative to
+    """
+    base_dir: Path | None = None
+    mid_dir: Path | None = None
+    for dd in data_directories:
+        dd_base = str(Path(dd).absolute())
+        if str(match_path).startswith(str(dd)) and len(dd_base) > len(str(base_dir)):
+            base_dir = Path(dd_base)
+            mid_dir = match_path.absolute().relative_to(Path(base_dir)).parent
+    return base_dir, mid_dir
+
+
 def determine_default_destination(
     visit: str,
     source: Path,
     destination: str,
     environment: MurfeyInstanceEnvironment,
-    analysers: Dict[Path, Analyser],
     token: str,
     touch: bool = False,
     extra_directory: str = "",
-    include_mid_path: bool = True,
     use_suggested_path: bool = True,
 ) -> str:
+    if not destination or not visit:
+        raise ValueError(f"No destination ({destination}) or visit ({visit}) supplied")
     machine_data = capture_get(
         base_url=str(environment.url.geturl()),
         router_name="session_control.router",
         function_name="machine_info_by_instrument",
         token=token,
         instrument_name=environment.instrument_name,
     ).json()
-    _default = ""
-    if environment.processing_only_mode and environment.sources:
-        logger.info(f"Processing only mode with sources {environment.sources}")
-        _default = str(environment.sources[0].absolute()) or str(Path.cwd())
-    elif machine_data.get("data_directories"):
-        for data_dir in machine_data["data_directories"]:
-            if source.absolute() == Path(data_dir).absolute():
-                _default = f"{destination}/{visit}"
-                break
+    base_path, mid_path = find_longest_data_directory(
+        source, machine_data.get("data_directories", [])
+    )
+    if not base_path:
+        raise ValueError(f"No data directory found for {source}")
+    if source.absolute() == base_path.absolute():
+        raise ValueError(
+            f"Source is the same as the base path {secure_path(source.absolute())}"
+        )
+
+    _default = f"{destination}/{visit}/{source.name}"
+    if use_suggested_path:
+        with global_env_lock:
+            if source.name == "Images-Disc1":
+                source_name = source.parent.name
+            elif source.name.startswith("Sample"):
+                source_name = f"{source.parent.name}_{source.name}"
+            else:
+                source_name = source.name
+            if environment.destination_registry.get(source_name):
+                _default = environment.destination_registry[source_name]
             else:
-                try:
-                    mid_path = source.absolute().relative_to(Path(data_dir).absolute())
-                    if use_suggested_path:
-                        with global_env_lock:
-                            if source.name == "Images-Disc1":
-                                source_name = source.parent.name
-                            elif source.name.startswith("Sample"):
-                                source_name = f"{source.parent.name}_{source.name}"
-                            else:
-                                source_name = source.name
-                            if environment.destination_registry.get(source_name):
-                                _default = environment.destination_registry[source_name]
-                            else:
-                                suggested_path_response = capture_post(
-                                    base_url=str(environment.url.geturl()),
-                                    router_name="file_io_instrument.router",
-                                    function_name="suggest_path",
-                                    token=token,
-                                    visit_name=visit,
-                                    session_id=environment.murfey_session,
-                                    data={
-                                        "base_path": f"{destination}/{visit}/{mid_path.parent if include_mid_path else ''}/raw",
-                                        "touch": touch,
-                                        "extra_directory": extra_directory,
-                                    },
-                                )
-                                if suggested_path_response is None:
-                                    raise RuntimeError("Murfey server is unreachable")
-                                _default = suggested_path_response.json().get(
-                                    "suggested_path"
-                                )
-                                environment.destination_registry[source_name] = _default
-                    else:
-                        _default = f"{destination}/{visit}/{mid_path if include_mid_path else source.name}"
-                    break
-                except (ValueError, KeyError):
-                    _default = ""
-        else:
-            _default = ""
-    else:
-        _default = f"{destination}/{visit}"
+                suggested_path_response = capture_post(
+                    base_url=str(environment.url.geturl()),
+                    router_name="file_io_instrument.router",
+                    function_name="suggest_path",
+                    token=token,
+                    visit_name=visit,
+                    session_id=environment.murfey_session,
+                    data={
+                        "base_path": f"{destination}/{visit}/raw",
+                        "touch": touch,
+                        "extra_directory": extra_directory,
+                    },
+                )
+                if suggested_path_response is None:
+                    raise RuntimeError("Murfey server is unreachable")
+                _default = suggested_path_response.json().get("suggested_path")
+                environment.destination_registry[source_name] = _default
     return (
         _default + f"/{extra_directory}"
         if not _default.endswith("/")

src/murfey/client/instance_environment.py

@@ -46,7 +46,6 @@ class MurfeyInstanceEnvironment(BaseModel):
     tilt_angles: Dict[str, List[List[str]]] = {}
     movie_counters: Dict[str, itertools.count] = {}
     visit: str = ""
-    processing_only_mode: bool = False
     dose_per_frame: Optional[float] = None
     gain_ref: Optional[str] = None
     symmetry: Optional[str] = None

src/murfey/client/multigrid_control.py

@@ -77,7 +77,6 @@ def __post_init__(self):
             symmetry=self.data_collection_parameters.get("symmetry"),
             eer_fractionation=self.data_collection_parameters.get("eer_fractionation"),
             instrument_name=self.instrument_name,
-            # processing_only_mode=server_routing_prefix_found,
         )
         self._machine_config = get_machine_config_client(
             str(self._environment.url.geturl()),
@@ -236,7 +235,6 @@ def _start_rsyncer_multigrid(
         self,
         source: Path,
         extra_directory: str = "",
-        include_mid_path: bool = True,
         use_suggested_path: bool = True,
         destination_overrides: Optional[Dict[Path, str]] = None,
         remove_files: bool = False,
@@ -274,11 +272,9 @@ def _start_rsyncer_multigrid(
                     source,
                     self._environment.default_destinations[source],
                     self._environment,
-                    self.analysers or {},
                     self.token,
                     touch=True,
                     extra_directory=extra_directory,
-                    include_mid_path=include_mid_path,
                     use_suggested_path=use_suggested_path,
                 )
         self._environment.sources.append(source)

src/murfey/client/watchdir_multigrid.py

@@ -55,7 +55,6 @@ def _handle_metadata(self, directory: Path, extra_directory: str):
         self.notify(
             directory,
             extra_directory=extra_directory,
-            include_mid_path=False,
             analyse=self._analyse,
             limited=True,
             tag="metadata",
@@ -71,7 +70,6 @@ def _handle_fractions(self, directory: Path, first_loop: bool):
                 # This allows you to avoid triggering processing again if Murfey is restarted
                 self.notify(
                     d02,
-                    include_mid_path=False,
                     remove_files=True,
                     analyse=(
                         not (first_loop and self._skip_existing_processing)
@@ -90,7 +88,6 @@ def _handle_fractions(self, directory: Path, first_loop: bool):
             ):
                 self.notify(
                     directory,
-                    include_mid_path=False,
                     analyse=(
                         not (first_loop and self._skip_existing_processing)
                         if self._analyse
@@ -108,7 +105,6 @@ def _process(self):
                     if d.is_dir() and d not in self._seen_dirs:
                         self.notify(
                             d,
-                            include_mid_path=False,
                             use_suggested_path=False,
                             analyse=(
                                 (

tests/client/test_destinations.py

@@ -0,0 +1,136 @@
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+from murfey.client.destinations import determine_default_destination
+
+source_list = [
+    ["X:/DoseFractions/cm12345-6/Supervisor", "Supervisor", True, "extra_name"],
+    ["X:/DoseFractions/Supervisor/Images-Disc1", "Supervisor", False, ""],
+    [
+        "X:/DoseFractions/DATA/Supervisor/Sample1",
+        "Supervisor_Sample1",
+        False,
+        "extra_name",
+    ],
+]
+
+
+@mock.patch("murfey.client.destinations.capture_get")
+@mock.patch("murfey.client.destinations.capture_post")
+@pytest.mark.parametrize("sources", source_list)
+def test_determine_default_destinations_suggested_path(mock_post, mock_get, sources):
+    mock_environment = mock.Mock()
+    mock_environment.murfey_session = 2
+    mock_environment.instrument_name = "m01"
+    mock_environment.destination_registry = {}
+
+    source, source_name, touch, extra_directory = sources
+
+    mock_get().json.return_value = {
+        "data_directories": ["X:/DoseFractions", "X:/DoseFractions/DATA"]
+    }
+    mock_post().json.return_value = {"suggested_path": "/base_path/2025/cm12345-6/raw"}
+
+    destination = determine_default_destination(
+        visit="cm12345-6",
+        source=Path(source),
+        destination="2025",
+        environment=mock_environment,
+        token="token",
+        touch=touch,
+        extra_directory=extra_directory,
+        use_suggested_path=True,
+    )
+    mock_get.assert_any_call(
+        base_url=str(mock_environment.url.geturl()),
+        router_name="session_control.router",
+        function_name="machine_info_by_instrument",
+        token="token",
+        instrument_name="m01",
+    )
+    mock_post.assert_any_call(
+        base_url=str(mock_environment.url.geturl()),
+        router_name="file_io_instrument.router",
+        function_name="suggest_path",
+        token="token",
+        visit_name="cm12345-6",
+        session_id=2,
+        data={
+            "base_path": "2025/cm12345-6/raw",
+            "touch": touch,
+            "extra_directory": extra_directory,
+        },
+    )
+
+    assert destination == f"/base_path/2025/cm12345-6/raw/{extra_directory}"
+    assert (
+        mock_environment.destination_registry.get(source_name)
+        == "/base_path/2025/cm12345-6/raw"
+    )
+
+
+@mock.patch("murfey.client.destinations.capture_get")
+@pytest.mark.parametrize("sources", source_list)
+def test_determine_default_destinations_skip_suggested(mock_get, sources):
+    mock_environment = mock.Mock()
+    mock_environment.murfey_session = 2
+    mock_environment.instrument_name = "m01"
+    mock_environment.destination_registry = {}
+
+    source, source_name, touch, extra_directory = sources
+
+    mock_get().json.return_value = {
+        "data_directories": ["X:/DoseFractions", "X:/DoseFractions/DATA"]
+    }
+
+    destination = determine_default_destination(
+        visit="cm12345-6",
+        source=Path(source),
+        destination="2025",
+        environment=mock_environment,
+        token="token",
+        touch=touch,
+        extra_directory=extra_directory,
+        use_suggested_path=False,
+    )
+    mock_get.assert_any_call(
+        base_url=str(mock_environment.url.geturl()),
+        router_name="session_control.router",
+        function_name="machine_info_by_instrument",
+        token="token",
+        instrument_name="m01",
+    )
+
+    assert destination == f"2025/cm12345-6/{Path(source).name}/{extra_directory}"
+
+
+parameter_list_fail_cases = [
+    ["X:/DoseFractions/cm12345-6", "", "2025", "X:/DoseFractions"],
+    ["X:/DoseFractions/cm12345-6", "cm12345-6", "", "X:/DoseFractions"],
+    ["X:/DoseFractions", "cm12345-6", "2025", "X:/DoseFractions"],
+    ["X:/cm12345-6", "cm12345-6", "2025", "X:/DoseFractions"],
+]
+
+
+@mock.patch("murfey.client.destinations.capture_get")
+@pytest.mark.parametrize("destination_params", parameter_list_fail_cases)
+def test_determine_default_destinations_failures(mock_get, destination_params):
+    """
+    Test failure of the following cases:
+    No visit, no destination, source = default, source not in default
+    """
+    mock_get().json.return_value = {
+        "data_directories": ["X:/DoseFractions", "X:/DoseFractions/DATA"]
+    }
+    source, visit, destination, default_dests = destination_params
+    mock_environment = mock.Mock()
+    with pytest.raises(ValueError):
+        determine_default_destination(
+            visit=visit,
+            source=Path(source),
+            destination=destination,
+            environment=mock_environment,
+            token="token",
+        )