MAINT: Consolidate workflow logic

src/fmu/dataio/_workflows/_utils.py

@@ -0,0 +1,54 @@
+import warnings
+from pathlib import Path
+
+from ert.runpaths import Runpaths as ErtRunpaths
+
+
+def validate_casepath(casepath: Path) -> Path:
+    """Validate that the case path is absolute and defined in the ERT config."""
+    if not casepath.is_absolute():
+        casepath_str = str(casepath)
+        if casepath_str.startswith("<") and casepath_str.endswith(">"):
+            raise ValueError(f"Ert variable for casepath is not defined: {casepath}")
+        raise ValueError(f"'casepath' must be an absolute path. Got: {casepath}")
+    return casepath
+
+
+def resolve_casepath(
+    run_paths: ErtRunpaths,
+    legacy_casepath: str | None,
+    require_sumo_casepath: bool = False,
+) -> Path:
+    """Resolve and validate case path from <SUMO_CASEPATH> or deprecated argument.
+
+    Uses <SUMO_CASEPATH> when defined and warns if deprecated <casepath> argument
+    is also provided. If Sumo is enabled but <SUMO_CASEPATH> is missing, an error
+    is raised, otherwise it falls back to <casepath> argument if provided.
+    """
+
+    sumo_casepath = run_paths.substitutions.get("<SUMO_CASEPATH>")
+
+    if sumo_casepath:
+        if legacy_casepath:
+            warnings.warn(
+                "Providing the case path as argument is deprecated. "
+                "It is no longer used and can safely be removed from the workflow. "
+                "The case path is now read from the <SUMO_CASEPATH> variable.",
+                FutureWarning,
+            )
+        return validate_casepath(Path(sumo_casepath))
+
+    if legacy_casepath:
+        if require_sumo_casepath:
+            raise ValueError(
+                "Missing required <SUMO_CASEPATH> definition. "
+                "Define it in your ERT config, for example:\n"
+                "DEFINE <SUMO_CASEPATH> <SCRATCH>/<USER>/<CASE_DIR>"
+            )
+        return validate_casepath(Path(legacy_casepath))
+
+    raise ValueError(
+        "The case path could not be resolved. Please define the "
+        "<SUMO_CASEPATH> variable in the ERT config, for example:\n"
+        "DEFINE <SUMO_CASEPATH> <SCRATCH>/<USER>/<CASE_DIR>"
+    )

src/fmu/dataio/_workflows/case/main.py

@@ -8,7 +8,6 @@
 import argparse
 import logging
 import shutil
-import warnings
 from pathlib import Path
 from typing import TYPE_CHECKING, Final
 
@@ -29,6 +28,7 @@
     get_fmu_directory,
 )
 
+from .._utils import resolve_casepath
 from ._config import CaseWorkflowConfig
 from ._mappings import get_stratigraphy_mappings_table, get_wellbore_mappings_table
 from ._observations import ObservationType, get_ert_observations_table
@@ -67,52 +67,6 @@
 """  # noqa: E501
 
 
-def _validate_casepath(casepath: Path) -> Path:
-    """Validate that the case path is absolute and defined in the ERT config."""
-    if not casepath.is_absolute():
-        casepath_str = str(casepath)
-        if casepath_str.startswith("<") and casepath_str.endswith(">"):
-            raise ValueError(f"Ert variable for casepath is not defined: {casepath}")
-        raise ValueError(f"'casepath' must be an absolute path. Got: {casepath}")
-    return casepath
-
-
-def _resolve_casepath(run_paths: ErtRunpaths, args: argparse.Namespace) -> Path:
-    """Resolve and validate case path from <SUMO_CASEPATH> or deprecated argument.
-
-    Uses <SUMO_CASEPATH> when defined and warns if deprecated <casepath> argument
-    is also provided. If Sumo is enabled but <SUMO_CASEPATH> is missing, an error
-    is raised, otherwise it falls back to <casepath> argument if provided.
-    """
-
-    sumo_casepath = run_paths.substitutions.get("<SUMO_CASEPATH>")
-
-    if sumo_casepath:
-        if args.casepath:
-            warnings.warn(
-                "The argument 'casepath' is deprecated. It is no longer used and can "
-                "safely be removed from WF_CREATE_CASE_METADATA. The case path is now "
-                "read from the <SUMO_CASEPATH> variable.",
-                FutureWarning,
-            )
-        return _validate_casepath(Path(sumo_casepath))
-
-    if args.casepath:
-        if args.sumo:
-            raise ValueError(
-                "Missing required <SUMO_CASEPATH> definition. "
-                "Define it in your ERT config, for example:\n"
-                "DEFINE <SUMO_CASEPATH> <SCRATCH>/<USER>/<CASE_DIR>"
-            )
-        return _validate_casepath(Path(args.casepath))
-
-    raise ValueError(
-        "The case path could not be resolved. Please define the <SUMO_CASEPATH> "
-        "variable in the ERT config, for example:\n\n    "
-        "DEFINE <SUMO_CASEPATH> <SCRATCH>/<USER>/<CASE_DIR>"
-    )
-
-
 def _get_ensemble_name(
     ensemble: ErtEnsemble,
     run_paths: ErtRunpaths,
@@ -492,7 +446,9 @@ def run(
         parser = get_parser()
         args = parser.parse_args(workflow_args)
 
-        casepath = _resolve_casepath(run_paths, args)
+        casepath = resolve_casepath(
+            run_paths, args.casepath, require_sumo_casepath=args.sumo
+        )
         maybe_fmu_dir = _copy_fmu_directory(casepath)
 
         cfg = CaseWorkflowConfig.from_presim_workflow(

src/fmu/dataio/_workflows/copy_preprocessed.py

@@ -18,6 +18,8 @@
 
 from fmu.dataio import ExportPreprocessedData
 
+from ._utils import resolve_casepath
+
 if TYPE_CHECKING:
     from ert.runpaths import Runpaths as ErtRunpaths
 
@@ -50,19 +52,6 @@
 """  # noqa
 
 
-def main() -> None:
-    """Entry point from command line
-
-    When script is called from an ERT workflow, it will be called through the 'run'
-    method on the WfCopyPreprocessedData class. This context is the intended usage.
-    The command line entry point is still included, to clarify the difference and
-    for debugging purposes.
-    """
-    parser = get_parser()
-    commandline_args = parser.parse_args()
-    copy_preprocessed_data_main(commandline_args, run_paths=None)
-
-
 class WfCopyPreprocessedData(ert.ErtScript):
     """A class with a run() function that can be registered as an ERT plugin.
 
@@ -81,17 +70,18 @@ def run(
 
 
 def copy_preprocessed_data_main(
-    args: argparse.Namespace, run_paths: ErtRunpaths | None = None
+    args: argparse.Namespace, run_paths: ErtRunpaths
 ) -> None:
     """Copy the preprocessed data to scratch and upload it to sumo."""
 
     _normalize_workflow_arguments(args)
     check_arguments(args)
     logger.setLevel(args.verbosity)
 
-    casepath = _resolve_casepath(run_paths, args.ert_caseroot)
-    ert_config_path = _resolve_ert_config_path(run_paths, args.ert_config_path)
+    casepath = resolve_casepath(run_paths, args.ert_caseroot)
+    ert_config_path = Path(run_paths.substitutions["<CONFIG_PATH>"])
     searchpath = ert_config_path / args.inpath
+
     match_pattern = "[!.]*"  # ignore metafiles (starts with '.')
     files = [
         filepath
@@ -126,6 +116,13 @@ def check_arguments(args: argparse.Namespace) -> None:
             FutureWarning,
         )
 
+    if args.ert_config_path:
+        warnings.warn(
+            "The argument 'ert_config_path' is deprecated. It is no longer used "
+            "and can safely be removed from WF_COPY_PREPROCESSED_DATAIO.",
+            FutureWarning,
+        )
+
     if Path(args.inpath).is_absolute():
         logger.debug("Argument 'inpath' is absolute: %s", args.inpath)
         raise ValueError(
@@ -151,74 +148,6 @@ def _normalize_workflow_arguments(args: argparse.Namespace) -> None:
     )
 
 
-def _validate_casepath(casepath: Path) -> Path:
-    """Validate that the case path is absolute and resolved."""
-    if not casepath.is_absolute():
-        casepath_str = str(casepath)
-        if casepath_str.startswith("<") and casepath_str.endswith(">"):
-            raise ValueError(f"Ert variable for casepath is not defined: {casepath}")
-        raise ValueError(f"'casepath' must be an absolute path. Got: {casepath}")
-    return casepath
-
-
-def _resolve_casepath(
-    run_paths: ErtRunpaths | None, legacy_casepath: str | None
-) -> Path:
-    """Resolve the case path from ERT, with legacy argument fallback."""
-    sumo_casepath = (
-        run_paths.substitutions.get("<SUMO_CASEPATH>") if run_paths else None
-    )
-
-    if sumo_casepath:
-        if legacy_casepath:
-            warnings.warn(
-                "The argument 'ert_caseroot' is deprecated. It is no longer used "
-                "and can safely be removed from WF_COPY_PREPROCESSED_DATAIO.",
-                FutureWarning,
-            )
-        return _validate_casepath(Path(sumo_casepath))
-    if legacy_casepath:
-        warnings.warn(
-            "The argument 'ert_caseroot' is deprecated. Define <SUMO_CASEPATH> "
-            "in the ERT config before removing it from "
-            "WF_COPY_PREPROCESSED_DATAIO.",
-            FutureWarning,
-        )
-        return _validate_casepath(Path(legacy_casepath))
-
-    raise ValueError(
-        "The case path could not be resolved. Please define the <SUMO_CASEPATH> "
-        "variable in the ERT config."
-    )
-
-
-def _resolve_ert_config_path(
-    run_paths: ErtRunpaths | None, legacy_config_path: str | None
-) -> Path:
-    """Resolve the ERT config path from ERT, with legacy argument fallback."""
-    ert_config_path = (
-        run_paths.substitutions.get("<CONFIG_PATH>") if run_paths else None
-    )
-
-    if ert_config_path:
-        if legacy_config_path:
-            warnings.warn(
-                "The argument 'ert_config_path' is deprecated. It is no longer used "
-                "and can safely be removed from WF_COPY_PREPROCESSED_DATAIO.",
-                FutureWarning,
-            )
-        return Path(ert_config_path)
-    if legacy_config_path:
-        warnings.warn(
-            "The argument 'ert_config_path' is deprecated. Run this workflow "
-            "through ERT before removing it from WF_COPY_PREPROCESSED_DATAIO.",
-            FutureWarning,
-        )
-        return Path(legacy_config_path)
-
-    raise ValueError("The ERT config path could not be resolved from <CONFIG_PATH>.")
-
-
 def get_parser() -> argparse.ArgumentParser:
     """Construct parser object."""
     parser = argparse.ArgumentParser()
@@ -254,7 +183,3 @@ def ertscript_workflow(config: ert.WorkflowConfigs) -> None:
         examples=EXAMPLES,
         category="export",
     )
-
-
-if __name__ == "__main__":
-    main()

tests/test_ert_integration/test_simple_export_run.py

@@ -93,7 +93,7 @@ def test_snakeoil_wf_case_metadata_includes_config_user_and_casename(
         ["ert", "ensemble_experiment", "snakeoil.ert", "--disable-monitoring"],
     )
     with (
-        pytest.warns(FutureWarning, match="'casepath' is deprecated"),
+        pytest.warns(FutureWarning, match="case path as argument is deprecated"),
         pytest.warns(FutureWarning, match="'ert_config_path' is deprecated"),
         pytest.warns(FutureWarning, match="'ert_casename' is deprecated"),
         pytest.warns(FutureWarning, match="'ert_username' is deprecated"),

tests/test_ert_integration/test_wf_copy_preprocessed_data.py

@@ -155,7 +155,7 @@ def test_deprecated_path_arguments_warn_and_are_ignored(
         ["ert", "test_run", "snakeoil.ert", "--disable-monitoring"],
     )
     with (
-        pytest.warns(FutureWarning, match="'ert_caseroot' is deprecated"),
+        pytest.warns(FutureWarning, match="case path as argument is deprecated"),
         pytest.warns(FutureWarning, match="'ert_config_path' is deprecated"),
     ):
         ert.__main__.main()

tests/test_ert_integration/test_wf_create_case_metadata.py

@@ -426,7 +426,7 @@ def test_create_case_metadata_deprecated_arguments_warn(
 
     with (
         patch("sys.argv", ["ert", "test_run", "snakeoil.ert", "--disable-monitoring"]),
-        pytest.warns(FutureWarning, match="The argument 'casepath' is deprecated"),
+        pytest.warns(FutureWarning, match="case path as argument is deprecated"),
         pytest.warns(
             FutureWarning, match="The argument 'ert_config_path' is deprecated"
         ),

tests/test_units/test_workflows/test_workflow_utils.py

@@ -0,0 +1,70 @@
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from fmu.dataio._workflows._utils import resolve_casepath, validate_casepath
+
+
+def test_validate_casepath_returns_absolute_path() -> None:
+    casepath = Path("/tmp/scratch/user/case")
+
+    assert validate_casepath(casepath) == casepath
+
+
+def test_validate_casepath_raises_for_relative_path() -> None:
+    with pytest.raises(ValueError, match="'casepath' must be an absolute path"):
+        validate_casepath(Path("relative/case"))
+
+
+def test_validate_casepath_raises_for_unresolved_ert_variable() -> None:
+    with pytest.raises(ValueError, match="Ert variable for casepath is not defined"):
+        validate_casepath(Path("<SUMO_CASEPATH>"))
+
+
+def test_resolve_casepath_uses_sumo_casepath_when_defined() -> None:
+    run_paths = MagicMock()
+    run_paths.substitutions = {"<SUMO_CASEPATH>": "/tmp/scratch/user/case"}
+
+    assert resolve_casepath(run_paths, legacy_casepath=None) == Path(
+        "/tmp/scratch/user/case"
+    )
+
+
+def test_resolve_casepath_warns_when_legacy_casepath_is_provided() -> None:
+    run_paths = MagicMock()
+    run_paths.substitutions = {"<SUMO_CASEPATH>": "/tmp/scratch/user/case"}
+
+    with pytest.warns(FutureWarning, match="Providing the case path as argument"):
+        resolved = resolve_casepath(run_paths, legacy_casepath="/tmp/legacy/case")
+
+    assert resolved == Path("/tmp/scratch/user/case")
+
+
+def test_resolve_casepath_uses_legacy_casepath_when_allowed() -> None:
+    run_paths = MagicMock()
+    run_paths.substitutions = {}  # missing <SUMO_CASEPATH>
+
+    assert resolve_casepath(run_paths, legacy_casepath="/tmp/legacy/case") == Path(
+        "/tmp/legacy/case"
+    )
+
+
+def test_resolve_casepath_raises_when_sumo_required_but_missing() -> None:
+    run_paths = MagicMock()
+    run_paths.substitutions = {}  # missing <SUMO_CASEPATH>
+
+    with pytest.raises(ValueError, match="Missing required <SUMO_CASEPATH> definition"):
+        resolve_casepath(
+            run_paths,
+            legacy_casepath="/tmp/legacy/case",
+            require_sumo_casepath=True,
+        )
+
+
+def test_resolve_casepath_raises_when_no_casepath_is_available() -> None:
+    run_paths = MagicMock()
+    run_paths.substitutions = {}  # missing <SUMO_CASEPATH>
+
+    with pytest.raises(ValueError, match="The case path could not be resolved"):
+        resolve_casepath(run_paths, legacy_casepath=None)