Add hostmeta fact collection

humitifier/infra/facts/__init__.py

@@ -1,6 +1,7 @@
 import asyncio
 from .blocks import Blocks
 from .groups import Groups
+from .hostmeta import HostMeta
 from .hostnamectl import HostnameCtl
 from .memory import Memory
 from .package_list import PackageList
@@ -17,6 +18,7 @@
     Blocks,
     Groups,
     HostnameCtl,
+    HostMeta,
     Memory,
     PackageList,
     Uptime,
@@ -27,6 +29,7 @@
     Blocks,
     Groups,
     HostnameCtl,
+    HostMeta,
     Memory,
     PackageList,
     Uptime,
@@ -37,6 +40,7 @@
     Blocks.alias: Blocks,
     Groups.alias: Groups,
     HostnameCtl.alias: HostnameCtl,
+    HostMeta.alias: HostMeta,
     Memory.alias: Memory,
     PackageList.alias: PackageList,
     Uptime.alias: Uptime,

humitifier/infra/facts/hostmeta.py

@@ -0,0 +1,46 @@
+import json
+from dataclasses import asdict, dataclass
+
+
+@dataclass
+class HostMeta:
+    department: str | None
+    contact: str | None
+    update_policy: dict[str, bool] | None
+    webdav: str | None
+    vhosts: list[dict] | None
+    fileservers: list[str] | None
+    databases: dict[dict[str, list[str]]] | None
+
+    @classmethod
+    @property
+    def alias(cls) -> str:
+        return "hostmeta"
+
+    @classmethod
+    def from_stdout(cls, output: list[str]) -> "HostMeta":
+        base_args = {
+            "department": None,
+            "contact": None,
+            "update_policy": None,
+            "webdav": None,
+            "vhosts": None,
+            "fileservers": None,
+            "databases": None,
+        }
+        if output[0] == "cat: /hum/doc/server_facts.json: No such file or directory":
+            return cls(**base_args)
+        json_str = "\n".join(output)
+        json_args = json.loads(json_str)
+        return cls(**{**base_args, **json_args})
+
+    @staticmethod
+    def ssh_command(_) -> str:
+        return "cat /hum/doc/server_facts.json"
+
+    @classmethod
+    def from_sql(cls, sql_data) -> "HostMeta":
+        return cls(**sql_data)
+
+    def to_sql(self):
+        return asdict(self)