Test workspaces and datastores

Makefile

@@ -8,6 +8,7 @@ help: ## Display this help message
 
 .PHONY: install
 install: ## Install package
+	poetry lock --no-update
 	poetry install
 
 .PHONY: build

geoservercloud/geoservercloud.py

@@ -102,12 +102,14 @@ def delete_workspace(self, workspace: str) -> Response:
 
     def recreate_workspace(
         self, workspace: str, set_default_workspace: bool = False
-    ) -> None:
+    ) -> Response:
         """
         Create a workspace in GeoServer, and first delete it if it already exists.
         """
         self.delete_workspace(workspace)
-        self.create_workspace(workspace, set_default_workspace=set_default_workspace)
+        return self.create_workspace(
+            workspace, set_default_workspace=set_default_workspace
+        )
 
     def publish_workspace(self, workspace) -> Response:
         path: str = f"{self.workspace_wms_settings_path(workspace)}"
@@ -333,7 +335,7 @@ def publish_gwc_layer(
         self.post_request(
             "/gwc/rest/reload",
             headers={"Content-Type": "application/json"},
-            data="reload_configuration=1",
+            data="reload_configuration=1",  # type: ignore
         )
         payload = Templates.gwc_layer(workspace, layer, f"EPSG:{epsg}")
         return self.put_request(

pyproject.toml

@@ -15,6 +15,7 @@ xmltodict = "0.13.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "8.3.2"
+responses = "0.25.3"
 
 [build-system]
 requires = [

tests/conftest.py

@@ -0,0 +1,11 @@
+import pytest
+
+from geoservercloud import GeoServerCloud
+
+GEOSERVER_URL = "http://localhost:8080/geoserver"
+
+
+@pytest.fixture(scope="session")
+def geoserver():
+    geoserver = GeoServerCloud(url=GEOSERVER_URL)
+    yield geoserver

tests/test_datastore.py

@@ -0,0 +1,184 @@
+from collections.abc import Generator
+from typing import Any
+
+import pytest
+import responses
+from responses import matchers
+
+from geoservercloud.geoservercloud import GeoServerCloud
+from tests.conftest import GEOSERVER_URL
+
+WORKSPACE = "test_workspace"
+STORE = "test_store"
+HOST = "localhost"
+PORT = 5432
+DATABASE = "test_db"
+USER = "test_user"
+PASSWORD = "test_password"
+SCHEMA = "test_schema"
+JNDI = "java:comp/env/jdbc/data"
+DESCRIPTION = "test description"
+
+
+@pytest.fixture(scope="module")
+def pg_payload() -> Generator[dict[str, dict[str, Any]], Any, None]:
+    yield {
+        "dataStore": {
+            "name": STORE,
+            "connectionParameters": {
+                "entry": [
+                    {"@key": "dbtype", "$": "postgis"},
+                    {"@key": "host", "$": HOST},
+                    {"@key": "port", "$": PORT},
+                    {"@key": "database", "$": DATABASE},
+                    {"@key": "user", "$": USER},
+                    {"@key": "passwd", "$": PASSWORD},
+                    {"@key": "schema", "$": SCHEMA},
+                    {
+                        "@key": "namespace",
+                        "$": f"http://{WORKSPACE}",
+                    },
+                    {"@key": "Expose primary keys", "$": "true"},
+                ]
+            },
+        }
+    }
+
+
+@pytest.fixture(scope="module")
+def jndi_payload() -> Generator[dict[str, dict[str, Any]], Any, None]:
+    yield {
+        "dataStore": {
+            "name": STORE,
+            "description": DESCRIPTION,
+            "connectionParameters": {
+                "entry": [
+                    {"@key": "dbtype", "$": "postgis"},
+                    {
+                        "@key": "jndiReferenceName",
+                        "$": JNDI,
+                    },
+                    {
+                        "@key": "schema",
+                        "$": SCHEMA,
+                    },
+                    {
+                        "@key": "namespace",
+                        "$": f"http://{WORKSPACE}",
+                    },
+                    {"@key": "Expose primary keys", "$": "true"},
+                ]
+            },
+        }
+    }
+
+
+def test_create_pg_datastore(
+    geoserver: GeoServerCloud, pg_payload: dict[str, dict[str, Any]]
+) -> None:
+    with responses.RequestsMock() as rsps:
+        rsps.get(
+            url=f"{GEOSERVER_URL}/rest/workspaces/{WORKSPACE}/datastores/{STORE}.json",
+            status=404,
+        )
+        rsps.post(
+            url=f"{GEOSERVER_URL}/rest/workspaces/{WORKSPACE}/datastores.json",
+            status=201,
+            json={"workspace": {"name": WORKSPACE}},
+            match=[matchers.json_params_matcher(pg_payload)],
+        )
+
+        response = geoserver.create_pg_datastore(
+            workspace=WORKSPACE,
+            datastore=STORE,
+            pg_host=HOST,
+            pg_port=PORT,
+            pg_db=DATABASE,
+            pg_user=USER,
+            pg_password=PASSWORD,
+            pg_schema=SCHEMA,
+        )
+
+        assert response
+        assert response.status_code == 201
+
+
+def test_update_pg_datastore(
+    geoserver: GeoServerCloud, pg_payload: dict[str, dict[str, Any]]
+) -> None:
+    with responses.RequestsMock() as rsps:
+        rsps.get(
+            url=f"{GEOSERVER_URL}/rest/workspaces/{WORKSPACE}/datastores/{STORE}.json",
+            status=200,
+        )
+        rsps.put(
+            url=f"{GEOSERVER_URL}/rest/workspaces/{WORKSPACE}/datastores/{STORE}.json",
+            status=200,
+            match=[matchers.json_params_matcher(pg_payload)],
+        )
+
+        response = geoserver.create_pg_datastore(
+            workspace=WORKSPACE,
+            datastore=STORE,
+            pg_host=HOST,
+            pg_port=PORT,
+            pg_db=DATABASE,
+            pg_user=USER,
+            pg_password=PASSWORD,
+            pg_schema=SCHEMA,
+        )
+
+        assert response
+        assert response.status_code == 200
+
+
+def test_create_jndi_datastore(
+    geoserver: GeoServerCloud, jndi_payload: dict[str, dict[str, Any]]
+) -> None:
+    with responses.RequestsMock() as rsps:
+        rsps.get(
+            url=f"{GEOSERVER_URL}/rest/workspaces/{WORKSPACE}/datastores/{STORE}.json",
+            status=404,
+        )
+        rsps.post(
+            url=f"{GEOSERVER_URL}/rest/workspaces/{WORKSPACE}/datastores.json",
+            status=201,
+            match=[matchers.json_params_matcher(jndi_payload)],
+        )
+
+        response = geoserver.create_jndi_datastore(
+            workspace=WORKSPACE,
+            datastore=STORE,
+            jndi_reference=JNDI,
+            pg_schema=SCHEMA,
+            description=DESCRIPTION,
+        )
+
+        assert response
+        assert response.status_code == 201
+
+
+def test_update_jndi_datastore(
+    geoserver: GeoServerCloud, jndi_payload: dict[str, dict[str, Any]]
+) -> None:
+    with responses.RequestsMock() as rsps:
+        rsps.get(
+            url=f"{GEOSERVER_URL}/rest/workspaces/{WORKSPACE}/datastores/{STORE}.json",
+            status=200,
+        )
+        rsps.put(
+            url=f"{GEOSERVER_URL}/rest/workspaces/{WORKSPACE}/datastores/{STORE}.json",
+            status=200,
+            match=[matchers.json_params_matcher(jndi_payload)],
+        )
+
+        response = geoserver.create_jndi_datastore(
+            workspace=WORKSPACE,
+            datastore=STORE,
+            jndi_reference=JNDI,
+            pg_schema=SCHEMA,
+            description=DESCRIPTION,
+        )
+
+        assert response
+        assert response.status_code == 200

tests/test_workspace.py

@@ -0,0 +1,80 @@
+import responses
+from responses import matchers
+
+from geoservercloud.geoservercloud import GeoServerCloud
+from tests.conftest import GEOSERVER_URL
+
+
+def test_create_workspace(geoserver: GeoServerCloud) -> None:
+    workspace = "test_workspace"
+    isolated = True
+
+    with responses.RequestsMock() as rsps:
+        rsps.post(
+            url=f"{GEOSERVER_URL}/rest/workspaces.json",
+            status=201,
+            match=[
+                matchers.json_params_matcher(
+                    {
+                        "workspace": {
+                            "name": workspace,
+                            "isolated": isolated,
+                        }
+                    }
+                )
+            ],
+        )
+
+        response = geoserver.create_workspace(workspace, isolated=isolated)
+
+        assert response.status_code == 201
+
+
+def test_update_workspace(geoserver: GeoServerCloud) -> None:
+    workspace = "test_workspace"
+
+    with responses.RequestsMock() as rsps:
+        rsps.post(
+            url=f"{GEOSERVER_URL}/rest/workspaces.json",
+            status=409,
+        )
+        rsps.put(
+            url=f"{GEOSERVER_URL}/rest/workspaces/{workspace}.json",
+            status=200,
+        )
+
+        response = geoserver.create_workspace(workspace)
+
+        assert response.status_code == 200
+
+
+def test_delete_workspace(geoserver: GeoServerCloud) -> None:
+    workspace = "test_workspace"
+
+    with responses.RequestsMock() as rsps:
+        rsps.delete(
+            url=f"{GEOSERVER_URL}/rest/workspaces/{workspace}.json",
+            status=200,
+        )
+
+        response = geoserver.delete_workspace(workspace)
+
+        assert response.status_code == 200
+
+
+def test_recreate_workspace(geoserver: GeoServerCloud) -> None:
+    workspace = "test_workspace"
+
+    with responses.RequestsMock() as rsps:
+        rsps.delete(
+            url=f"{GEOSERVER_URL}/rest/workspaces/{workspace}.json",
+            status=200,
+        )
+        rsps.post(
+            url=f"{GEOSERVER_URL}/rest/workspaces.json",
+            status=201,
+        )
+
+        response = geoserver.recreate_workspace(workspace)
+
+        assert response.status_code == 201