Implement ListStorageSpaces call

cs3client/cs3client.py

@@ -19,6 +19,7 @@
 from .app import App
 from .checkpoint import Checkpoint
 from .config import Config
+from .space import Space
 
 
 class CS3Client:
@@ -54,6 +55,7 @@ def __init__(self, config: ConfigParser, config_category: str, log: logging.Logg
             self._config, self._log, self._gateway, self._status_code_handler
         )
         self.share = Share(self._config, self._log, self._gateway, self._status_code_handler)
+        self.space = Space(self._config, self._log, self._gateway, self._status_code_handler)
 
     def _create_channel(self) -> grpc.Channel:
         """

cs3client/space.py

@@ -0,0 +1,99 @@
+"""
+space.py
+
+Authors: Rasmus Welander, Diogo Castro, Giuseppe Lo Presti.
+Emails: rasmus.oscar.welander@cern.ch, diogo.castro@cern.ch, giuseppe.lopresti@cern.ch
+Last updated: 23/02/2026
+"""
+
+import logging
+from cs3.gateway.v1beta1.gateway_api_pb2_grpc import GatewayAPIStub
+
+from .config import Config
+from .statuscodehandler import StatusCodeHandler
+import cs3.storage.provider.v1beta1.spaces_api_pb2 as cs3spp
+import cs3.storage.provider.v1beta1.resources_pb2 as cs3spr
+import cs3.identity.user.v1beta1.resources_pb2 as cs3iur
+
+
+
+class Space:
+    """
+    Space class to handle space related API calls with CS3 Gateway API.
+    """
+
+    def __init__(
+        self,
+        config: Config,
+        log: logging.Logger,
+        gateway: GatewayAPIStub,
+        status_code_handler: StatusCodeHandler,
+    ) -> None:
+        """
+        Initializes the Group class with logger, auth, and gateway stub,
+
+        :param log: Logger instance for logging.
+        :param gateway: GatewayAPIStub instance for interacting with CS3 Gateway.
+        :param auth: An instance of the auth class.
+        """
+        self._log: logging.Logger = log
+        self._gateway: GatewayAPIStub = gateway
+        self._config: Config = config
+        self._status_code_handler: StatusCodeHandler = status_code_handler
+
+    def list_storage_spaces(self, auth_token: tuple, filters) -> list[cs3spr.StorageSpace]:
+        """
+        Find a space based on a filter.
+
+        :param auth_token: tuple in the form ('x-access-token', <token>) (see auth.get_token/auth.check_token)
+        :param filters: Filters to search for.
+        :return: a list of space(s).
+        :raises: NotFoundException (Space not found)
+        :raises: AuthenticationException (Operation not permitted)
+        :raises: UnknownException (Unknown error)
+        """
+        req = cs3spp.ListStorageSpacesRequest(filters=filters)
+        res = self._gateway.ListStorageSpaces(request=req, metadata=[auth_token])
+        self._status_code_handler.handle_errors(res.status, "find storage spaces")
+        self._log.debug(f'msg="Invoked FindStorageSpaces" filter="{filter}" trace="{res.status.trace}"')
+        return res.storage_spaces
+
+    @classmethod
+    def create_storage_space_filter(cls, filter_type: str, space_type: str = None, path: str = None, opaque_id: str = None, user_idp: str = None, user_type: str = None) -> cs3spp.ListStorageSpacesRequest.Filter:
+        """
+        Create a filter for listing storage spaces.
+
+        :param filter_value: Value of the filter.
+        :param filter_type: Type of the filter. Supported values are "ID", "OWNER", "SPACE_TYPE",  "PATH" and "USER".
+        :param space_type: Space type to filter by (required if filter_type is "SPACE_TYPE").
+        :param path: Path to filter by (required if filter_type is "PATH").
+        :param opaque_id: Opaque ID to filter by (required if filter_type is "ID").
+        :param user_idp: User identity provider to filter by (required if filter_type is "OWNER" or "USER").
+        :param user_type: User type to filter by (required if filter_type is "OWNER" or "USER").
+        :param filter_value: Value of the filter.
+        :return: A cs3spp.ListStorageSpacesRequest.Filter object.
+        :raises: ValueError (Unsupported filter type)
+        """
+        try:
+            filter_type_value = cs3spp.ListStorageSpacesRequest.Filter.Type.Value(filter_type)
+
+            if filter_type is None:
+                raise ValueError(f'Unsupported filter type: {filter_type}. Supported values are "Id", "Owner", "SpaceType", "Path" and "User".')
+            if space_type and filter_type == "TYPE_SPACE_TYPE":
+                return cs3spp.ListStorageSpacesRequest.Filter(type=filter_type_value, space_type=space_type)
+            if path and filter_type == "TYPE_PATH":
+                return cs3spp.ListStorageSpacesRequest.Filter(type=filter_type_value, path=path)
+            if user_idp and user_type and opaque_id and filter_type == "TYPE_OWNER":
+                user_type = cs3iur.UserType.Value(user_type.upper())
+                user_id = cs3iur.UserId(idp=user_idp, type=user_type, opaque_id=opaque_id)
+                return cs3spp.ListStorageSpacesRequest.Filter(type=filter_type_value, owner=user_id)
+            if user_idp and user_type and opaque_id and filter_type == "TYPE_USER":
+                user_type = cs3iur.UserType.Value(user_type.upper())
+                user_id = cs3iur.UserId(idp=user_idp, type=user_type, opaque_id=opaque_id)
+                return cs3spp.ListStorageSpacesRequest.Filter(type=filter_type_value, user=user_id)
+            if opaque_id and filter_type == "TYPE_ID":
+                id = cs3spr.StorageSpaceId(opaque_id=opaque_id)
+                return cs3spp.ListStorageSpacesRequest.Filter(type=filter_type_value, id=id)
+        except ValueError as e:
+            raise ValueError(f"Failed to create storage space filter: {e}")
+        return None

tests/test_space.py

@@ -0,0 +1,75 @@
+"""
+test_space.py
+
+Tests that the Space class methods work as expected.
+
+Authors: Rasmus Welander, Diogo Castro, Giuseppe Lo Presti.
+Emails: rasmus.oscar.welander@cern.ch, diogo.castro@cern.ch, giuseppe.lopresti@cern.ch
+Last updated: 23/02/2026
+"""
+
+
+import pytest
+from unittest.mock import Mock, patch
+import cs3.rpc.v1beta1.code_pb2 as cs3code
+import cs3.storage.provider.v1beta1.spaces_api_pb2 as cs3spp
+import cs3.identity.user.v1beta1.resources_pb2 as cs3iur
+import cs3.storage.provider.v1beta1.resources_pb2 as cs3spr
+
+from cs3client.exceptions import (
+    AuthenticationException,
+    UnknownException,
+)
+from .fixtures import (  # noqa: F401 (they are used, the framework is not detecting it)
+    mock_config,
+    mock_logger,
+    mock_gateway,
+    mock_status_code_handler,
+)
+
+@pytest.fixture
+def space_instance(mock_config, mock_logger, mock_gateway, mock_status_code_handler):  # noqa: F811
+    """
+    Fixture for creating a Space instance with mocked dependencies.
+    """
+    from cs3client.space import Space
+
+    return Space(mock_config, mock_logger, mock_gateway, mock_status_code_handler)
+
+
+@pytest.mark.parametrize(
+    "status_code, status_message, expected_exception",
+    [
+        (cs3code.CODE_OK, None, None),
+        (cs3code.CODE_UNAUTHENTICATED, "error", AuthenticationException),
+        (cs3code.CODE_INTERNAL, "error", UnknownException),
+    ],
+)
+def test_list_storage_spaces(space_instance, status_code, status_message, expected_exception):  # noqa: F811
+    mock_response = Mock()
+    mock_response.status.code = status_code
+    mock_response.status.message = status_message
+    mock_response.storage_spaces = ["space1", "space2"]
+    auth_token = ('x-access-token', "some_token")
+
+    with patch.object(space_instance._gateway, "ListStorageSpaces", return_value=mock_response):
+        if expected_exception:
+            with pytest.raises(expected_exception):
+                space_instance.list_storage_spaces(auth_token, filters=[])
+        else:
+            result = space_instance.list_storage_spaces(auth_token, filters=[])
+            assert result == ["space1", "space2"]
+
+@pytest.mark.parametrize(
+    "filter_type, space_type, path, opaque_id, user_idp, user_type, expected_filter",
+    [
+        ("TYPE_SPACE_TYPE", "home", None, None, None, None, cs3spp.ListStorageSpacesRequest.Filter(type="TYPE_SPACE_TYPE", space_type="home")),
+        ("TYPE_PATH", None, "/path/to/space", None, None, None, cs3spp.ListStorageSpacesRequest.Filter(type="TYPE_PATH", path="/path/to/space")),
+        ("TYPE_OWNER", None, None, "opaque_id", "user_idp", "USER_TYPE_PRIMARY", cs3spp.ListStorageSpacesRequest.Filter(type="TYPE_OWNER", owner=cs3iur.UserId(idp="user_idp", type=cs3iur.UserType.Value("USER_TYPE_PRIMARY"), opaque_id="opaque_id"))),
+        ("TYPE_USER", None, None, "opaque_id", "user_idp", "USER_TYPE_PRIMARY", cs3spp.ListStorageSpacesRequest.Filter(type="TYPE_USER", user=cs3iur.UserId(idp="user_idp", type=cs3iur.UserType.Value("USER_TYPE_PRIMARY"), opaque_id="opaque_id"))),
+        ("TYPE_ID", None, None, "opaque_id", None, None, cs3spp.ListStorageSpacesRequest.Filter(type="TYPE_ID", id=cs3spr.StorageSpaceId(opaque_id="opaque_id"))),
+    ],
+)
+def test_create_storage_space_filter(space_instance, filter_type, space_type, path, opaque_id, user_idp, user_type, expected_filter):  # noqa: F811
+    result = space_instance.create_storage_space_filter(filter_type, space_type, path, opaque_id, user_idp, user_type)
+    assert result == expected_filter