Add facade class to copy GeoServer resources

geoservercloud/__init__.py

@@ -1,3 +1,4 @@
 from .geoservercloud import GeoServerCloud
+from .geoservercloudsync import GeoServerCloudSync
 
-__all__: list[str] = ["GeoServerCloud"]
+__all__: list[str] = ["GeoServerCloud", "GeoServerCloudSync"]

geoservercloud/geoservercloudsync.py

@@ -0,0 +1,51 @@
+from geoservercloud.services import RestService
+
+
+class GeoServerCloudSync:
+    """
+    Facade class allowing synchronization of GeoServer resources between two GeoServer instances
+
+    Attributes
+    ----------
+    src_url : str
+        base GeoServer URL for source GeoServer instance
+    src_user : str
+        GeoServer username for source GeoServer instance
+    src_password : str
+        GeoServer password for source GeoServer instance
+    dst_url : str
+        base GeoServer URL for destination GeoServer instance
+    dst_user : str
+        GeoServer username for destination GeoServer instance
+    dst_password : str
+        GeoServer password for destination GeoServer instance
+    """
+
+    def __init__(
+        self,
+        src_url: str,
+        src_user: str,
+        src_password: str,
+        dst_url: str,
+        dst_user: str,
+        dst_password: str,
+    ) -> None:
+        self.src_url: str = src_url.strip("/")
+        self.src_user: str = src_user
+        self.src_password: str = src_password
+        self.src_auth: tuple[str, str] = (src_user, src_password)
+        self.src_instance: RestService = RestService(src_url, self.src_auth)
+        self.dst_url: str = dst_url.strip("/")
+        self.dst_user: str = dst_user
+        self.dst_password: str = dst_password
+        self.dst_auth: tuple[str, str] = (dst_user, dst_password)
+        self.dst_instance: RestService = RestService(dst_url, self.dst_auth)
+
+    def copy_workspace(self, workspace_name: str) -> tuple[str, int]:
+        """
+        Shallow copy a workspace from source to destination GeoServer instance
+        """
+        workspace, status_code = self.src_instance.get_workspace(workspace_name)
+        if isinstance(workspace, str):
+            return workspace, status_code
+        return self.dst_instance.create_workspace(workspace)

geoservercloud/services/restservice.py

@@ -582,8 +582,9 @@ def resource_exists(self, path: str) -> bool:
         response: Response = self.rest_client.get(path)
         return response.status_code == 200
 
+    @staticmethod
     def deserialize_response(
-        self, response: Response, data_type: type[BaseModel]
+        response: Response, data_type: type[BaseModel]
     ) -> tuple[Any, int]:
         try:
             content = response.json()

tests/test_sync.py

@@ -0,0 +1,41 @@
+import pytest
+import responses
+from responses import matchers
+
+from geoservercloud import GeoServerCloudSync
+
+GEOSERVER_SRC_URL = "http://source-geoserver"
+GEOSERVER_DST_URL = "http://destination-geoserver"
+
+
+@pytest.fixture
+def geoserver_sync():
+    return GeoServerCloudSync(
+        GEOSERVER_SRC_URL,
+        "admin",
+        "geoserver",
+        GEOSERVER_DST_URL,
+        "admin",
+        "geoserver",
+    )
+
+
+def test_copy_workspace(geoserver_sync):
+
+    workspace_name = "test_workspace"
+    workspace = {"name": workspace_name, "isolated": True}
+
+    with responses.RequestsMock() as rsps:
+        rsps.get(
+            url=f"{GEOSERVER_SRC_URL}/rest/workspaces/{workspace_name}.json",
+            status=200,
+            json={"workspace": workspace},
+        )
+        rsps.post(
+            url=f"{GEOSERVER_DST_URL}/rest/workspaces.json",
+            status=201,
+            body=b"test_workspace",
+            match=[matchers.json_params_matcher({"workspace": workspace})],
+        )
+
+        assert geoserver_sync.copy_workspace(workspace_name) == (workspace_name, 201)