Improve introspection data manipulation

misc_scripts/update_introspection_data.py

@@ -22,7 +22,7 @@
 import sys
 import xml.etree.ElementTree as ET
 from enum import Enum
-from typing import List, Mapping, Optional, Sequence
+from typing import List, Mapping, MutableMapping, Sequence
 
 import dbus
 from dbus.proxies import ProxyObject
@@ -165,64 +165,102 @@ def setup_minimal_object_set(bus: dbus.SystemBus) -> dict[ProxyType, ProxyObject
     }
 
 
-def _make_python_spec(
-    proxies: Mapping[ProxyType, ProxyObject], *, revision_number: Optional[int] = None
-) -> dict[str, str]:
+def _get_revision_ext(
+    manager: ProxyObject, maybe_revision_number: int | None = None
+) -> str:
     """
-    Make the introspection spec for python consumption.
+    Return revision extension.
     """
+    return f"r{
+        Version(Manager.Properties.Version.Get(manager)).minor
+        if maybe_revision_number is None
+        else maybe_revision_number
+    }"
 
-    revision_number = (
-        Version(Manager.Properties.Version.Get(proxies[ProxyType.MANAGER])).minor
-        if revision_number is None
-        else revision_number
-    )
-
-    revision = f"r{revision_number}"
-
-    def get_current_interfaces(interface_prefixes: Sequence[str]) -> List[str]:
-        return [f"{prefix}.{revision}" for prefix in interface_prefixes]
 
-    specs = {}
-
-    def _add_data(proxy_key: ProxyType, interfaces: Sequence[str]):
-        """
-        Introspect on the proxy, get the information for the specified
-        interfaces, and add it to specs.
-
-        :param proxy: dbus Proxy object
-        :param list interfaces: list of interesting interface names
-        :raises: RuntimeError if some interface not found
-        """
-        string_data = Introspectable.Methods.Introspect(proxies[proxy_key], {})
-        xml_data = ET.fromstring(string_data)
+def _get_current_interfaces(
+    revision_ext: str, interface_prefixes: Sequence[str]
+) -> List[str]:
+    """
+    Return a list of interfaces names.
+    """
+    return [f"{prefix}.{revision_ext}" for prefix in interface_prefixes]
 
-        for interface_name in interfaces:
-            try:
-                interface = next(
-                    interface
-                    for interface in xml_data
-                    if interface.attrib["name"] == interface_name
-                )
-                specs[interface_name] = _xml_object_to_str(interface)
-            except StopIteration as err:
-                raise RuntimeError(
-                    f"interface {interface_name} not found in introspection data"
-                ) from err
 
-    _add_data(ProxyType.MANAGER, [OBJECT_MANAGER_INTERFACE])
+def _add_data(
+    specs: MutableMapping[str, str],
+    proxy_object: ProxyObject,
+    interfaces: Sequence[str],
+):
+    """
+    Introspect on the proxy, get the information for the specified
+    interfaces, and add it to specs.
 
-    _add_data(ProxyType.MANAGER, get_current_interfaces(TOP_OBJECT_INTERFACE_PREFIXES))
-    _add_data(ProxyType.POOL, get_current_interfaces(POOL_OBJECT_INTERFACE_PREFIXES))
+    :param proxy: dbus Proxy object
+    :param list interfaces: list of interesting interface names
+    :raises: RuntimeError if some interface not found
+    """
+    string_data = Introspectable.Methods.Introspect(proxy_object, {})
+    xml_data = ET.fromstring(string_data)
+
+    for interface_name in interfaces:
+        try:
+            interface = next(
+                interface
+                for interface in xml_data
+                if interface.attrib["name"] == interface_name
+            )
+            specs[interface_name] = _xml_object_to_str(interface)
+        except StopIteration as err:
+            raise RuntimeError(
+                f"interface {interface_name} not found in introspection data"
+            ) from err
+
+
+def _add_stratis_specs(
+    specs: MutableMapping[str, str],
+    proxies: Mapping[ProxyType, ProxyObject],
+    revision_ext: str,
+):
+    """
+    Add specs for Stratis interfaces.
+    """
+    _add_data(
+        specs,
+        proxies[ProxyType.MANAGER],
+        _get_current_interfaces(revision_ext, TOP_OBJECT_INTERFACE_PREFIXES),
+    )
+    _add_data(
+        specs,
+        proxies[ProxyType.POOL],
+        _get_current_interfaces(revision_ext, POOL_OBJECT_INTERFACE_PREFIXES),
+    )
 
     _add_data(
-        ProxyType.BLOCKDEV, get_current_interfaces(BLOCKDEV_OBJECT_INTERFACE_PREFIXES)
+        specs,
+        proxies[ProxyType.BLOCKDEV],
+        _get_current_interfaces(revision_ext, BLOCKDEV_OBJECT_INTERFACE_PREFIXES),
     )
     _add_data(
-        ProxyType.FILESYSTEM,
-        get_current_interfaces(FILESYSTEM_OBJECT_INTERFACE_PREFIXES),
+        specs,
+        proxies[ProxyType.FILESYSTEM],
+        _get_current_interfaces(revision_ext, FILESYSTEM_OBJECT_INTERFACE_PREFIXES),
     )
 
+
+def _make_python_spec(
+    proxies: Mapping[ProxyType, ProxyObject], *, revision_number: int | None = None
+) -> dict[str, str]:
+    """
+    Make the introspection spec for python consumption.
+    """
+    revision_ext = _get_revision_ext(proxies[ProxyType.MANAGER], revision_number)
+
+    specs: dict[str, str] = {}
+
+    _add_data(specs, proxies[ProxyType.MANAGER], [OBJECT_MANAGER_INTERFACE])
+    _add_stratis_specs(specs, proxies, revision_ext)
+
     return specs
 
 
@@ -254,62 +292,35 @@ def _python_output(namespace: argparse.Namespace):
     _print_python_spec(specs)
 
 
-def _make_docs_spec(proxies: Mapping[ProxyType, ProxyObject]) -> dict[ProxyType, str]:
+def _make_docs_spec(
+    proxies: Mapping[ProxyType, ProxyObject], revision_number: int | None
+) -> dict[str, str]:
     """
     Make the introspection spec for use in docs repo.
     """
-
-    specs = {}
-
-    def _add_data(proxy_key: ProxyType):
-        """
-        Get the introspection data, and add it to the spec.
-
-        :param proxy_key: key for proxies
-        """
-        string_data = Introspectable.Methods.Introspect(proxies[proxy_key], {})
-        specs[proxy_key] = _xml_object_to_str(ET.fromstring(string_data))
-
-    _add_data(ProxyType.MANAGER)
-    _add_data(ProxyType.POOL)
-    _add_data(ProxyType.BLOCKDEV)
-    _add_data(ProxyType.FILESYSTEM)
-
+    revision_ext = _get_revision_ext(proxies[ProxyType.MANAGER], revision_number)
+    specs: dict[str, str] = {}
+    _add_stratis_specs(specs, proxies, revision_ext)
     return specs
 
 
-def _print_docs_spec(specs: Mapping[ProxyType, str], namespace: argparse.Namespace):
+def _print_docs_spec(specs: Mapping[str, str], namespace: argparse.Namespace):
     """
     Print spec for inclusion on docs website.
 
     :param specs: the specification to print
-    :type specs: dict of ProxyType * XML object
+    :type specs: dict of str * str
     :param namespace: the namespace parsed from the command-line arguments
     """
 
-    def _proxy_type_to_filename(proxy_type: ProxyType) -> str:
-        """
-        Return filename for proxy type.
-        """
-        if proxy_type == ProxyType.MANAGER:
-            return namespace.manager_file_name
-        if proxy_type == ProxyType.POOL:
-            return namespace.pool_file_name
-        if proxy_type == ProxyType.FILESYSTEM:
-            return namespace.filesystem_file_name
-        if proxy_type == ProxyType.BLOCKDEV:
-            return namespace.blockdev_file_name
-
-        assert False, "unreachable"
-
     abs_output_dir = os.path.abspath(namespace.output_dir)
     try:
         os.mkdir(abs_output_dir)
     except FileExistsError as err:
         raise RuntimeError("Cannot create output dir for files") from err
 
-    for proxy_type, introspection_data in specs.items():
-        file_path = os.path.join(abs_output_dir, _proxy_type_to_filename(proxy_type))
+    for interface_name, introspection_data in specs.items():
+        file_path = os.path.join(abs_output_dir, f"{interface_name}.xml")
         with open(file_path, "w", encoding="utf-8") as file:
             print(introspection_data, file=file)
 
@@ -320,7 +331,7 @@ def _docs_output(namespace: argparse.Namespace):
     """
     bus = dbus.SystemBus()
     proxies = setup_minimal_object_set(bus)
-    specs = _make_docs_spec(proxies)
+    specs = _make_docs_spec(proxies, revision_number=namespace.revision_number)
     _print_docs_spec(specs, namespace)
 
 
@@ -337,44 +348,21 @@ def _gen_parser() -> argparse.ArgumentParser:
         )
     )
 
+    parser.add_argument(
+        "--revision-number", help="D-Bus interface revision number", type=int
+    )
+
     subparsers = parser.add_subparsers(title="subcommands")
 
     python_parser = subparsers.add_parser(
         "python", help="Generate introspection data for consumption by Python scripts"
     )
-    python_parser.add_argument(
-        "--revision-number", help="D-Bus interface revision number", type=int
-    )
     python_parser.set_defaults(func=_python_output)
 
     docs_parser = subparsers.add_parser(
         "docs", help="Generate introspection data for consumption by website docs"
     )
     docs_parser.add_argument("output_dir", help="directory for output files")
-    docs_parser.add_argument(
-        "--manager-file-name",
-        dest="manager_file_name",
-        default="manager.xml",
-        help="filename for manager object introspection data",
-    )
-    docs_parser.add_argument(
-        "--pool-file-name",
-        dest="pool_file_name",
-        default="pool.xml",
-        help="filename for pool object introspection data",
-    )
-    docs_parser.add_argument(
-        "--filesystem-file-name",
-        dest="filesystem_file_name",
-        default="filesystem.xml",
-        help="filename for filesystem object introspection data",
-    )
-    docs_parser.add_argument(
-        "--blockdev-file-name",
-        dest="blockdev_file_name",
-        default="blockdev.xml",
-        help="filename for blockdev object introspection data",
-    )
     docs_parser.set_defaults(func=_docs_output)
 
     parser.set_defaults(func=lambda _: parser.error("missing sub-command"))