Add filter creation methods

cs3client/group.py

@@ -7,6 +7,8 @@
 """
 
 import logging
+from typing import Optional
+
 import cs3.identity.group.v1beta1.resources_pb2 as cs3igr
 import cs3.identity.group.v1beta1.group_api_pb2 as cs3ig
 import cs3.identity.user.v1beta1.resources_pb2 as cs3iur
@@ -122,3 +124,26 @@ def find_groups(self, auth_token: tuple, filters) -> list[cs3igr.Group]:
         self._status_code_handler.handle_errors(res.status, "find groups")
         self._log.debug(f'msg="Invoked FindGroups" filter="{filter}" trace="{res.status.trace}"')
         return res.groups
+
+    @classmethod
+    def create_find_group_filter(cls, filter_type: str, query: Optional[str], group_type: Optional[str]) -> cs3ig.Filter:
+        """
+        Create a filter for finding groups.
+
+        :param filter_type: The type of filter to create. Supported types: TYPE_GROUPTYPE, TYPE_QUERY.
+        :param query: The query string for TYPE_QUERY filter, or GROUP_TYPE_FEDERATED/GROUP_TYPE_REGULAR for TYPE_GROUPTYPE.
+        :return: A filter object.
+        :raises: ValueError (Unsupported filter type)
+        """
+        filter_type_value = cs3ig.Filter.Type.Value(filter_type.upper())
+        if filter_type_value == cs3ig.Filter.Type.TYPE_QUERY:
+            if query is None:
+                raise ValueError("query must be provided for TYPE_QUERY filter")
+            return cs3ig.Filter(type=filter_type_value, query=query)
+        elif filter_type_value == cs3ig.Filter.Type.TYPE_GROUPTYPE:
+            if group_type is None:
+                raise ValueError("group_type must be provided for TYPE_GROUPTYPE filter")
+            group_type_value = cs3igr.GroupType.Value(group_type.upper())
+            return cs3ig.Filter(type=filter_type_value, grouptype=group_type_value)
+        else:
+            raise ValueError(f"Unsupported filter type: {filter_type}")

cs3client/user.py

@@ -7,6 +7,8 @@
 """
 
 import logging
+from typing import Optional
+
 import cs3.identity.user.v1beta1.resources_pb2 as cs3iur
 import cs3.identity.user.v1beta1.user_api_pb2 as cs3iu
 from cs3.gateway.v1beta1.gateway_api_pb2_grpc import GatewayAPIStub
@@ -106,3 +108,27 @@ def find_users(self, auth_token: tuple, filters) -> list[cs3iur.User]:
         self._status_code_handler.handle_errors(res.status, "find users")
         self._log.debug(f'msg="Invoked FindUsers" filter="{filter}" trace="{res.status.trace}"')
         return res.users
+
+    @classmethod
+    def create_find_user_filter(cls, filter_type: str, query: Optional[str] = None, user_type: Optional[str] = None) -> cs3iu.Filter:
+        """
+        Create a filter for finding users.
+
+        :param filter_type: The type of filter to create. Supported types: TYPE_QUERY, TYPE_USER_TYPE.
+        :param query: The query string for TYPE_QUERY filter.
+        :param user_type: The user type for TYPE_USER_TYPE filter. Supported types: USER_TYPE_PRIMARY,
+            USER_TYPE_SECONDARY, USER_TYPE_SERVICE, USER_TYPE_GUEST, USER_TYPE_FEDERATED, USER_TYPE_LIGHTWEIGHT,
+            USER_TYPE_SPACE_OWNER.
+        :return: A filter object.
+        :raises: ValueError (Unsupported filter type)
+        """
+        filter_type = cs3iu.Filter.Type.Value(filter_type.upper())
+        if filter_type == cs3iu.Filter.Type.TYPE_QUERY:
+            return cs3iu.Filter(type=filter_type, query=query)
+        elif filter_type == cs3iu.Filter.Type.TYPE_USERTYPE:
+            if user_type is None:
+                raise ValueError("user_type must be provided for TYPE_USERTYPE filter")
+            user_type = cs3iur.UserType.Value(user_type.upper())
+            return cs3iu.Filter(type=filter_type, usertype=user_type)
+        else:
+            raise ValueError(f"Unsupported filter type: {filter_type}")

tests/test_group.py

@@ -13,6 +13,9 @@
 import cs3.rpc.v1beta1.code_pb2 as cs3code
 import cs3.identity.group.v1beta1.group_api_pb2 as cs3ig
 import cs3.identity.group.v1beta1.resources_pb2 as cs3igr
+from cs3client.group import Group
+
+
 
 from cs3client.exceptions import (
     AuthenticationException,
@@ -192,4 +195,95 @@ def test_get_group_by_claim(
                 group_instance.get_group_by_claim(auth_token, claim, value)
         else:
             result = group_instance.get_group_by_claim(auth_token, claim, value)
-            assert result == group_data
+            assert result == group_data
+
+
+@pytest.mark.parametrize(
+    "filter_type, query, group_type, expected_exception",
+    [
+        ("TYPE_QUERY", "test_group", None, None),
+        ("TYPE_GROUPTYPE", None, "GROUP_TYPE_FEDERATED", None),
+        ("TYPE_GROUPTYPE", None, "GROUP_TYPE_REGULAR", None),
+        ("TYPE_GROUPTYPE", None, None, ValueError),
+        ("TYPE_INVALID", "test", None, ValueError),
+    ],
+)
+def test_create_find_group_filter(filter_type, query, group_type, expected_exception):
+    """Test the create_find_group_filter classmethod."""
+
+    if expected_exception:
+        with pytest.raises(expected_exception):
+            Group.create_find_group_filter(filter_type, query, group_type)
+    else:
+        result = Group.create_find_group_filter(filter_type, query, group_type)
+        assert result is not None
+        assert isinstance(result, cs3ig.Filter)
+
+
+@pytest.mark.parametrize(
+    "status_code, status_message, expected_exception, groups, filter_type, query, group_type",
+    [
+        (cs3code.CODE_OK, None, None, [Mock(), Mock()], "TYPE_QUERY", "test_group", None),
+        (cs3code.CODE_OK, None, None, [Mock()], "TYPE_GROUPTYPE", None, "GROUP_TYPE_FEDERATED"),
+        (cs3code.CODE_OK, None, None, [Mock()], "TYPE_GROUPTYPE", None, "GROUP_TYPE_REGULAR"),
+        (cs3code.CODE_NOT_FOUND, "error", NotFoundException, None, "TYPE_QUERY", "nonexistent", None),
+        (cs3code.CODE_UNAUTHENTICATED, "error", AuthenticationException, None, "TYPE_QUERY", "test", None),
+        (-2, "error", UnknownException, None, "TYPE_GROUPTYPE", None, "GROUP_TYPE_REGULAR"),
+    ],
+)
+def test_find_groups_with_filter_creation(
+    group_instance, status_code, status_message, expected_exception, groups, filter_type, query, group_type  # noqa: F811 (not a redefinition)
+):
+    """Test find_groups using the create_find_group_filter classmethod."""
+
+    # Create filter using the classmethod
+    group_filter = Group.create_find_group_filter(filter_type, query, group_type)
+    filters = [group_filter]
+
+    mock_response = Mock()
+    mock_response.status.code = status_code
+    mock_response.status.message = status_message
+    mock_response.groups = groups
+    auth_token = ('x-access-token', "some_token")
+
+    with patch.object(group_instance._gateway, "FindGroups", return_value=mock_response):
+        if expected_exception:
+            with pytest.raises(expected_exception):
+                group_instance.find_groups(auth_token, filters)
+        else:
+            result = group_instance.find_groups(auth_token, filters)
+            assert result == groups
+
+
+@pytest.mark.parametrize(
+    "status_code, status_message, expected_exception, groups",
+    [
+        (cs3code.CODE_OK, None, None, [Mock(), Mock()]),
+        (cs3code.CODE_NOT_FOUND, "error", NotFoundException, None),
+        (cs3code.CODE_UNAUTHENTICATED, "error", AuthenticationException, None),
+        (-2, "error", UnknownException, None),
+    ],
+)
+def test_find_groups_with_multiple_filters(
+    group_instance, status_code, status_message, expected_exception, groups  # noqa: F811 (not a redefinition)
+):
+    """Test find_groups with multiple filters using the create_find_group_filter classmethod."""
+
+    # Create multiple filters using the classmethod
+    filter1 = Group.create_find_group_filter("TYPE_QUERY", "test", None)
+    filter2 = Group.create_find_group_filter("TYPE_GROUPTYPE", None, "GROUP_TYPE_FEDERATED")
+    filters = [filter1, filter2]
+
+    mock_response = Mock()
+    mock_response.status.code = status_code
+    mock_response.status.message = status_message
+    mock_response.groups = groups
+    auth_token = ('x-access-token', "some_token")
+
+    with patch.object(group_instance._gateway, "FindGroups", return_value=mock_response):
+        if expected_exception:
+            with pytest.raises(expected_exception):
+                group_instance.find_groups(auth_token, filters)
+        else:
+            result = group_instance.find_groups(auth_token, filters)
+            assert result == groups

tests/test_user.py

@@ -13,6 +13,8 @@
 import cs3.rpc.v1beta1.code_pb2 as cs3code
 import cs3.identity.user.v1beta1.user_api_pb2 as cs3iu
 import cs3.identity.user.v1beta1.resources_pb2 as cs3iur
+from cs3client.user import User
+
 
 from cs3client.exceptions import (
     AuthenticationException,
@@ -145,3 +147,99 @@ def test_find_users(
         else:
             result = user_instance.find_users(auth_token, filters)
             assert result == users
+
+
+@pytest.mark.parametrize(
+    "filter_type, query, user_type, expected_exception",
+    [
+        ("TYPE_QUERY", "test_user", None, None),
+        ("TYPE_USERTYPE", None, "USER_TYPE_PRIMARY", None),
+        ("TYPE_USERTYPE", None, "USER_TYPE_SECONDARY", None),
+        ("TYPE_USERTYPE", None, "USER_TYPE_SERVICE", None),
+        ("TYPE_USERTYPE", None, "USER_TYPE_GUEST", None),
+        ("TYPE_USERTYPE", None, "USER_TYPE_FEDERATED", None),
+        ("TYPE_USERTYPE", None, "USER_TYPE_LIGHTWEIGHT", None),
+        ("TYPE_USERTYPE", None, "USER_TYPE_SPACE_OWNER", None),
+        ("TYPE_USERTYPE", None, None, ValueError),
+        ("TYPE_INVALID", "test", None, ValueError),
+    ],
+)
+def test_create_find_user_filter(filter_type, query, user_type, expected_exception):
+    """Test the create_find_user_filter classmethod."""
+
+    if expected_exception:
+        with pytest.raises(expected_exception):
+            User.create_find_user_filter(filter_type, query, user_type)
+    else:
+        result = User.create_find_user_filter(filter_type, query, user_type)
+        assert result is not None
+        assert isinstance(result, cs3iu.Filter)
+
+
+@pytest.mark.parametrize(
+    "status_code, status_message, expected_exception, users, filter_type, query, user_type",
+    [
+        (cs3code.CODE_OK, None, None, [Mock(), Mock()], "TYPE_QUERY", "test_user", None),
+        (cs3code.CODE_OK, None, None, [Mock()], "TYPE_USERTYPE", None, "USER_TYPE_PRIMARY"),
+        (cs3code.CODE_OK, None, None, [Mock()], "TYPE_USERTYPE", None, "USER_TYPE_FEDERATED"),
+        (cs3code.CODE_NOT_FOUND, "error", NotFoundException, None, "TYPE_QUERY", "nonexistent", None),
+        (cs3code.CODE_UNAUTHENTICATED, "error", AuthenticationException, None, "TYPE_QUERY", "test", None),
+        (-2, "error", UnknownException, None, "TYPE_USERTYPE", None, "USER_TYPE_SERVICE"),
+    ],
+)
+def test_find_users_with_filter_creation(
+    user_instance, status_code, status_message, expected_exception, users, filter_type, query, user_type  # noqa: F811 (not a redefinition)
+):
+    """Test find_users using the create_find_user_filter classmethod."""
+
+    # Create filter using the classmethod
+    user_filter = User.create_find_user_filter(filter_type, query, user_type)
+    filters = [user_filter]
+
+    mock_response = Mock()
+    mock_response.status.code = status_code
+    mock_response.status.message = status_message
+    mock_response.users = users
+    auth_token = ('x-access-token', "some_token")
+
+    with patch.object(user_instance._gateway, "FindUsers", return_value=mock_response):
+        if expected_exception:
+            with pytest.raises(expected_exception):
+                user_instance.find_users(auth_token, filters)
+        else:
+            result = user_instance.find_users(auth_token, filters)
+            assert result == users
+
+
+@pytest.mark.parametrize(
+    "status_code, status_message, expected_exception, users",
+    [
+        (cs3code.CODE_OK, None, None, [Mock(), Mock()]),
+        (cs3code.CODE_NOT_FOUND, "error", NotFoundException, None),
+        (cs3code.CODE_UNAUTHENTICATED, "error", AuthenticationException, None),
+        (-2, "error", UnknownException, None),
+    ],
+)
+def test_find_users_with_multiple_filters(
+    user_instance, status_code, status_message, expected_exception, users  # noqa: F811 (not a redefinition)
+):
+    """Test find_users with multiple filters using the create_find_user_filter classmethod."""
+
+    # Create multiple filters using the classmethod
+    filter1 = User.create_find_user_filter("TYPE_QUERY", "test", None)
+    filter2 = User.create_find_user_filter("TYPE_USERTYPE", None, "USER_TYPE_PRIMARY")
+    filters = [filter1, filter2]
+
+    mock_response = Mock()
+    mock_response.status.code = status_code
+    mock_response.status.message = status_message
+    mock_response.users = users
+    auth_token = ('x-access-token', "some_token")
+
+    with patch.object(user_instance._gateway, "FindUsers", return_value=mock_response):
+        if expected_exception:
+            with pytest.raises(expected_exception):
+                user_instance.find_users(auth_token, filters)
+        else:
+            result = user_instance.find_users(auth_token, filters)
+            assert result == users