Add option to exit with error status if workflow is incomplete

recipe/meta.json

@@ -48,5 +48,5 @@
       "zarr ==3.1.*"
     ]
   },
-  "version": "0.6.0"
+  "version": "0.7.0"
 }

src/setup.py

@@ -13,8 +13,7 @@
 
 recipe = os.environ.get("RECIPE_DIR", "../recipe")
 metasrc = Path(recipe, "meta.json")
-with metasrc.open() as f:
-    meta = json.load(f)
+meta = json.loads(metasrc.read_text())
 name_conda = meta["name"]
 name_py = name_conda.replace("-", "_")
 

src/wxvx/cli.py

@@ -30,13 +30,15 @@ def main() -> None:
         c = validated_config(yc)
         if args.check:
             sys.exit(0)
-        logging.info("Preparing task graph for %s", args.task)
+        logging.info("Preparing task graph for '%s'", args.task)
         task = getattr(workflow, args.task)
         if args.threads > 1:
             logging.info("Using %s threads", args.threads)
         initialize_pool(args.threads)
         initialize_session(args.threads)
-        task(c, threads=args.threads)
+        node = task(c, threads=args.threads)
+        if args.fail and not node.ready:
+            fail(f"Task '{args.task}' is incomplete")
     except WXVXError as e:
         for line in traceback.format_exc().strip().split("\n"):
             logging.debug(line)
@@ -85,6 +87,12 @@ def _parse_args(argv: list[str]) -> Namespace:
         action="store_true",
         help="Log all messages",
     )
+    optional.add_argument(
+        "-f",
+        "--fail",
+        action="store_true",
+        help="Exit with error status if task is incomplete",
+    )
     optional.add_argument(
         "-h",
         "--help",

src/wxvx/resources/info.json

@@ -1,4 +1,4 @@
 {
   "buildnum": "0",
-  "version": "0.6.0"
+  "version": "0.7.0"
 }

src/wxvx/tests/test_cli.py

@@ -2,16 +2,16 @@
 Tests for wxvx.cli.
 """
 
-import logging
 import re
 from argparse import ArgumentTypeError, Namespace
 from pathlib import Path
 from unittest.mock import patch
 
 import yaml
+from iotaa import Asset, external
 from pytest import mark, raises
 
-from wxvx import cli
+from wxvx import cli, workflow
 from wxvx.strings import S
 from wxvx.types import Config
 from wxvx.util import WXVXError, pkgname, resource_path
@@ -25,8 +25,7 @@
 @mark.parametrize("threads", [1, 2])
 def test_cli_main(config_data, logged, switch_c, switch_n, switch_t, threads, tmp_path):
     cfgfile = tmp_path / "config.yaml"
-    with cfgfile.open("w") as f:
-        yaml.safe_dump(config_data, f)
+    cfgfile.write_text(yaml.safe_dump(config_data))
     argv = [pkgname, switch_c, str(cfgfile), switch_n, str(threads), switch_t, S.plots]
     with (
         patch.object(cli, "_parse_args", wraps=cli._parse_args) as _parse_args,
@@ -85,9 +84,30 @@ def test_cli_main__exception(logged):
     assert e.value.code == 1
 
 
+@mark.parametrize("switch", ["-f", "--fail", None])
+def test_cli_main__fail(config_data, switch, tmp_path):
+    @external
+    def bad(_):
+        yield "bad"
+        yield Asset(None, lambda: False)
+
+    cfgfile = tmp_path / "config.yaml"
+    cfgfile.write_text(yaml.safe_dump(config_data))
+    argv = [pkgname, "-c", str(cfgfile), "-t", "bad", switch]
+    with (
+        patch.object(cli.sys, "argv", list(filter(None, argv))),
+        patch.object(workflow, "bad", create=True, new=bad),
+    ):
+        if switch:
+            with raises(SystemExit) as e:
+                cli.main()
+            assert e.value.code == 1
+        else:
+            cli.main()
+
+
 @mark.parametrize("switch", ["-l", "--list"])
 def test_cli_main__task_list(caplog, switch, tidy):
-    caplog.set_level(logging.INFO)
     with (
         patch.object(
             cli.sys, "argv", [pkgname, "-c", str(resource_path("config-grid.yaml")), switch]
@@ -126,14 +146,13 @@ def test_cli_main__show_config(capsys, check, fs, switch):
     assert isinstance(yaml.safe_load(capsys.readouterr().out), dict)
 
 
-def test_cli_main__task_missing(caplog):
-    caplog.set_level(logging.INFO)
+def test_cli_main__task_missing(logged):
     argv = [pkgname, "-c", str(resource_path("config-grid.yaml")), "-t", "foo"]
     with patch.object(cli.sys, "argv", argv), patch.object(cli, "use_uwtools_logger"):
         with raises(SystemExit) as e:
             cli.main()
         assert e.value.code == 1
-        assert "No such task: foo" in caplog.messages
+        assert logged("No such task: foo")
 
 
 def test_cli__arg_type_int_greater_than_zero__pass():

src/wxvx/tests/test_schema.py

@@ -372,8 +372,7 @@ def validator(fs: FakeFilesystem, *args: Any) -> Callable:
     """
     schema_path = resource_path("config.jsonschema")
     fs.add_real_file(schema_path)
-    with schema_path.open() as f:
-        schema = json.load(f)
+    schema = json.loads(schema_path.read_text())
     defs = schema.get("$defs", {})
     for arg in args:
         schema = schema[arg]

src/wxvx/tests/test_util.py

@@ -117,17 +117,17 @@ def test_util_expand_stop_precedes_start(utc):
     assert str(e.value) == "Stop time 2024-12-19 06:00:00 precedes start time 2024-12-19 12:00:00"
 
 
-def test_util_fail(caplog):
+def test_util_fail(caplog, logged):
     caplog.set_level(logging.INFO)
     with raises(SystemExit) as e:
         util.fail()
     assert not caplog.messages
     with raises(SystemExit) as e:
         util.fail("foo")
-    assert "foo" in caplog.messages
+    assert logged("foo")
     with raises(SystemExit) as e:
         util.fail("foo %s", "bar")
-    assert "foo bar" in caplog.messages
+    assert logged("foo bar")
     assert e.value.code == 1
 
 

src/wxvx/util.py

@@ -178,8 +178,7 @@ def render(template: str, tc: TimeCoords, context: dict | None = None) -> str:
 
 
 def resource(relpath: str | Path) -> str:
-    with resource_path(relpath).open("r") as f:
-        return f.read()
+    return resource_path(relpath).read_text()
 
 
 def resource_path(relpath: str | Path) -> Path: