facts: Add podman.PodmanSystemInfo and podman.PodmanPs

docs/contributing.md

@@ -50,6 +50,16 @@ scripts/dev-test.sh
 
 Use `pytest` to run the unit tests, or `pytest --cov` to run with coverage. Pull requests are expected to be tested and not drop overall project coverage by >1%.
 
+To limit the pytests to a specific fact or operation:
+
+```sh
+# Only run fact tests for facts.efibootmgr.EFIBootMGR
+pytest tests/test_facts.py -k "efibootmgr.EFIBootMGR"
+
+# Only run operation tests for operations.selinux
+pytest tests/test_operations.py -k "selinux."
+```
+
 #### End to End Tests
 
 The end to end tests are also executed via `pytest` but not selected by default, options/usage:

pyinfra/facts/efibootmgr.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Iterable, List, Optional, Tuple, TypedDict
+
+from pyinfra.api import FactBase
+
+BootEntry = Tuple[bool, str]
+EFIBootMgrInfoDict = TypedDict(
+    "EFIBootMgrInfoDict",
+    {
+        "BootNext": Optional[int],
+        "BootCurrent": Optional[int],
+        "Timeout": Optional[int],
+        "BootOrder": Optional[List[int]],
+        "Entries": Dict[int, BootEntry],
+    },
+)
+
+
+class EFIBootMgr(FactBase[Optional[EFIBootMgrInfoDict]]):
+    """
+    Returns information about the UEFI boot variables:
+
+    .. code:: python
+
+        {
+            "BootNext": 6,
+            "BootCurrent": 6,
+            "Timeout": 0,
+            "BootOrder": [1,4,3],
+            "Entries": {
+                1: (True, "myefi1"),
+                2: (False, "myefi2.efi"),
+                3: (True, "myefi3.efi"),
+                4: (True, "grub2.efi"),
+            },
+        }
+    """
+
+    def requires_command(self, *args: Any, **kwargs: Any) -> str:
+        return "efibootmgr"
+
+    def command(self) -> str:
+        # FIXME: Use '|| true' to properly handle the case where
+        #        'efibootmgr' is run on a non-UEFI system
+        return "efibootmgr || true"
+
+    def process(self, output: Iterable[str]) -> Optional[EFIBootMgrInfoDict]:
+        # This parsing code closely follows the printing code of efibootmgr
+        # at <https://github.com/rhboot/efibootmgr/blob/main/src/efibootmgr.c#L2020-L2048>
+
+        info: EFIBootMgrInfoDict = {
+            "BootNext": None,
+            "BootCurrent": None,
+            "Timeout": None,
+            "BootOrder": [],
+            "Entries": {},
+        }
+
+        output = iter(output)
+
+        line: Optional[str] = next(output, None)
+
+        if line is None:
+            # efibootmgr run on a non-UEFI system, likely printed
+            # "EFI variables are not supported on this system."
+            # to stderr
+            return None
+
+        # 1. Maybe have BootNext
+        if line and line.startswith("BootNext: "):
+            info["BootNext"] = int(line[len("BootNext: ") :], 16)
+            line = next(output, None)
+
+        # 2. Maybe have BootCurrent
+        if line and line.startswith("BootCurrent: "):
+            info["BootCurrent"] = int(line[len("BootCurrent: ") :], 16)
+            line = next(output, None)
+
+        # 3. Maybe have Timeout
+        if line and line.startswith("Timeout: "):
+            info["Timeout"] = int(line[len("Timeout: ") : -len(" seconds")])
+            line = next(output, None)
+
+        # 4. `show_order`
+        if line and line.startswith("BootOrder: "):
+            entries = line[len("BootOrder: ") :]
+            info["BootOrder"] = list(map(lambda x: int(x, 16), entries.split(",")))
+            line = next(output, None)
+
+        # 5. `show_vars`: The actual boot entries
+        while line is not None and line.startswith("Boot"):
+            number = int(line[4:8], 16)
+
+            # Entries marked with a * are active
+            active = line[8:9] == "*"
+
+            # TODO: Maybe split and parse (name vs. arguments ?), might require --verbose ?
+            entry = line[10:]
+            info["Entries"][number] = (active, entry)
+            line = next(output, None)
+
+        # 6. `show_mirror`
+        # Currently not implemented, since I haven't actually encountered this in the wild.
+        if line is not None:
+            raise ValueError(f"Unexpected line '{line}' while parsing")
+
+        return info

pyinfra/facts/mysql.py

@@ -214,8 +214,7 @@ def command(  # type: ignore[override]
             mysql_port,
         )
 
-    @staticmethod
-    def process(output):
+    def process(self, output):
         database_table_privileges = defaultdict(set)
 
         for line in output:

pyinfra/facts/podman.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, Iterable, List, TypeVar
+
+from pyinfra.api import FactBase
+
+T = TypeVar("T")
+
+
+class PodmanFactBase(FactBase[T]):
+    """
+    Base for facts using `podman` to retrieve
+    """
+
+    abstract = True
+
+    def requires_command(self, *args, **kwargs) -> str:
+        return "podman"
+
+
+class PodmanSystemInfo(PodmanFactBase[Dict[str, Any]]):
+    """
+    Output of 'podman system info'
+    """
+
+    def command(self) -> str:
+        return "podman system info --format=json"
+
+    def process(self, output: Iterable[str]) -> Dict[str, Any]:
+        output = json.loads(("").join(output))
+        assert isinstance(output, dict)
+        return output
+
+
+class PodmanPs(PodmanFactBase[List[Dict[str, Any]]]):
+    """
+    Output of 'podman ps'
+    """
+
+    def command(self) -> str:
+        return "podman ps --format=json --all"
+
+    def process(self, output: Iterable[str]) -> List[Dict[str, Any]]:
+        output = json.loads(("").join(output))
+        assert isinstance(output, list)
+        return output  # type: ignore

pyinfra/facts/server.py

@@ -32,8 +32,7 @@ class Home(FactBase[Optional[str]]):
     Returns the home directory of the given user, or the current user if no user is given.
     """
 
-    @staticmethod
-    def command(user=""):
+    def command(self, user=""):
         return f"echo ~{user}"
 
 
@@ -124,8 +123,7 @@ class Command(FactBase[str]):
     Returns the raw output lines of a given command.
     """
 
-    @staticmethod
-    def command(command):
+    def command(self, command):
         return command
 
 
@@ -134,8 +132,7 @@ class Which(FactBase[Optional[str]]):
     Returns the path of a given command according to `command -v`, if available.
     """
 
-    @staticmethod
-    def command(command):
+    def command(self, command):
         return "command -v {0} || true".format(command)
 
 

pyinfra/facts/zfs.py

@@ -19,39 +19,34 @@ class Pools(FactBase):
     def command(self):
         return "zpool get -H all"
 
-    @staticmethod
-    def process(output):
+    def process(self, output):
         return _process_zfs_props_table(output)
 
 
 class Datasets(FactBase):
     def command(self):
         return "zfs get -H all"
 
-    @staticmethod
-    def process(output):
+    def process(self, output):
         return _process_zfs_props_table(output)
 
 
 class Filesystems(ShortFactBase):
     fact = Datasets
 
-    @staticmethod
-    def process_data(data):
+    def process_data(self, data):
         return {name: props for name, props in data.items() if props.get("type") == "filesystem"}
 
 
 class Snapshots(ShortFactBase):
     fact = Datasets
 
-    @staticmethod
-    def process_data(data):
+    def process_data(self, data):
         return {name: props for name, props in data.items() if props.get("type") == "snapshot"}
 
 
 class Volumes(ShortFactBase):
     fact = Datasets
 
-    @staticmethod
-    def process_data(data):
+    def process_data(self, data):
         return {name: props for name, props in data.items() if props.get("type") == "volume"}

tests/facts/efibootmgr.EFIBootMgr/boot_entries.json

@@ -0,0 +1,18 @@
+{
+    "command": "efibootmgr || true",
+    "requires_command": "efibootmgr",
+    "output": [
+        "BootCurrent: 0002",
+        "BootOrder: 0002",
+        "Boot0002* debian"
+    ],
+    "fact": {
+        "BootNext": null,
+        "BootCurrent": 2,
+        "Timeout": null,
+        "BootOrder": [2],
+        "Entries": {
+            "2": [true, "debian"]
+        }
+    }
+}

tests/facts/efibootmgr.EFIBootMgr/boot_entries2.json

@@ -0,0 +1,20 @@
+{
+    "command": "efibootmgr || true",
+    "requires_command": "efibootmgr",
+    "output": [
+        "BootCurrent: 0000",
+        "BootOrder: 0000,0003",
+        "Boot0000* debian",
+        "Boot0003* EFI Fixed Disk Boot Device 1"
+    ],
+    "fact": {
+        "BootNext": null,
+        "BootCurrent": 0,
+        "Timeout": null,
+        "BootOrder": [0,3],
+        "Entries": {
+            "0": [true, "debian"],
+            "3": [true, "EFI Fixed Disk Boot Device 1"]
+        }
+    }
+}

tests/facts/efibootmgr.EFIBootMgr/boot_entries_complex.json

@@ -0,0 +1,30 @@
+{
+    "command": "efibootmgr || true",
+    "requires_command": "efibootmgr",
+    "output": [
+        "BootNext: 0006",
+        "BootCurrent: 0006",
+        "Timeout: 2 seconds",
+        "BootOrder: 0006,0001,0002,0003,0007,0000",
+        "Boot0000* Diagnostic Program",
+        "Boot0001* Windows Boot Manager",
+        "Boot0002* UEFI: PXE IP4 Realtek PCIe GBE Family Controller",
+        "Boot0003* UEFI: PXE IP6 Realtek PCIe GBE Family Controller",
+        "Boot0006* debian",
+        "Boot0007* debian"
+    ],
+    "fact": {
+        "BootNext": 6,
+        "BootCurrent": 6,
+        "Timeout": 2,
+        "BootOrder": [6,1,2,3,7,0],
+        "Entries": {
+            "0": [true, "Diagnostic Program"],
+            "1": [true, "Windows Boot Manager"],
+            "2": [true, "UEFI: PXE IP4 Realtek PCIe GBE Family Controller"],
+            "3": [true, "UEFI: PXE IP6 Realtek PCIe GBE Family Controller"],
+            "6": [true, "debian"],
+            "7": [true, "debian"]
+        }
+    }
+}

tests/facts/efibootmgr.EFIBootMgr/not_uefi.json

@@ -0,0 +1,6 @@
+{
+    "command": "efibootmgr || true",
+    "requires_command": "efibootmgr",
+    "output": [],
+    "fact": null
+}

tests/facts/podman.PodmanPs/empty.json

@@ -0,0 +1,8 @@
+{
+  "command": "podman ps --format=json --all",
+  "requires_command": "podman",
+  "output": [
+    "[]"
+  ],
+  "fact": []
+}