From 72154ef298ba68cbf6f44558d0d6f57ffce37041 Mon Sep 17 00:00:00 2001 From: Rasmus Welander Date: Tue, 9 Dec 2025 13:45:37 +0100 Subject: [PATCH] Add filter creation methods --- cs3client/group.py | 25 ++++++++++++ cs3client/user.py | 26 ++++++++++++ tests/test_group.py | 96 +++++++++++++++++++++++++++++++++++++++++++- tests/test_user.py | 98 +++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 244 insertions(+), 1 deletion(-) diff --git a/cs3client/group.py b/cs3client/group.py index 5015268..f351598 100644 --- a/cs3client/group.py +++ b/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}") diff --git a/cs3client/user.py b/cs3client/user.py index cb2aa7d..7c7e341 100644 --- a/cs3client/user.py +++ b/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}") diff --git a/tests/test_group.py b/tests/test_group.py index 983cc68..5dd5273 100644 --- a/tests/test_group.py +++ b/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 \ No newline at end of file + 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 \ No newline at end of file diff --git a/tests/test_user.py b/tests/test_user.py index 652e9cc..758e094 100644 --- a/tests/test_user.py +++ b/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