Refactor introduce runners ii

airflow_src/dags/impl/handler_impl.py

@@ -59,13 +59,12 @@
     DDA_FLAG_IN_RAW_FILE_NAME,
 )
 from shared.path_layout import get_raw_file_folder_rel_path
-from shared.path_views import CLUSTER_VIEW, Locations
 from shared.settings_scope_resolver import resolve_scoped_settings
 from shared.validation import FORBIDDEN_RAW_FILE_NAME_CHARACTERS_PATTERN
-from shared.yamlsettings import is_s3_upload_enabled
+from shared.yamlsettings import BACKUP_BASE_PATH, is_s3_upload_enabled
 
 # special mode that does not copy (e.g. because another instance handles it)
-# point locations.backup.absolute_path to the folder where the files can be picked up for quanting
+# point backup.backup_base_path to the folder where the files can be picked up for quanting
 SKIP_COPYING = False
 
 
@@ -232,7 +231,9 @@ def copy_raw_file(ti: TaskInstance, **kwargs) -> None:
     }
 
     raw_file = get_raw_file_by_id(raw_file_id)
-    backup_base_path = get_backup_base_path(raw_file)
+    backup_base_path = PurePosixPath(BACKUP_BASE_PATH) / get_raw_file_folder_rel_path(
+        raw_file
+    )
 
     if SKIP_COPYING:
         update_raw_file(
@@ -323,14 +324,6 @@ def _handle_file_copying(
     return copied_files
 
 
-# TODO: move
-def get_backup_base_path(raw_file: RawFile) -> PurePosixPath:
-    """Get the backup base path for the given raw file, e.g. /fs/pool/backup/test2/2025_07 ."""
-    return CLUSTER_VIEW.resolve(
-        Locations.BACKUP, get_raw_file_folder_rel_path(raw_file)
-    )
-
-
 def _verify_copied_files(
     copied_files: dict[Path, tuple[float, str]],
     files_dst_paths: dict[Path, Path],

airflow_src/tests/dags/impl/test_handler_impl.py

@@ -1,9 +1,11 @@
 """Unit tests for handler_impl.py."""
 
+from datetime import datetime
 from pathlib import Path
 from unittest.mock import MagicMock, Mock, call, patch
 
 import pytest
+import pytz
 from airflow.exceptions import AirflowFailException, AirflowSkipException
 from common.keys import AcquisitionMonitorErrors, DagContext, DagParams, OpArgs
 from common.settings import _INSTRUMENTS
@@ -485,18 +487,14 @@ def test_compute_checksum_no_files_found(
 
 @patch("dags.impl.handler_impl.get_xcom")
 @patch("dags.impl.handler_impl.get_raw_file_by_id")
-@patch(
-    "dags.impl.handler_impl.get_backup_base_path",
-    return_value=Path("some_backup_folder"),
-)
+@patch("dags.impl.handler_impl.BACKUP_BASE_PATH", "/fs/pool/backup")
 @patch("dags.impl.handler_impl._handle_file_copying")
 @patch("dags.impl.handler_impl._verify_copied_files")
 @patch("dags.impl.handler_impl.update_raw_file")
-def test_copy_raw_file_calls_update_with_correct_args(  # noqa: PLR0913
+def test_copy_raw_file_calls_update_with_correct_args(
     mock_update_raw_file: MagicMock,
     mock_verify_copied_files: MagicMock,
     mock_handle_file_copying: MagicMock,
-    mock_get_backup_base_path: MagicMock,  # noqa: ARG001
     mock_get_raw_file_by_id: MagicMock,
     mock_get_xcom: MagicMock,
 ) -> None:
@@ -513,7 +511,9 @@ def test_copy_raw_file_calls_update_with_correct_args(  # noqa: PLR0913
         {src_path: (1000, "some_hash")},
     ]
 
-    mock_raw_file = MagicMock()
+    mock_raw_file = MagicMock(
+        instrument_id="test1", created_at=datetime(2025, 7, 1, tzinfo=pytz.UTC)
+    )
     mock_get_raw_file_by_id.return_value = mock_raw_file
 
     mock_handle_file_copying.return_value = {Path(src_path): (1000, "some_hash")}
@@ -532,7 +532,7 @@ def test_copy_raw_file_calls_update_with_correct_args(  # noqa: PLR0913
             call(
                 "test_file.raw",
                 new_status=RawFileStatus.COPYING,
-                backup_base_path="some_backup_folder",
+                backup_base_path="/fs/pool/backup/test1/2025_07",
                 backup_status="copying_in_progress",
             ),
             call(
@@ -551,18 +551,14 @@ def test_copy_raw_file_calls_update_with_correct_args(  # noqa: PLR0913
 
 @patch("dags.impl.handler_impl.get_xcom")
 @patch("dags.impl.handler_impl.get_raw_file_by_id")
-@patch(
-    "dags.impl.handler_impl.get_backup_base_path",
-    return_value=Path("some_backup_folder"),
-)
+@patch("dags.impl.handler_impl.BACKUP_BASE_PATH", "/fs/pool/backup")
 @patch("dags.impl.handler_impl._handle_file_copying")
 @patch("dags.impl.handler_impl._verify_copied_files")
 @patch("dags.impl.handler_impl.update_raw_file")
-def test_copy_raw_file_verify_fails(  # noqa: PLR0913
+def test_copy_raw_file_verify_fails(
     mock_update_raw_file: MagicMock,
     mock_verify_copied_files: MagicMock,
     mock_handle_file_copying: MagicMock,  # noqa: ARG001
-    mock_get_backup_base_path: MagicMock,  # noqa: ARG001
     mock_get_raw_file_by_id: MagicMock,
     mock_get_xcom: MagicMock,
 ) -> None:
@@ -579,7 +575,9 @@ def test_copy_raw_file_verify_fails(  # noqa: PLR0913
         {src_path: (1000, "some_hash")},
     ]
 
-    mock_raw_file = MagicMock()
+    mock_raw_file = MagicMock(
+        instrument_id="test1", created_at=datetime(2025, 7, 1, tzinfo=pytz.UTC)
+    )
     mock_get_raw_file_by_id.return_value = mock_raw_file
 
     mock_verify_copied_files.side_effect = ValueError("File copy failed with errors")
@@ -594,7 +592,7 @@ def test_copy_raw_file_verify_fails(  # noqa: PLR0913
             call(
                 "test_file.raw",
                 new_status=RawFileStatus.COPYING,
-                backup_base_path="some_backup_folder",
+                backup_base_path="/fs/pool/backup/test1/2025_07",
                 backup_status="copying_in_progress",
             ),
             call(
@@ -611,16 +609,12 @@ def test_copy_raw_file_verify_fails(  # noqa: PLR0913
 @patch("dags.impl.handler_impl.get_xcom")
 @patch("dags.impl.handler_impl.get_raw_file_by_id")
 @patch("dags.impl.handler_impl.get_airflow_variable")
-@patch(
-    "dags.impl.handler_impl.get_backup_base_path",
-    return_value=Path("some_backup_folder"),
-)
+@patch("dags.impl.handler_impl.BACKUP_BASE_PATH", "/fs/pool/backup")
 @patch("dags.impl.handler_impl._handle_file_copying")
 @patch("dags.impl.handler_impl.update_raw_file")
 def test_copy_raw_file_calls_update_with_correct_args_overwrite(  # noqa: PLR0913
     mock_update_raw_file: MagicMock,  # noqa: ARG001
     mock_handle_file_copying: MagicMock,
-    mock_get_backup_base_path: MagicMock,  # noqa: ARG001
     mock_get_airflow_variable: MagicMock,
     mock_get_raw_file_by_id: MagicMock,
     mock_get_xcom: MagicMock,
@@ -639,7 +633,7 @@ def test_copy_raw_file_calls_update_with_correct_args_overwrite(  # noqa: PLR091
         {"/path/to/instrument/test_file.raw": (1000, "some_hash")},
     ]
 
-    mock_raw_file = MagicMock()
+    mock_raw_file = MagicMock(created_at=datetime(2025, 7, 1, tzinfo=pytz.UTC))
     mock_raw_file.id = "test_file.raw"
     mock_raw_file.instrument_id = "instrument1"
     mock_get_raw_file_by_id.return_value = mock_raw_file

envs/alphakraken.local.yaml

@@ -59,6 +59,8 @@ general:
     webapp_url: "http://localhost:8501"
 
 backup:
+  # absolute path of the backup folder on the shared file system, persisted in the DB for display
+  backup_base_path: /fs/pool/pool-0/alphakraken_sandbox/backup
   backup_type: local # or 's3'
   purging_verification_type: local # 'local', 's3', or 'force_local'
   s3:

envs/alphakraken.production.yaml

@@ -39,6 +39,7 @@ general:
     webapp_url: "<webapp_url>"
 
 backup:
+  backup_base_path: /fs/pool-1/backup
   backup_type: local # or 's3'
   purging_verification_type: local # 'local', 's3', or 'force_local'
   s3:

envs/alphakraken.sandbox.yaml

@@ -39,6 +39,7 @@ general:
     webapp_url: "<webapp_url>"
 
 backup:
+  backup_base_path: /fs/pool-0/alphakraken_sandbox/backup
   backup_type: local # or 's3'
   purging_verification_type: local # 'local', 's3', or 'force_local'
   s3:

shared/tests/test_deployment_paths.py

@@ -112,6 +112,12 @@ def test_instrument_mount_targets_are_below_the_instruments_location(
         )
 
 
+@pytest.mark.parametrize(("file_name", "config"), _env_yamls())
+def test_backup_base_path_is_declared(file_name: str, config: dict) -> None:
+    """Test that each environment declares the folder the backups are displayed under."""
+    assert config["backup"]["backup_base_path"], file_name
+
+
 def test_the_logs_are_not_mounted_below_the_mounts_folder() -> None:
     """Test the one bind that deliberately breaks the mirroring, so that it stays deliberate."""
     compose = yaml.safe_load((_REPO_ROOT / "docker-compose.yaml").read_text())

shared/tests/test_yamlsettings.py

@@ -90,6 +90,7 @@ def test_returns_test_settings_for_test_environment(
                 "webapp_url": "http://localhost:8501",
             }
         },
+        "backup": {"backup_base_path": "./tmp/test/backup"},
         "locations": {
             "settings": {"absolute_path": "./tmp/test/settings"},
             "output": {"absolute_path": "./tmp/test/output"},
@@ -100,6 +101,24 @@ def test_returns_test_settings_for_test_environment(
     }
 
 
+def test_read_backup_base_path() -> None:
+    """Test that the backup base path is read from the `backup` block."""
+    from shared.yamlsettings import _read_backup_base_path
+
+    assert (
+        _read_backup_base_path({"backup": {"backup_base_path": "/some/backup"}})
+        == "/some/backup"
+    )
+
+
+def test_read_backup_base_path_raises_naming_the_key() -> None:
+    """Test that a missing backup base path is reported with its full yaml key."""
+    from shared.yamlsettings import _read_backup_base_path
+
+    with pytest.raises(KeyError, match="backup.backup_base_path"):
+        _read_backup_base_path({"backup": {}})
+
+
 class TestGetPurgingVerificationType:
     """Tests for get_purging_verification_type cross-validation."""
 

shared/yamlsettings.py

@@ -21,6 +21,8 @@ class YamlKeys:
     LOCATIONS = "locations"
     ABSOLUTE_PATH = "absolute_path"
 
+    BACKUP = "backup"
+
     NOTIFICATIONS = "notifications"
     OPS_ALERTS_WEBHOOK_URL = "ops_alerts_webhook_url"
     BUSINESS_ALERTS_WEBHOOK_URL = "business_alerts_webhook_url"
@@ -43,6 +45,8 @@ class Backup:
 
         # TODO: incorporate
 
+        BACKUP_BASE_PATH = "backup_base_path"
+
         TYPE = "backup.backup_type"
         S3_REGION = "backup.s3.region"
         S3_BUCKET_PREFIX = "backup.s3.bucket_prefix"
@@ -78,6 +82,7 @@ def load_alphakraken_yaml(cls) -> dict[str, dict[str, Any]]:
                         "webapp_url": "http://localhost:8501",
                     }
                 },
+                "backup": {"backup_base_path": "./tmp/test/backup"},
                 "locations": {
                     "settings": {"absolute_path": "./tmp/test/settings"},
                     "output": {"absolute_path": "./tmp/test/output"},
@@ -102,6 +107,19 @@ def load_alphakraken_yaml(cls) -> dict[str, dict[str, Any]]:
 )
 
 
+def _read_backup_base_path(settings: dict[str, dict[str, Any]]) -> str:
+    """Read the absolute path of the backup folder on the shared file system."""
+    try:
+        return settings[YamlKeys.BACKUP][YamlKeys.Backup.BACKUP_BASE_PATH]
+    except KeyError as e:
+        raise KeyError(
+            f"Key `{YamlKeys.BACKUP}.{YamlKeys.Backup.BACKUP_BASE_PATH}` not found in alphakraken.yaml."
+        ) from e
+
+
+BACKUP_BASE_PATH: str = _read_backup_base_path(YAMLSETTINGS)
+
+
 def get_notification_setting(setting_key: str) -> str:
     """Get a notification setting from the yaml settings."""
     setting_value = (

tasks/todo.md

@@ -37,13 +37,13 @@ Three yamls and the `_test_` stub gain `backup.backup_base_path` = former
 reworded. Consistency test: key present in every in-repo yaml. Spec 1.2.3, 2.2, 7.3 second half.
 
 **Acceptance criteria:**
-- [ ] `get_backup_base_path` returns the same string as before for the same yaml values (regression test with patched `BACKUP_BASE_PATH`).
-- [ ] The reader function raises naming `backup.backup_base_path` on a dict without the key.
-- [ ] `handler_impl.py` no longer imports `CLUSTER_VIEW`.
+- [x] `get_backup_base_path` returns the same string as before for the same yaml values (regression test with patched `BACKUP_BASE_PATH`).
+- [x] The reader function raises naming `backup.backup_base_path` on a dict without the key.
+- [x] `handler_impl.py` no longer imports `CLUSTER_VIEW`.
 
 **Verification:**
-- [ ] `pytest shared/tests/test_yamlsettings.py shared/tests/test_deployment_paths.py airflow_src/tests/dags/impl/test_handler_impl.py`
-- [ ] All three suites
+- [x] `pytest shared/tests/test_yamlsettings.py shared/tests/test_deployment_paths.py airflow_src/tests/dags/impl/test_handler_impl.py`
+- [x] All three suites
 
 **Dependencies:** None
 **Files:** `envs/alphakraken.{local,sandbox,production}.yaml`, `shared/yamlsettings.py`, `shared/tests/test_yamlsettings.py`, `airflow_src/dags/impl/handler_impl.py`, `airflow_src/tests/dags/impl/test_handler_impl.py`, `shared/tests/test_deployment_paths.py`