refactor: ha_cluster_info: pcs permissions

split responsibilities between a loader and an exporter

Author: tomjelinek

library/ha_cluster_info.py

@@ -138,9 +138,13 @@ def export_pcsd_configuration() -> Dict[str, Any]:
     """
     result: dict[str, Any] = dict()
 
-    pcs_permissions = loader.get_pcsd_local_cluster_permissions()
-    if pcs_permissions is not None:
-        result["ha_cluster_pcs_permission_list"] = pcs_permissions
+    pcsd_settings_dict = loader.get_pcsd_settings_conf()
+    if pcsd_settings_dict is not None:
+        pcs_permissions = exporter.export_pcs_permission_list(
+            pcsd_settings_dict
+        )
+        if pcs_permissions is not None:
+            result["ha_cluster_pcs_permission_list"] = pcs_permissions
 
     return result
 

module_utils/ha_cluster_lsr/info/exporter.py

@@ -12,7 +12,7 @@
 __metaclass__ = type
 
 from contextlib import contextmanager
-from typing import Any, Dict, Iterator, List
+from typing import Any, Dict, Iterator, List, Optional
 
 
 class JsonMissingKey(Exception):
@@ -216,3 +216,32 @@ def export_cluster_nodes(
             # finish one node export
             node_list.append(one_node)
         return node_list
+
+
+def export_pcs_permission_list(
+    pcs_settings_conf_dict: Dict[str, Any],
+) -> Optional[List[Dict[str, Any]]]:
+    """
+    Extract local cluster permissions from pcs_settings config or None on error
+
+    pcs_settings -- JSON parsed pcs_settings.conf
+    """
+    # Currently, only format version 2 is in use, so we don't check for file
+    # format version
+    result: List[Dict[str, Any]] = []
+    with _handle_missing_key(pcs_settings_conf_dict, "pcs_settings.conf"):
+        permission_dict = pcs_settings_conf_dict["permissions"]
+        if not isinstance(permission_dict, dict):
+            return None
+        permission_list = permission_dict["local_cluster"]
+        if not isinstance(permission_list, list):
+            return None
+        for permission_item in permission_list:
+            result.append(
+                {
+                    "type": permission_item["type"],
+                    "name": permission_item["name"],
+                    "allow_list": list(permission_item["allow"]),
+                }
+            )
+        return result

module_utils/ha_cluster_lsr/info/loader.py

@@ -218,42 +218,19 @@ def get_pcsd_known_hosts() -> Dict[str, str]:
         raise JsonParseError(str(e), "not logging data", "known hosts") from e
 
 
-def get_pcsd_local_cluster_permissions() -> Optional[List[Dict[str, Any]]]:
+def get_pcsd_settings_conf() -> Optional[Dict[str, Any]]:
     """
-    Load pcsd local cluster permissions
+    Load pcsd config file pcs_settings.conf
     """
     # There is no pcs interface for the file yet, so we parse the file directly.
     if not os.path.exists(PCSD_SETTINGS_PATH):
         return None
 
-    result: List[Dict[str, Any]] = []
     try:
         with open(
             PCSD_SETTINGS_PATH, "r", encoding="utf-8"
         ) as pcsd_settings_file:
-            pcsd_settings = json.load(pcsd_settings_file)
-        permission_dict = pcsd_settings.get("permissions")
-        if not isinstance(permission_dict, dict):
-            return result
-        permission_list = permission_dict.get("local_cluster")
-        if not isinstance(permission_list, list):
-            return result
-        for permission_item in permission_list:
-            if (
-                "type" not in permission_item
-                or "name" not in permission_item
-                or "allow" not in permission_item
-                or not isinstance(permission_item["allow"], list)
-            ):
-                continue
-            result.append(
-                {
-                    "type": permission_item["type"],
-                    "name": permission_item["name"],
-                    "allow_list": list(permission_item["allow"]),
-                }
-            )
-        return result
+            file_data = pcsd_settings_file.read()
+            return json.loads(file_data)
     except json.JSONDecodeError as e:
-        # cannot show actual data as they contain sensitive information
-        raise JsonParseError(str(e), "not logging data", "pcsd settings") from e
+        raise JsonParseError(str(e), file_data, "pcsd settings") from e

tests/unit/test_ha_cluster_info.py

@@ -371,48 +371,96 @@ def test_non_rhel(self) -> None:
 class ExportPcsdConfiguration(TestCase):
     maxDiff = None
 
-    @mock.patch("ha_cluster_info.loader.get_pcsd_local_cluster_permissions")
-    def test_no_permissions(self, mock_load_pcsd_permisions: mock.Mock) -> None:
-        mock_load_pcsd_permisions.return_value = None
-
-        self.assertEqual(
-            ha_cluster_info.export_pcsd_configuration(),
-            {},
-        )
-
-    @mock.patch("ha_cluster_info.loader.get_pcsd_local_cluster_permissions")
+    @mock.patch("ha_cluster_info.loader.get_pcsd_settings_conf")
     def test_permissions_defined(
         self, mock_load_pcsd_permisions: mock.Mock
     ) -> None:
-        mock_load_pcsd_permisions.return_value = [
-            {
-                "type": "group",
-                "name": "haclient",
-                "allow": ["grant", "read", "write"],
-            },
-            {
-                "type": "user",
-                "name": "admin",
-                "allow": ["full"],
-            },
-        ]
+        mock_load_pcsd_permisions.return_value = {
+            "permissions": {
+                "local_cluster": [
+                    {
+                        "type": "group",
+                        "name": "haclient",
+                        "allow": ["grant", "read", "write"],
+                    },
+                    {
+                        "type": "user",
+                        "name": "admin",
+                        "allow": ["full"],
+                    },
+                ]
+            }
+        }
 
         self.assertEqual(
             ha_cluster_info.export_pcsd_configuration(),
             {
-                "ha_cluster_pcs_permission_list": mock_load_pcsd_permisions.return_value,
+                "ha_cluster_pcs_permission_list": [
+                    {
+                        "type": "group",
+                        "name": "haclient",
+                        "allow_list": ["grant", "read", "write"],
+                    },
+                    {
+                        "type": "user",
+                        "name": "admin",
+                        "allow_list": ["full"],
+                    },
+                ]
             },
         )
 
-    @mock.patch("ha_cluster_info.loader.get_pcsd_local_cluster_permissions")
+    @mock.patch("ha_cluster_info.loader.get_pcsd_settings_conf")
     def test_empty_permissions_defined(
         self, mock_load_pcsd_permisions: mock.Mock
     ) -> None:
-        mock_load_pcsd_permisions.return_value = []
+        mock_load_pcsd_permisions.return_value = {
+            "permissions": {
+                "local_cluster": [],
+            }
+        }
 
         self.assertEqual(
             ha_cluster_info.export_pcsd_configuration(),
             {
                 "ha_cluster_pcs_permission_list": [],
             },
         )
+
+    @mock.patch("ha_cluster_info.loader.get_pcsd_settings_conf")
+    def test_permission_load_error(
+        self, mock_load_pcsd_permisions: mock.Mock
+    ) -> None:
+        mock_load_pcsd_permisions.return_value = None
+
+        self.assertEqual(
+            ha_cluster_info.export_pcsd_configuration(),
+            {},
+        )
+
+    @mock.patch("ha_cluster_info.loader.get_pcsd_settings_conf")
+    def test_permission_load_bad_format(
+        self, mock_load_pcsd_permisions: mock.Mock
+    ) -> None:
+        mock_load_pcsd_permisions.return_value = {
+            "permissions": {
+                "local": [
+                    {
+                        "type": "group",
+                        "name": "haclient",
+                        "allow": ["grant", "read", "write"],
+                    },
+                ]
+            }
+        }
+
+        with self.assertRaises(ha_cluster_info.exporter.JsonMissingKey) as cm:
+            ha_cluster_info.export_pcsd_configuration()
+        self.assertEqual(
+            cm.exception.kwargs,
+            dict(
+                data=mock_load_pcsd_permisions.return_value,
+                key="local_cluster",
+                data_desc="pcs_settings.conf",
+            ),
+        )

tests/unit/test_info_exporter.py

@@ -540,3 +540,87 @@ def test_pcs_nodes(self) -> None:
                 ),
             ],
         )
+
+
+class ExportPcsPermissionList(TestCase):
+    maxDiff = None
+
+    def test_minimal(self) -> None:
+        pcs_settings_dict: Dict[str, Any] = {
+            "permissions": {
+                "local_cluster": [],
+            }
+        }
+
+        self.assertEqual(
+            exporter.export_pcs_permission_list(pcs_settings_dict),
+            [],
+        )
+
+    def test_success(self) -> None:
+        pcs_settings_dict: Dict[str, Any] = {
+            "permissions": {
+                "local_cluster": [
+                    {"name": "test1", "type": "user", "allow": []},
+                    {"name": "test2", "type": "user", "allow": ["read"]},
+                    {
+                        "name": "test3",
+                        "type": "group",
+                        "allow": ["write", "grant"],
+                    },
+                ]
+            }
+        }
+
+        self.assertEqual(
+            exporter.export_pcs_permission_list(pcs_settings_dict),
+            [
+                {"name": "test1", "type": "user", "allow_list": []},
+                {"name": "test2", "type": "user", "allow_list": ["read"]},
+                {
+                    "name": "test3",
+                    "type": "group",
+                    "allow_list": ["write", "grant"],
+                },
+            ],
+        )
+
+    def assert_missing_key(
+        self, pcs_settings_dict: Dict[str, Any], key: str
+    ) -> None:
+        with self.assertRaises(exporter.JsonMissingKey) as cm:
+            exporter.export_pcs_permission_list(pcs_settings_dict)
+        self.assertEqual(
+            cm.exception.kwargs,
+            dict(
+                data=pcs_settings_dict,
+                key=key,
+                data_desc="pcs_settings.conf",
+            ),
+        )
+
+    def test_missing_key(self) -> None:
+        self.assert_missing_key(
+            dict(),
+            "permissions",
+        )
+        self.assert_missing_key(
+            dict(permissions=dict()),
+            "local_cluster",
+        )
+        self.assert_missing_key(
+            dict(permissions=dict(local_cluster=[dict()])),
+            "type",
+        )
+        self.assert_missing_key(
+            dict(permissions=dict(local_cluster=[dict(type="user")])),
+            "name",
+        )
+        self.assert_missing_key(
+            dict(
+                permissions=dict(
+                    local_cluster=[dict(type="user", name="user1")]
+                )
+            ),
+            "allow",
+        )