Add data model for Workspaces

geoservercloud/models/__init__.py

@@ -0,0 +1,5 @@
+from .common import KeyDollarListDict
+from .workspace import Workspace
+from .workspaces import Workspaces
+
+__all__ = ["KeyDollarListDict", "Workspaces", "Workspace"]

geoservercloud/models/common.py

@@ -0,0 +1,39 @@
+import json
+import logging
+
+log = logging.getLogger()
+
+
+class KeyDollarListDict(dict):
+    def __init__(self, input_list=None, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.key_prefix = "@key"
+        self.value_prefix = "$"
+        if input_list:
+            self.deserialize(input_list)
+        log.debug(self)
+
+    def deserialize(self, input_list):
+        for item in input_list:
+            key = item[self.key_prefix]
+            if self.value_prefix in item:
+                value = item[self.value_prefix]
+            else:
+                value = None
+            super().__setitem__(key, value)
+
+    def serialize(self):
+        return [
+            {self.key_prefix: key, self.value_prefix: value}
+            for key, value in self.items()
+        ]
+
+    def __repr__(self) -> str:
+        return str(self.serialize())
+
+    def __str__(self):
+        return json.dumps(self.serialize())
+
+    # def update(self, other: dict):
+    #     for key, value in other.items():
+    #         super().__setitem__(key, value)

geoservercloud/models/workspace.py

@@ -0,0 +1,31 @@
+import logging
+
+import jsonschema
+import requests
+
+log = logging.getLogger()
+
+
+class Workspace:
+
+    def __init__(self, name, isolated: bool = False) -> None:
+        self.name = name
+        self.isolated = isolated
+
+    def put_payload(self):
+        payload = {"workspace": {"name": self.name}}
+        if self.isolated:
+            payload["workspace"]["isolated"] = self.isolated
+        return payload
+
+    def post_payload(self):
+        return self.put_payload()
+
+    @classmethod
+    def from_response(cls, response):
+        json_data = response.json()
+        return cls(
+            json_data.get("workspace", {}).get("name", None),
+            json_data.get("workspace", {}).get("isolated", False),
+        )
+        return cls(json_data.get("workspace", {}).get("name", None))

geoservercloud/models/workspaces.py

@@ -0,0 +1,45 @@
+import logging
+
+import jsonschema
+import requests
+
+log = logging.getLogger()
+
+
+class Workspaces:
+
+    def __init__(self, workspaces: dict = {}) -> None:
+        self._workspaces = workspaces
+
+    @classmethod
+    def validate(self, response):
+        try:
+            jsonschema.validate(response, self.response_schema)
+        except jsonschema.exceptions.ValidationError as err:
+            print(err)
+            return False
+        return True
+
+    def find(self, workspace_name):
+        return self.workspaces.get(workspace_name, None)
+
+    @property
+    def workspaces(self):
+        return self._workspaces
+
+    @classmethod
+    def from_response(cls, response):
+        # Parse the JSON response
+        json_data = response.json()
+
+        workspaces = []
+        # Map the response to a list of Workspace instances
+        for ws in json_data.get("workspaces", {}).get("workspace", []):
+            workspaces.append(ws["name"])
+
+        # Now 'workspaces' is a list of Workspace instances
+        log.debug("Parsed Workspaces:")
+        for workspace in workspaces:
+            log.debug(f"Name: {workspace}")
+
+        return cls(workspaces)

pyproject.toml

@@ -14,6 +14,8 @@ OWSLib = "0.31.0"
 xmltodict = "0.13.0"
 types-requests = "2.32.0.20240914"
 types-xmltodict = "0.13.0.3"
+jsonschema = "^4.23.0"
+pytest-mock = "^3.14.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "8.3.3"

tests/test_acl.py

@@ -27,7 +27,7 @@ def test_create_acl_admin_rule(geoserver: GeoServerCloud) -> None:
             access="ADMIN",
             role="TEST_ROLE",
             user="TEST_USER",
-            workspace="TEST_WORKSPACE",
+            workspace_name="TEST_WORKSPACE",
         )
         assert response.status_code == 201
 
@@ -54,7 +54,7 @@ def test_create_acl_rule(geoserver: GeoServerCloud) -> None:
             priority=1,
             access="ALLOW",
             role="TEST_ROLE",
-            workspace="TEST_WORKSPACE",
+            workspace_name="TEST_WORKSPACE",
             service="WMS",
         )
         assert response.status_code == 201
@@ -99,7 +99,7 @@ def test_create_acl_rule_for_requests(geoserver: GeoServerCloud) -> None:
             priority=1,
             access="ALLOW",
             role="TEST_ROLE",
-            workspace="TEST_WORKSPACE",
+            workspace_name="TEST_WORKSPACE",
             service="WMS",
             requests=["GetCapabilities", "GetMap"],
         )

tests/test_cascaded_wmts.py

@@ -83,7 +83,7 @@ def test_create_wmts_store(
             status=201,
         )
         response = geoserver.create_wmts_store(
-            workspace=WORKSPACE,
+            workspace_name=WORKSPACE,
             name=STORE,
             capabilities=CAPABILITIES_URL,
         )
@@ -105,7 +105,7 @@ def test_update_wmts_store(
             status=200,
         )
         response = geoserver.create_wmts_store(
-            workspace=WORKSPACE,
+            workspace_name=WORKSPACE,
             name=STORE,
             capabilities=CAPABILITIES_URL,
         )
@@ -138,7 +138,7 @@ def test_create_wmts_layer(
             status=201,
         )
         response = geoserver.create_wmts_layer(
-            workspace=WORKSPACE,
+            workspace_name=WORKSPACE,
             wmts_store=STORE,
             native_layer=NATIVE_LAYER,
             published_layer=LAYER,
@@ -178,7 +178,7 @@ def test_create_wmts_layer_already_exists(
             status=201,
         )
         response = geoserver.create_wmts_layer(
-            workspace=WORKSPACE,
+            workspace_name=WORKSPACE,
             wmts_store=STORE,
             native_layer=NATIVE_LAYER,
             published_layer=LAYER,
@@ -214,7 +214,7 @@ def test_create_wmts_layer_international_title(
             status=201,
         )
         response = geoserver.create_wmts_layer(
-            workspace=WORKSPACE,
+            workspace_name=WORKSPACE,
             wmts_store=STORE,
             native_layer=NATIVE_LAYER,
             published_layer=LAYER,

tests/test_datastore.py

@@ -89,7 +89,7 @@ def test_create_pg_datastore(
         )
 
         response = geoserver.create_pg_datastore(
-            workspace=WORKSPACE,
+            workspace_name=WORKSPACE,
             datastore=STORE,
             pg_host=HOST,
             pg_port=PORT,
@@ -118,7 +118,7 @@ def test_update_pg_datastore(
         )
 
         response = geoserver.create_pg_datastore(
-            workspace=WORKSPACE,
+            workspace_name=WORKSPACE,
             datastore=STORE,
             pg_host=HOST,
             pg_port=PORT,
@@ -147,7 +147,7 @@ def test_create_jndi_datastore(
         )
 
         response = geoserver.create_jndi_datastore(
-            workspace=WORKSPACE,
+            workspace_name=WORKSPACE,
             datastore=STORE,
             jndi_reference=JNDI,
             pg_schema=SCHEMA,
@@ -173,7 +173,7 @@ def test_update_jndi_datastore(
         )
 
         response = geoserver.create_jndi_datastore(
-            workspace=WORKSPACE,
+            workspace_name=WORKSPACE,
             datastore=STORE,
             jndi_reference=JNDI,
             pg_schema=SCHEMA,

tests/test_feature_type.py

@@ -70,7 +70,7 @@ def test_create_feature_type(
             status=201,
         )
         response = geoserver.create_feature_type(
-            workspace=WORKSPACE,
+            workspace_name=WORKSPACE,
             datastore=STORE,
             layer=LAYER,
             title={"en": "English"},
@@ -95,7 +95,7 @@ def test_update_feature_type(
             status=200,
         )
         response = geoserver.create_feature_type(
-            workspace=WORKSPACE,
+            workspace_name=WORKSPACE,
             datastore=STORE,
             layer=LAYER,
             title={"en": "English"},

tests/test_layer_group.py

@@ -58,7 +58,7 @@ def test_create_layer_group(
 
         response = geoserver.create_layer_group(
             group=LAYER_GROUP,
-            workspace=WORKSPACE,
+            workspace_name=WORKSPACE,
             layers=LAYERS,
             title=TITLE,
             abstract=ABSTRACT,
@@ -83,7 +83,7 @@ def test_update_layer_group(
 
         response = geoserver.create_layer_group(
             group=LAYER_GROUP,
-            workspace=WORKSPACE,
+            workspace_name=WORKSPACE,
             layers=LAYERS,
             title=TITLE,
             abstract=ABSTRACT,

tests/test_models_workspace.py

@@ -0,0 +1,75 @@
+from unittest.mock import Mock
+
+import pytest
+
+from geoservercloud.models import Workspace
+
+
+# Test normal initialization of the Workspace class
+def test_workspace_initialization():
+    workspace = Workspace("test_workspace", isolated=True)
+
+    assert workspace.name == "test_workspace"
+    assert workspace.isolated is True
+
+
+# Test the put_payload method with isolated=True
+def test_workspace_put_payload_isolated():
+    workspace = Workspace("test_workspace", isolated=True)
+
+    expected_payload = {"workspace": {"name": "test_workspace", "isolated": True}}
+
+    assert workspace.put_payload() == expected_payload
+
+
+# Test the put_payload method with isolated=False
+def test_workspace_put_payload_not_isolated():
+    workspace = Workspace("test_workspace", isolated=False)
+
+    expected_payload = {"workspace": {"name": "test_workspace"}}
+
+    assert workspace.put_payload() == expected_payload
+
+
+# Test the post_payload method (should be the same as put_payload)
+def test_workspace_post_payload():
+    workspace = Workspace("test_workspace", isolated=True)
+
+    expected_payload = workspace.put_payload()
+
+    assert workspace.post_payload() == expected_payload
+
+
+# Test the from_response class method with isolated=True in response
+def test_workspace_from_response_isolated():
+    mock_response = Mock()
+    mock_response.json.return_value = {
+        "workspace": {"name": "test_workspace", "isolated": True}
+    }
+
+    workspace = Workspace.from_response(mock_response)
+
+    assert workspace.name == "test_workspace"
+    assert workspace.isolated is True
+
+
+# Test the from_response class method with isolated=False (not present) in response
+def test_workspace_from_response_not_isolated():
+    mock_response = Mock()
+    mock_response.json.return_value = {"workspace": {"name": "test_workspace"}}
+
+    workspace = Workspace.from_response(mock_response)
+
+    assert workspace.name == "test_workspace"
+    assert workspace.isolated is False
+
+
+# Test the from_response class method with missing workspace name
+def test_workspace_from_response_missing_name():
+    mock_response = Mock()
+    mock_response.json.return_value = {"workspace": {}}
+
+    workspace = Workspace.from_response(mock_response)
+
+    assert workspace.name is None
+    assert workspace.isolated is False

tests/test_models_workspaces.py

@@ -0,0 +1,39 @@
+from unittest.mock import Mock
+
+import jsonschema
+import pytest
+
+from geoservercloud.models import Workspaces
+
+
+# Test initialization of the Workspaces class
+def test_workspaces_initialization():
+    initial_workspaces = {"Workspace1": "http://example.com/ws1"}
+    workspaces = Workspaces(initial_workspaces)
+
+    assert workspaces.workspaces == initial_workspaces
+
+
+# Test the find method to ensure it finds existing workspaces
+def test_workspaces_find_existing():
+    initial_workspaces = {"Workspace1": "http://example.com/ws1"}
+    workspaces = Workspaces(initial_workspaces)
+
+    assert workspaces.find("Workspace1") == "http://example.com/ws1"
+
+
+# Test the find method to ensure it returns None for non-existing workspaces
+def test_workspaces_find_non_existing():
+    workspaces = Workspaces({"Workspace1": "http://example.com/ws1"})
+
+    assert workspaces.find("NonExistingWorkspace") is None
+
+
+# Test the from_response method with an empty response
+def test_workspaces_from_response_empty():
+    mock_response = Mock()
+    mock_response.json.return_value = {"workspaces": {}}
+
+    workspaces = Workspaces.from_response(mock_response)
+
+    assert len(workspaces.workspaces) == 0