Atomic improvements and null driver executable (#973)

Author: maddenp-cu

src/uwtools/drivers/driver.py

@@ -31,6 +31,7 @@
 from uwtools.strings import STR
 from uwtools.utils.file import writable
 from uwtools.utils.processing import run_shell_cmd
+from uwtools.utils.tasks import poison
 
 if TYPE_CHECKING:
     from datetime import datetime, timedelta
@@ -373,6 +374,9 @@ def run(self):
         A run.
         """
         yield self.taskname(STR.run)
+        if self.config[STR.execution][STR.executable] is None:  # e.g. from a YAML null
+            msg = "%s must define 'executable' or implement custom run() method"
+            yield poison(msg % self.__class__.__name__)
         yield self._run_via_batch_submission() if self._batch else self._run_via_local_execution()
 
     @task

src/uwtools/resources/jsonschema/execution-parallel.jsonschema

@@ -11,7 +11,10 @@
       "type": "array"
     },
     "executable": {
-      "type": "string"
+      "type": [
+        "null",
+        "string"
+      ]
     },
     "incantation": {
       "type": "string"

src/uwtools/resources/jsonschema/execution-serial.jsonschema

@@ -11,7 +11,10 @@
       "type": "array"
     },
     "executable": {
-      "type": "string"
+      "type": [
+        "null",
+        "string"
+      ]
     },
     "incantation": {
       "type": "string"

src/uwtools/tests/drivers/test_driver.py

@@ -445,9 +445,11 @@ def test_Driver_run(batch, driverobj, node):
     driverobj._batch = batch
     executable = Path(driverobj.config["execution"]["executable"])
     executable.touch()
-    with patch.object(driverobj, "_run_via_batch_submission", return_value=node) as rvbs:
-        with patch.object(driverobj, "_run_via_local_execution", return_value=node) as rvle:
-            driverobj.run()
+    with (
+        patch.object(driverobj, "_run_via_batch_submission", return_value=node) as rvbs,
+        patch.object(driverobj, "_run_via_local_execution", return_value=node) as rvle,
+    ):
+        driverobj.run()
         if batch:
             rvbs.assert_called_once_with()
             rvle.assert_not_called()
@@ -456,6 +458,21 @@ def test_Driver_run(batch, driverobj, node):
             rvle.assert_called_once_with()
 
 
+def test_Driver_run__no_executable(driverobj, node, uwcaplog):
+    # Replace driverobj's run() with Driver's, as if it was never overriden:
+    driverobj.run = driver.Driver.run.__get__(driverobj)
+    driverobj._config["execution"]["executable"] = None
+    with (
+        patch.object(driverobj, "_run_via_batch_submission", return_value=node) as rvbs,
+        patch.object(driverobj, "_run_via_local_execution", return_value=node) as rvle,
+    ):
+        node = driverobj.run()
+        assert not node.ready
+        assert "must define 'executable' or implement custom run() method" in uwcaplog.text
+        rvbs.assert_not_called()
+        rvle.assert_not_called()
+
+
 @mark.parametrize(
     ("arg", "type_"),
     [("envcmds", list), ("envvars", dict), ("execution", list), ("scheduler", Slurm)],

src/uwtools/tests/test_schemas.py

@@ -1310,14 +1310,16 @@ def test_schema_esg_grid_rundir(esg_grid_prop):
 # execution-parallel
 
 
-def test_schema_parallel_execution():
+def test_schema_execution_parallel():
     config = {"executable": "fv3"}
     batchargs = {"batchargs": {"queue": "string", "walltime": "string"}}
     mpiargs = {"mpiargs": ["--flag1", "--flag2"]}
     threads = {"threads": 32}
     errors = schema_validator("execution-parallel")
     # Basic correctness:
     assert not errors(config)
+    # execution is required:
+    assert "'executable' is a required property" in errors({})
     # batchargs may optionally be specified:
     assert not errors({**config, **batchargs})
     # mpiargs may be optionally specified:
@@ -1332,15 +1334,17 @@ def test_schema_parallel_execution():
     )
 
 
-def test_schema_parallel_execution_executable():
+def test_schema_execution_parallel__executable():
     errors = schema_validator("execution-parallel", "properties", "executable")
     # String value is ok:
     assert not errors("fv3.exe")
+    # Null value is ok:
+    assert not errors(None)
     # Anything else is not:
-    assert "42 is not of type 'string'\n" in errors(42)
+    assert "42 is not of type 'null', 'string'\n" in errors(42)
 
 
-def test_schema_parallel_execution_mpiargs():
+def test_schema_execution_parallel__mpiargs():
     errors = schema_validator("execution-parallel", "properties", "mpiargs")
     # Basic correctness:
     assert not errors(["string1", "string2"])
@@ -1350,7 +1354,7 @@ def test_schema_parallel_execution_mpiargs():
     assert "42 is not of type 'string'\n" in errors(["string1", 42])
 
 
-def test_schema_parallel_execution_threads():
+def test_schema_execution_parallel__threads():
     errors = schema_validator("execution-parallel", "properties", "threads")
     # threads must be non-negative, and an integer:
     assert not errors(1)
@@ -1368,6 +1372,8 @@ def test_schema_execution_serial():
     errors = schema_validator("execution-serial")
     # Basic correctness:
     assert not errors(config)
+    # execution is required:
+    assert "'executable' is a required property" in errors({})
     # batchargs may optionally be specified:
     assert not errors({**config, **batchargs})
     # All properties are ok:
@@ -1376,6 +1382,16 @@ def test_schema_execution_serial():
     assert "Additional properties are not allowed" in errors({**config, "foo": "bar"})
 
 
+def test_schema_execution_serial__executable():
+    errors = schema_validator("execution-serial", "properties", "executable")
+    # String value is ok:
+    assert not errors("fv3.exe")
+    # Null value is ok:
+    assert not errors(None)
+    # Anything else is not:
+    assert "42 is not of type 'null', 'string'\n" in errors(42)
+
+
 # files-to-stage
 
 

src/uwtools/tests/utils/test_file.py

@@ -57,6 +57,18 @@ def test_atomic(tmp_path, uwcaplog):
     assert "Atomically renaming %s -> %s" % (tmp, path) in uwcaplog.text
 
 
+def test_atomic__fail(tmp_path, uwcaplog):
+    path = tmp_path / "foo"
+    with file.atomic(path) as tmp:
+        assert str(tmp).startswith(str(path))
+        assert not tmp.is_file()
+        # Avoid creating file.
+        assert not path.is_file()
+    assert not tmp.is_file()
+    assert not path.is_file()
+    assert "Skipping atomic rename: %s not found" % tmp in uwcaplog.text
+
+
 @mark.parametrize(
     ("ext", "file_type"),
     {

src/uwtools/tests/utils/test_tasks.py

@@ -200,7 +200,8 @@ def test_utils_tasks_filecopy__simple(tmp_path):
     assert dst.is_file()
 
 
-def test_utils_tasks_filecopy_hsi(logged, ready_task, tmp_path):
+@mark.parametrize("success", [True, False])
+def test_utils_tasks_filecopy_hsi(logged, ready_task, success, tmp_path):
     src = "/path/to/src"
     dst = tmp_path / "dst"
     tmp = tmp_path / "tmp"
@@ -210,15 +211,19 @@ def test_utils_tasks_filecopy_hsi(logged, ready_task, tmp_path):
         patch.object(tasks, "run_shell_cmd") as run_shell_cmd,
     ):
         atomic.return_value.__enter__.return_value = tmp
-        run_shell_cmd.side_effect = lambda *_a, **_kw: (dst.touch(), (True, "msg1\nmsg2\n"))[1]
+        action = lambda: dst.touch() if success else dst.exists()
+        run_shell_cmd.side_effect = lambda *_a, **_k: (action(), (success, "msg1\nmsg2\n"))[1]
         tasks.filecopy_hsi(src=src, dst=Path(dst))
     existing_hpss.assert_called_once_with(src)
     atomic.assert_called_once_with(dst)
     taskname = f"HSI {src} -> {dst}"
     run_shell_cmd.assert_called_once_with(f"hsi -q get '{tmp}' : '{src}'", taskname=taskname)
     assert logged(f"{taskname}: => msg1")
     assert logged(f"{taskname}: => msg2")
-    assert dst.exists()
+    if success:
+        assert dst.exists()
+    else:
+        assert not dst.exists()
 
 
 def test_utils_tasks_filecopy_htar(logged, ready_task, tmp_path):
@@ -320,6 +325,18 @@ def test_utils_tasks_link_target(tmp_path, wrapper):
     assert not tasks.link_target(path=tmp_path / "foo").ready
 
 
+def test_utils_tasks_poison():
+    node = tasks.poison(taskname="Unfulfilled requirement")
+    assert not node.ready
+    assert node.ref is None
+
+
+def test_utils_tasks__bad_scheme():
+    with raises(UWConfigError) as e:
+        tasks._bad_scheme(path="foo://x/y/z", scheme="foo")
+    assert str(e.value) == "Scheme 'foo' in 'foo://x/y/z' not supported"
+
+
 def test_utils_tasks__local__path_fail():
     path = "foo://bucket/a/b"
     with patch.object(tasks, "_bad_scheme") as _bad_scheme:

src/uwtools/utils/file.py

@@ -56,11 +56,15 @@ def atomic(path: Path) -> Iterator[Path]:
     :yieldtype: Path.
     """
     path.parent.mkdir(parents=True, exist_ok=True)
-    with NamedTemporaryFile(dir=path.parent, prefix="%s." % path.name) as ntf:
+    with NamedTemporaryFile(dir=path.parent, prefix="%s.tmp." % path.name) as ntf:
+        ntf.close()  # also deletes: some callers may balk at an existing file
         tmp = Path(ntf.name)
     yield tmp
-    log.debug("Atomically renaming %s -> %s", str(tmp), str(path))
-    tmp.rename(path)
+    if tmp.is_file():
+        log.debug("Atomically renaming %s -> %s", tmp, path)
+        tmp.rename(path)
+    else:
+        log.debug("Skipping atomic rename: %s not found", tmp)
 
 
 def get_config_format(path: str | Path | None, desc: str | None = None) -> str:

src/uwtools/utils/tasks.py

@@ -141,7 +141,10 @@ def filecopy_hsi(src: str, dst: Path, check: bool = True):
     dst.parent.mkdir(parents=True, exist_ok=True)
     with atomic(dst) as tmp:
         cmd = f"{STR.hsi} -q get '{tmp}' : '{src}'"
-        _, output = run_shell_cmd(cmd, taskname=taskname)
+        success, output = run_shell_cmd(cmd, taskname=taskname)
+        if not success:
+            log.error("Failed to copy %s via HSI", src)
+            tmp.unlink(missing_ok=True)
     for line in output.strip().split("\n"):
         log.info("%s: => %s", taskname, line)
 
@@ -245,6 +248,25 @@ def hardlink(
             raise UWError("Could not hardlink %s -> %s" % (dst, src)) from e
 
 
+@external
+def link_target(path: Path | str):
+    """
+    An existing file, link, or directory.
+
+    :param path: Path to the file, link, or directory.
+    :param context: Optional additional context for the file.
+    """
+    path = _local_path(path)
+    yield "Target %s" % path
+    yield Asset(path, path.exists)
+
+
+@external
+def poison(taskname: str):
+    yield taskname
+    yield Asset(None, lambda: False)
+
+
 @task
 def symlink(target: Path | str, linkname: Path | str, check: bool = True):
     """
@@ -264,19 +286,6 @@ def symlink(target: Path | str, linkname: Path | str, check: bool = True):
     Path(dst).symlink_to(src)
 
 
-@external
-def link_target(path: Path | str):
-    """
-    An existing file, link, or directory.
-
-    :param path: Path to the file, link, or directory.
-    :param context: Optional additional context for the file.
-    """
-    path = _local_path(path)
-    yield "Target %s" % path
-    yield Asset(path, path.exists)
-
-
 # Private helpers