Fix tuple conversion

pcs/common/interface/dto.py

@@ -4,6 +4,7 @@
 from typing import (
     TYPE_CHECKING,
     Any,
+    Callable,
     Iterable,
     NewType,
     TypeVar,
@@ -39,6 +40,24 @@ class DataclassInstance:
 META_NAME = ToDictMetaKey("META_NAME")
 
 
+DTO_TYPE_HOOKS_MAP: dict[type[Any], Callable[[Any], Any]] = {
+    # JSON does not support tuples, only lists. However, tuples are
+    # used e.g. to express fixed-length structures. If a tuple is
+    # expected and a list is provided, we convert it to a tuple.
+    # Unfortunately, we cannot apply this rule generically to all
+    # tuples, so we must handle specific cases manually.
+    #
+    # Covered cases:
+    # * acl.create_role:
+    #   permission_info_list: list[tuple[str, str, str]]
+    tuple[str, str, str]: lambda v: tuple(v) if isinstance(v, list) else v,
+    # Covered cases:
+    # * resource.get_cibsecrets:
+    #   queries: Sequence[tuple[str, str]]
+    tuple[str, str]: lambda v: tuple(v) if isinstance(v, list) else v,
+}
+
+
 class PayloadConversionError(Exception):
     pass
 
@@ -247,20 +266,7 @@ def from_dict(
                 permissions_types.PermissionGrantedType,
                 permissions_types.PermissionTargetType,
             ],
-            type_hooks={
-                # JSON does not support tuples, only lists. However, tuples are
-                # used e.g. to express fixed-length structures. If a tuple is
-                # expected and a list is provided, we convert it to a tuple.
-                # Unfortunately, we cannot apply this rule generically to all
-                # tuples, so we must handle specific cases manually.
-                #
-                # Covered cases:
-                # * acl.create_role:
-                #   permission_info_list: list[tuple[str, str, str]]
-                tuple[str, str, str]: (
-                    lambda v: tuple(v) if isinstance(v, list) else v
-                ),
-            },
+            type_hooks=DTO_TYPE_HOOKS_MAP,
             strict=strict,
         ),
     )

pcs_test/tier0/common/interface/test_dto.py

@@ -1,7 +1,8 @@
 import importlib
 import pkgutil
+from collections.abc import Sequence
 from dataclasses import dataclass, field, is_dataclass
-from typing import Any
+from typing import Any, Optional
 from unittest import TestCase
 
 from dacite.exceptions import WrongTypeError
@@ -61,8 +62,10 @@ class MyDto3(DataTransferObject):
 
 
 @dataclass
-class AclCreateLike(DataTransferObject):
-    permission_info_list: list[tuple[str, str, str]]
+class TypeHooksDto(DataTransferObject):
+    list_of_tuple_str_str_str: list[tuple[str, str, str]]
+    sequence_of_tuple_str_str: Sequence[tuple[str, str]]
+    optional_tuple_str_str: Optional[tuple[str, str]]
 
 
 class DictName(TestCase):
@@ -117,52 +120,70 @@ def test_nested_from_dict(self):
         self.assertEqual(self.nested_dto, from_dict(MyDto3, self.nested_dict))
 
 
-class FromDictTupleConversion(TestCase):
+class FromDictConversion(TestCase):
+    _valid_payload = dict(
+        list_of_tuple_str_str_str=[["a", "b", "c"]],
+        sequence_of_tuple_str_str=[["a", "b"]],
+        optional_tuple_str_str=["a", "b"],
+    )
+    _valid_dto = TypeHooksDto(
+        list_of_tuple_str_str_str=[("a", "b", "c")],
+        sequence_of_tuple_str_str=[("a", "b")],
+        optional_tuple_str_str=("a", "b"),
+    )
+
     def test_success(self):
         self.assertEqual(
-            AclCreateLike([("read", "xpath", "/cib")]),
-            from_dict(
-                AclCreateLike,
-                dict(permission_info_list=[["read", "xpath", "/cib"]]),
-            ),
+            self._valid_dto,
+            from_dict(TypeHooksDto, self._valid_payload),
         )
 
-    def test_fail_on_different_list_length(self):
-        # Not worth to test misleading exception message:
-        # wrong value type for field "permission_info_list" - should be "list"
-        # instead of value "[('read', 'xpath', '/cib', 'extra string')]" of type
-        # "list"
-        with self.assertRaises(WrongTypeError):
-            from_dict(
-                AclCreateLike,
-                dict(
-                    permission_info_list=[
-                        ["read", "xpath", "/cib", "extra string"]
-                    ]
-                ),
-            )
-
-    def test_fail_on_different_list_item_type(self):
-        # Not worth to test misleading exception message:
-        # wrong value type for field "permission_info_list" - should be "list"
-        # instead of value "[('read', 'id', 1)]" of type "list"
-        with self.assertRaises(WrongTypeError):
-            from_dict(
-                AclCreateLike,
-                dict(permission_info_list=[["read", "id", 1]]),
-            )
-
-    def test_fail_on_no_list(self):
-        # Not worth to test misleading exception message:
-        # wrong value type for field "permission_info_list" - should be "list"
-        # instead of value "['<generator object
-        # _build_value_for_collection.<locals>.<genexpr> at 0x7f7718f5b010>']"
-        # of type "list"
-        with self.assertRaises(WrongTypeError):
-            from_dict(
-                AclCreateLike,
-                dict(permission_info_list=["read xpath /cib"]),
-            )
+    def test_success_optional_none(self):
+        payload = {**self._valid_payload, "optional_tuple_str_str": None}
+        dto = from_dict(TypeHooksDto, payload)
+        self.assertIsNone(dto.optional_tuple_str_str)
+
+    def test_fail_on_wrong_length(self):
+        cases = {
+            "list_of_tuple_str_str_str": [["a", "b", "c", "d"]],
+            "sequence_of_tuple_str_str": [["a", "b", "c"]],
+            "optional_tuple_str_str": ["a", "b", "c"],
+        }
+        for field_name, bad_value in cases.items():
+            with self.subTest(field=field_name):
+                payload = {**self._valid_payload, field_name: bad_value}
+                # Not worth testing exception message as dacite produces
+                # misleading messages for type hook failures
+                with self.assertRaises(WrongTypeError):
+                    from_dict(TypeHooksDto, payload)
+
+    def test_fail_on_wrong_item_type(self):
+        cases = {
+            "list_of_tuple_str_str_str": [["a", "b", 1]],
+            "sequence_of_tuple_str_str": [["a", 1]],
+            "optional_tuple_str_str": ["a", 1],
+        }
+        for field_name, bad_value in cases.items():
+            with self.subTest(field=field_name):
+                payload = {**self._valid_payload, field_name: bad_value}
+                # Not worth testing exception message as dacite produces
+                # misleading messages for type hook failures
+                with self.assertRaises(WrongTypeError):
+                    from_dict(TypeHooksDto, payload)
+
+    def test_fail_on_plain_string(self):
+        cases = {
+            "list_of_tuple_str_str_str": ["a b c"],
+            "sequence_of_tuple_str_str": ["a b"],
+            "optional_tuple_str_str": "a b",
+        }
+        for field_name, bad_value in cases.items():
+            with self.subTest(field=field_name):
+                payload = {**self._valid_payload, field_name: bad_value}
+                # Not worth testing exception message as dacite produces
+                # misleading messages for type hook failures
+                with self.assertRaises(WrongTypeError):
+                    from_dict(TypeHooksDto, payload)
 
 
 @dataclass

pcs_test/tier0/daemon/async_tasks/test_command_mapping.py

@@ -8,9 +8,11 @@
     Iterable,
     get_args,
     get_origin,
+    get_type_hints,
 )
 from unittest import TestCase
 
+from pcs.common.interface.dto import DTO_TYPE_HOOKS_MAP
 from pcs.daemon.async_tasks.worker.command_mapping import COMMAND_MAP
 
 
@@ -26,8 +28,11 @@ def prohibited_types_used(_type, prohibited_types):
             for arg in get_args(_type)
         )
     if is_dataclass(_type):
+        # resolve forward references in type hints, because type-detecting
+        # functions do not work with forward references
+        type_hints = get_type_hints(_type)
         return any(
-            prohibited_types_used(field.type, prohibited_types)
+            prohibited_types_used(type_hints[field.name], prohibited_types)
             for field in fields(_type)
         )
     return False
@@ -37,6 +42,42 @@ def _get_generic(annotation):
     return getattr(annotation, "__origin__", None)
 
 
+def _find_disallowed_types(_type, allowed_types, hooked_origins, _seen=None):
+    if _seen is None:
+        _seen = set()
+    type_id = id(_type)
+    if type_id in _seen:
+        return set()
+    _seen.add(type_id)
+
+    disallowed = set()
+    generic = get_origin(_type)
+    if (
+        generic in hooked_origins
+        and Ellipsis not in get_args(_type)
+        and _type not in allowed_types
+    ):
+        disallowed.add(_type)
+    if generic:
+        for arg in get_args(_type):
+            disallowed.update(
+                _find_disallowed_types(
+                    arg, allowed_types, hooked_origins, _seen
+                )
+            )
+    if is_dataclass(_type):
+        # resolve forward references in type hints, because type-detecting
+        # functions do not work with forward references
+        type_hints = get_type_hints(_type)
+        for field in fields(_type):
+            disallowed.update(
+                _find_disallowed_types(
+                    type_hints[field.name], allowed_types, hooked_origins, _seen
+                )
+            )
+    return disallowed
+
+
 class DaciteTypingCompatibilityTest(TestCase):
     def test_all(self):
         prohibited_types = (Iterable, Container)
@@ -54,3 +95,27 @@ def test_all(self):
                         ),
                         f"Prohibited type used in command: {cmd_name}; argument: {param}; prohibited_types: {prohibited_types}",
                     )
+
+    def test_check_type_hooks_map_types_in_commands(self):
+        allowed_types = set(DTO_TYPE_HOOKS_MAP.keys())
+        hooked_origins = {get_origin(_type) for _type in allowed_types}
+        for cmd_name, cmd in COMMAND_MAP.items():
+            for param in list(inspect.signature(cmd.cmd).parameters.values())[
+                1:
+            ]:
+                if param.annotation == inspect.Parameter.empty:
+                    continue
+                disallowed = _find_disallowed_types(
+                    param.annotation, allowed_types, hooked_origins
+                )
+                self.assertFalse(
+                    disallowed,
+                    f"Type(s) {disallowed} in "
+                    f"command: {cmd_name}; "
+                    f"argument: {param}; "
+                    f"not covered by DTO_TYPE_HOOKS_MAP.keys(): "
+                    f"{allowed_types}. "
+                    "Add the missing type(s) to DTO_TYPE_HOOKS_MAP "
+                    "and update FromDictConversion tests in "
+                    "test_dto.py accordingly.",
+                )