otk/command: implement 'validate' command

example/fedora/minimal-40-aarch64.yaml

@@ -43,7 +43,7 @@ otk.define:
       options: "defaults,uid=0,gid=0,umask=077,shortname=winnt"
       passno: 2
 
-otk.target.osbuild:
+otk.target.osbuild.qcow2:
   pipelines:
     - otk.include: "osbuild/buildroot.yaml"
     - otk.include: "osbuild/pipeline/tree.yaml"

example/fedora/minimal-40-x86_64.yaml

@@ -43,7 +43,7 @@ otk.define:
       options: "defaults,uid=0,gid=0,umask=077,shortname=winnt"
       passno: 2
 
-otk.target.osbuild:
+otk.target.osbuild.qcow2:
   pipelines:
     - otk.include: "osbuild/buildroot.yaml"
     - otk.include: "osbuild/pipeline/tree.yaml"

src/otk/command.py

@@ -21,7 +21,7 @@ def root() -> int:
     parser = parser_create()
     arguments = parser.parse_args(argv)
 
-    # turn on logging as *soon* as possible so it can be used early
+    # turn on logging as *soon* as possible, so it can be used early
     logging.basicConfig(
         level=logging.WARNING - (10 * arguments.verbose),
         handlers=[
@@ -41,25 +41,27 @@ def root() -> int:
 
     if arguments.command == "compile":
         return compile(parser, arguments)
+    elif arguments.command == "validate":
+        return validate(parser, arguments)
 
     raise RuntimeError("Unknown subcommand")
 
 
-def compile(
-    parser: argparse.ArgumentParser, arguments: argparse.Namespace
+def _process(
+    parser: argparse.ArgumentParser, arguments: argparse.Namespace, dry_run: bool
 ) -> int:
     src = pathlib.Path(
         "/dev/stdin" if arguments.input is None else arguments.input
     )
     dst = pathlib.Path(
         "/dev/stdout" if arguments.output is None else arguments.output
-    )
+    ) if not dry_run else None
 
     if not src.exists():
         log.fatal("INPUT path %r does not exist", str(src))
         return 1
 
-    if not dst.exists():
+    if not dry_run and not dst.parent.exists():
         log.fatal("OUTPUT path %r does not exist", str(dst))
         return 1
 
@@ -107,7 +109,7 @@ def compile(
         kind, name = target_requested.split(".")
     except ValueError:
         # TODO handle earlier
-        log.fatal("malformatted target name %r", target_requested)
+        log.fatal("malformed target name %r. We need a format of '<TARGET_KIND>.<TARGET_NAME>'.", target_requested)
         return 1
 
     # re-resolve the specific target with the specific context and target if
@@ -116,13 +118,22 @@ def compile(
     tree = resolve(spec, tree[f"{PREFIX_TARGET}{kind}.{name}"])
 
     # and then output by writing to the output
-    dst.write_text(
-        target_registry.get(kind, CommonTarget)().as_string(spec, tree)
-    )
+    if not dry_run:
+        dst.write_text(
+            target_registry.get(kind, CommonTarget)().as_string(spec, tree)
+        )
 
     return 0
 
 
+def compile(parser: argparse.ArgumentParser, arguments: argparse.Namespace) -> int:
+    return _process(parser, arguments, dry_run=False)
+
+
+def validate(parser: argparse.ArgumentParser, arguments: argparse.Namespace) -> int:
+    return _process(parser, arguments, dry_run=True)
+
+
 def parser_create() -> argparse.Namespace:
     # set up the main parser arguments
     parser = argparse.ArgumentParser(
@@ -180,4 +191,19 @@ def parser_create() -> argparse.Namespace:
         help="Target to output, required if more than one target exists in an omnifest.",
     )
 
+    parser_validate = subparsers.add_parser("validate", help="Validate an omnifest.")
+    parser_validate.add_argument(
+        "input",
+        metavar="INPUT",
+        nargs="?",
+        default=None,
+        help="Omnifest to validate to or none for STDIN.",
+    )
+    parser_validate.add_argument(
+        "-t",
+        "--target",
+        default=None,
+        help="Target to validate, required if more than one target exists in an omnifest.",
+    )
+
     return parser

src/otk/external.py

@@ -68,11 +68,11 @@ def call(context: Context, directive: str, tree: Any) -> Any:
 registry = {
     "otk.external.osbuild.depsolve_dnf4": (
         OSBuildContext,
-        ["otk-osbuild", "depsolve-dnf4"],
+        ["otk-osbuild", "depsolve_dnf4"],
     ),
     "otk.external.osbuild.depsolve_dnf5": (
         OSBuildContext,
-        ["otk-osbuild", "depsolve-dnf5"],
+        ["otk-osbuild", "depsolve_dnf5"],
     ),
     "otk.external.osbuild.file_from_text": (
         OSBuildContext,

test/test_command.py

@@ -1,65 +1,68 @@
+import pytest
+
 from otk.command import parser_create
 
 
-def test_parse_no_command():
+def test_parse_no_command(capsys):
     p = parser_create()
+    with pytest.raises(SystemExit) as sys_exit:
+        p.parse_args([])
+    assert sys_exit.type == SystemExit
+    assert sys_exit.value.code == 2
 
-    r = p.parse_args([])
-
-    assert r.command is None
-    assert r.verbose == 0
-    assert not r.json
-    assert r.identifier is None
-
-    r = p.parse_args(["-j"])
-    assert r.command is None
-    assert r.json
-    assert r.verbose == 0
-    assert r.identifier is None
-
-    r = p.parse_args(["-j", "-v"])
-    assert r.command is None
-    assert r.json
-    assert r.verbose == 1
-    assert r.identifier is None
-
-    r = p.parse_args(["-j", "-vvvv"])
-    assert r.command is None
-    assert r.json
-    assert r.verbose == 4
-    assert r.identifier is None
+    captured_stderr = capsys.readouterr().err
+    assert "the following arguments are required: command" in captured_stderr
 
-    r = p.parse_args(["--json", "--verbose", "--verbose"])
-    assert r.command is None
-    assert r.json
-    assert r.verbose == 2
-    assert r.identifier is None
 
-    r = p.parse_args(["-i", "foo"])
-    assert r.command is None
-    assert not r.json
-    assert r.verbose == 0
-    assert r.identifier == "foo"
-
-
-def test_parse_compile():
+@pytest.mark.parametrize(
+    "command,results",
+    [
+        (["compile"], {"command": "compile", "output": None, "input": None}),
+        (["compile", "foo"], {"command": "compile", "output": None, "input": "foo"}),
+        (
+            ["compile", "-o", "output_manifest.yaml"],
+            {"command": "compile", "output": "output_manifest.yaml", "input": None},
+        ),
+        (
+            ["compile", "-o", "output_manifest.yaml", "input_omifest.yaml"],
+            {
+                "command": "compile",
+                "output": "output_manifest.yaml",
+                "input": "input_omifest.yaml",
+            },
+        ),
+        (["validate", "foo.yaml"], {"command": "validate", "input": "foo.yaml"}),
+    ],
+)
+def test_parse_commands_success(capsys, command, results):
     p = parser_create()
+    r = p.parse_args(command)
+    for k in results.keys():
+        assert getattr(r, k) == results[k]
 
-    r = p.parse_args(["compile"])
-    assert r.command == "compile"
-    assert r.output is None
 
-    r = p.parse_args(["compile", "foo"])
-    assert r.command == "compile"
-    assert r.input == "foo"
-    assert r.output is None
-
-    r = p.parse_args(["compile", "-o", "file.yaml"])
-    assert r.command == "compile"
-    assert r.input is None
-    assert r.output == "file.yaml"
+@pytest.mark.parametrize(
+    "command,sys_exit_code,sys_exit_message",
+    [
+        (
+            ["compile", "--no_such_option"],
+            2,
+            "unrecognized arguments: --no_such_option",
+        ),
+        (
+            ["validate", "-o", "output_manifest.yaml", "input_omifest.yaml"],
+            2,
+            # deliberately we expect '-o input_omifest.yaml' as argparse acts like this if -o is not defined
+            "error: unrecognized arguments: -o input_omifest.yaml",
+        ),
+    ],
+)
+def test_parse_commands_failure(capsys, command, sys_exit_code, sys_exit_message):
+    p = parser_create()
+    with pytest.raises(SystemExit) as sys_exit:
+        p.parse_args(command)
+    assert sys_exit.type == SystemExit
+    assert sys_exit.value.code == sys_exit_code
 
-    r = p.parse_args(["compile", "-o", "file.yaml", "foo.yaml"])
-    assert r.command == "compile"
-    assert r.input == "foo.yaml"
-    assert r.output == "file.yaml"
+    captured_stderr = capsys.readouterr().err
+    assert sys_exit_message in captured_stderr

test/test_compile.py

@@ -0,0 +1,57 @@
+import argparse
+import os.path
+import pytest
+
+from otk.command import compile, parser_create
+
+
+@pytest.mark.parametrize(
+    "arguments, input_data, output_data, sys_exit_code, log_message",
+    [
+        # "GENERATE" is a placeholder to append "tmp_path" from pytest
+        (
+            argparse.Namespace(input="GENERATE", output="GENERATE", target=None),
+            """otk.version: 1
+otk.target.osbuild.qcow2: { test: 1 }
+""",
+            """{
+  "test": 1,
+  "version": "2",
+  "sources": {}
+}""",
+            0,
+            None,
+        ),
+    ],
+)
+def test_compile(
+    tmp_path,
+    caplog,
+    capsys,
+    monkeypatch,
+    arguments,
+    input_data,
+    output_data,
+    sys_exit_code,
+    log_message,
+):
+    if arguments.input == "GENERATE":
+        input_filename = "input.yaml"
+        arguments.input = os.path.join(tmp_path, input_filename)
+        with open(arguments.input, "w") as f:
+            f.write(input_data)
+
+    if arguments.output == "GENERATE":
+        # fix path so we only write to tmp_path
+        arguments.output = os.path.join(tmp_path, "output.yaml")
+    parser = parser_create()
+
+    ret = compile(parser, arguments)
+    assert ret == sys_exit_code
+
+    if log_message:
+        assert log_message in caplog.text
+
+    assert os.path.exists(arguments.output)
+    with open(arguments.output) as f:
+        assert f.readlines() == output_data.splitlines(True)

test/test_validate.py

@@ -0,0 +1,67 @@
+import argparse
+import os.path
+import pytest
+
+from otk.command import validate, parser_create
+
+
+@pytest.mark.parametrize(
+    "arguments, input_data, sys_exit_code, log_message",
+    [
+        # "GENERATE" is a placeholder to append "tmp_path" from pytest
+        (
+            argparse.Namespace(input="GENERATE", output="GENERATE", target=None),
+            """otk.version: 1
+otk.target.osbuild.qcow2: { test: 1 }
+""",
+            0,
+            None,
+        ),
+        (
+            argparse.Namespace(input="GENERATE", output=None, target=None),
+            """otk.version: 1
+otk.target.osbuild.qcow2: { test: 1 }
+""",
+            0,
+            None,
+        ),
+        (
+            argparse.Namespace(input="GENERATE", output="GENERATE", target=None),
+            """otk.version: 1
+otk.target.osbuild.qcow2: { test: 1 }
+otk.target.osbuild.ami: { test: 2 }
+""",
+            1,
+            "INPUT contains multiple targets, `-t` is required",
+        ),
+    ],
+)
+def test_validate(
+    tmp_path,
+    caplog,
+    capsys,
+    monkeypatch,
+    arguments,
+    input_data,
+    sys_exit_code,
+    log_message,
+):
+    if arguments.input == "GENERATE":
+        input_filename = "input.yaml"
+        arguments.input = os.path.join(tmp_path, input_filename)
+        with open(arguments.input, "w") as f:
+            f.write(input_data)
+
+    if arguments.output == "GENERATE":
+        # fix path so we only write to tmp_path
+        arguments.output = os.path.join(tmp_path, "output.yaml")
+    parser = parser_create()
+
+    ret = validate(parser, arguments)
+    assert ret == sys_exit_code
+
+    if log_message:
+        assert log_message in caplog.text
+
+    if arguments.output:
+        assert not os.path.exists(arguments.output)