facts: add server.Group based on getent and tests

pyinfra/facts/server.py

@@ -5,7 +5,7 @@
 import shutil
 from datetime import datetime
 from tempfile import mkdtemp
-from typing import Dict, List, Optional
+from typing import Dict, Iterable, List, Optional, Union
 
 from dateutil.parser import parse as parse_date
 from distro import distro
@@ -385,6 +385,57 @@ def process(self, output):
         return sysctls
 
 
+class GroupInfo(TypedDict):
+    name: str
+    password: str
+    gid: int
+    user_list: list[str]
+
+
+def _group_info_from_group_str(info: str) -> GroupInfo:
+    """
+    Parses an entry from /etc/group or a similar source, e.g.
+    'plugdev:x:46:sysadmin,user2' into a GroupInfo dict object
+    """
+
+    fields = info.split(":")
+
+    if len(fields) != 4:
+        raise ValueError(f"Error parsing group '{info}', expected exactly 4 fields separated by :")
+
+    return {
+        "name": fields[0],
+        "password": fields[1],
+        "gid": int(fields[2]),
+        "user_list": fields[3].split(","),
+    }
+
+
+class Group(FactBase[GroupInfo]):
+    """
+    Returns information on a specific group on the system.
+    """
+
+    def command(self, group):
+        # FIXME: the '|| true' ensures 'process' is called, even if
+        #        getent was unable to find information on the group
+        #        There must be a better way to do this !
+        #        e.g. allow facts 'process' method access to the process
+        #        return code ?
+        return f"getent group {group} || true"
+
+    default = None
+
+    def process(self, output: Iterable[str]) -> str:
+        group_string = next(iter(output), None)
+
+        if group_string is None:
+            # This will happen if the group was simply not found
+            return None
+
+        return _group_info_from_group_str(group_string)
+
+
 class Groups(FactBase[List[str]]):
     """
     Returns a list of groups on the system.
@@ -417,7 +468,20 @@ def process(self, output) -> list[str]:
 Crontab = crontab.Crontab
 
 
-class Users(FactBase):
+class UserInfo(TypedDict):
+    name: str
+    comment: str
+    home: str
+    shell: str
+    group: str
+    groups: list[str]
+    uid: int
+    gid: int
+    lastlog: str
+    password: str
+
+
+class Users(FactBase[dict[str, UserInfo]]):
     """
     Returns a dictionary of users -> details.
 
@@ -457,7 +521,7 @@ def command(self) -> str:
 
     default = dict
 
-    def process(self, output):
+    def process(self, output: Iterable[str]) -> dict[str, UserInfo]:
         users = {}
         rex = r"[A-Z][a-z]{2} [A-Z][a-z]{2} {1,2}\d+ .+$"
 

tests/facts/server.Group/group.json

@@ -0,0 +1,14 @@
+{
+    "arg": "plugdev",
+    "command": "getent group plugdev || true",
+    "requires_command": "getent",
+    "output": [
+        "plugdev:x:46:sysadmin,myuser,abc"
+    ],
+    "fact": {
+        "name": "plugdev",
+        "password": "x",
+        "gid": 46,
+        "user_list": ["sysadmin", "myuser", "abc"]
+    }
+}

tests/facts/server.Group/nogroup.json

@@ -0,0 +1,7 @@
+{
+    "arg": "doesnotexist",
+    "command": "getent group doesnotexist || true",
+    "requires_command": "getent",
+    "output": [],
+    "fact": null
+}

tests/test_facts_utils.py

@@ -0,0 +1,21 @@
+import pytest
+
+from pyinfra.facts.server import _group_info_from_group_str
+
+
+def test__group_info_from_group_str() -> None:
+    test_str = "plugdev:x:46:sysadmin,user2"
+    group_info = _group_info_from_group_str(test_str)
+
+    assert group_info["name"] == "plugdev"
+    assert group_info["password"] == "x"
+    assert group_info["gid"] == 46
+    assert group_info["user_list"] == ["sysadmin", "user2"]
+
+
+def test__group_info_from_group_str_empty() -> None:
+    with pytest.raises(ValueError):
+        _group_info_from_group_str("")
+
+    with pytest.raises(ValueError):
+        _group_info_from_group_str("a:b:")