osbuild-dev: a new tool to help with manifests

tools/osbuild-dev

@@ -0,0 +1,209 @@
+#!/usr/bin/env python3
+"""`osbuild-dev` provides helper functionality for `osbuild` development
+mostly centered around manifest reading."""
+
+# pylint: disable=unsupported-membership-test,unsupported-delete-operation
+# pylint: disable=unsubscriptable-object
+
+import json
+import os
+import secrets
+import subprocess
+import tempfile
+from typing import Any, Optional
+
+try:
+    import attrs
+    import rich
+    import typer
+    from rich.tree import Tree
+except ImportError as import_error:
+    print(
+        "You are missing dependencies, please install `python3-attrs`, `python3-rich`, `python3-typer` or their `pip` equivalents."
+    )
+    raise
+
+
+cli = typer.Typer()  # Main command.
+man = typer.Typer()  # Manifest subcommand.
+
+cli.add_typer(man, name="manifest")
+
+con = rich.console.Console()
+
+
+@cli.callback()
+def main() -> int:
+    return 0
+
+
+def json_as_terminal_tree(tree: Optional[Tree], data: Any, name: str) -> Tree:
+    """Convert JSON into a `rich` tree."""
+
+    if tree is None:
+        tree = Tree("")
+
+    if isinstance(data, (int, str)):
+        subtree = tree.add(f"{name}: [bold]{data}[/bold]")
+    elif isinstance(data, dict):
+        subtree = tree.add(str(name))
+        for key, val in data.items():
+            json_as_terminal_tree(subtree, val, key)
+    elif isinstance(data, list):
+        name = f"{name} [italic]({len(data)})[/italic]"
+        subtree = tree.add(name)
+        for index, item in enumerate(data):
+            json_as_terminal_tree(subtree, item, index)
+    else:
+        raise ValueError(
+            f"json_as_terminal_tree does not know how to deal with {type(data)}"
+        )
+
+    return subtree
+
+
+@attrs.define()
+class Manifest:
+    name: str
+    data: dict[str, Any] = attrs.Factory(dict)
+
+    def ignore_a_stage(self, name: str) -> None:
+        """Remove a stage from the data we represent."""
+
+        for pipeline in self.data["pipelines"]:
+            to_pop = []
+
+            for index, stage in enumerate(pipeline["stages"]):
+                if stage["type"] == name:
+                    to_pop.append(index)
+
+            for index in to_pop:
+                pipeline["stages"].pop(index)
+
+    def ignore_sources(self) -> None:
+        """Remove the `sources` section from the manifest."""
+
+        if "sources" in self.data:
+            del self.data["sources"]
+
+    def resolve_content_hashes(self) -> None:
+        # If we're resolving content hashes back to names we adjust the data structure
+        # in-place.
+        sources = {}
+
+        # We can't handle all source types but some we can
+        if "org.osbuild.curl" in self.data["sources"]:
+            for name, source in self.data["sources"]["org.osbuild.curl"][
+                "items"
+            ].items():
+                sources[name] = source["url"]
+
+        for pipeline in self.data["pipelines"]:
+            for stage in pipeline["stages"]:
+                if stage["type"] == "org.osbuild.rpm":
+                    for index, reference in enumerate(
+                        stage["inputs"]["packages"]["references"]
+                    ):
+                        stage["inputs"]["packages"]["references"][
+                            index
+                        ] = sources[reference["id"]].split("/")[-1]
+
+    def print_for_terminal(self, path: Optional[str] = None) -> None:
+        if path is None:
+            con.print(json_as_terminal_tree(None, self.data, self.name))
+        else:
+            with open(path, "w", encoding="utf8") as f:
+                rich.print(
+                    json_as_terminal_tree(None, self.data, self.name), file=f
+                )
+
+    def print_for_html(self) -> None:
+        pass
+
+    @classmethod
+    def from_path(cls, path: str) -> "Manifest":
+        try:
+            with open(path, encoding="utf8") as f:
+                data = json.load(f)
+        except FileNotFoundError:
+            con.print(f"[bold][red]Could not open file {path!r}[/red][/bold]")
+            return 1
+
+        # We deal with this possibly being a 'wrapped' manifest, one produced
+        # by `osbuild-composer`.
+        if "manifest" in data:
+            data = data["manifest"]
+
+        return cls(os.path.basename(path), data)
+
+
+@man.command(name="print")
+def pretty_print(
+    manifest_path: str,
+    ignore_stage: list[str] = typer.Option([]),
+    resolve_sources: bool = typer.Option(
+        True, help="Resolve content hashes of sources to their names."
+    ),
+    skip_sources: bool = typer.Option(
+        True, help="Skips display of the sources in the manifest."
+    ),
+) -> int:
+    """Pretty print an `osbuild` manifest file."""
+
+    manifest = Manifest.from_path(manifest_path)
+
+    for name in ignore_stage:
+        manifest.ignore_a_stage(name)
+
+    if resolve_sources:
+        manifest.resolve_content_hashes()
+
+    if skip_sources:
+        manifest.ignore_sources()
+
+    manifest.print_for_terminal()
+
+    return 0
+
+
+@man.command(name="diff")
+def pretty_diff(
+    manifest_paths: list[str],
+    ignore_stage: list[str] = typer.Option([]),
+    resolve_sources: bool = typer.Option(
+        True, help="Resolve content hashes of sources to their names."
+    ),
+    skip_sources: bool = typer.Option(
+        True, help="Skips display of the sources in the manifest."
+    ),
+) -> int:
+    """Pretty print a diff of `osbuild` manifest files."""
+
+    with tempfile.TemporaryDirectory() as temporary:
+        paths = []
+
+        for manifest_path in manifest_paths:
+            manifest = Manifest.from_path(manifest_path)
+
+            for name in ignore_stage:
+                manifest.ignore_a_stage(name)
+
+            if resolve_sources:
+                manifest.resolve_content_hashes()
+
+            if skip_sources:
+                manifest.ignore_sources()
+
+            path = f"{temporary}/{os.path.basename(manifest_path)}-{secrets.token_hex(2)}"
+
+            manifest.print_for_terminal(path)
+
+            paths.append(path)
+
+        subprocess.run(["vimdiff"] + paths, check=True)
+
+    return 0
+
+
+if __name__ == "__main__":
+    cli()