Copy style images

geoservercloud/geoservercloudsync.py

@@ -1,3 +1,4 @@
+from geoservercloud.models.resourcedirectory import ResourceDirectory
 from geoservercloud.services import RestService
 
 
@@ -41,17 +42,47 @@ def __init__(
         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]:
+    def copy_workspace(
+        self, workspace_name: str, include_styles: bool = False
+    ) -> tuple[str, int]:
         """
-        Shallow copy a workspace from source to destination GeoServer instance
+        Copy a workspace from source to destination GeoServer instance, optionally including styles
         """
         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)
+        new_workspace, status_code = self.dst_instance.create_workspace(workspace)
+        if status_code >= 400:
+            return new_workspace, status_code
+        if include_styles:
+            content, code = self.copy_styles(workspace_name)
+            if code >= 400:
+                return content, code
+        return new_workspace, status_code
+
+    def copy_styles(
+        self, workspace_name: str | None = None, include_images: bool = True
+    ) -> tuple[str, int]:
+        """
+        Copy all styles in a workspace (if a workspace is provided) or all global styles
+        """
+        if include_images:
+            content, code = self.copy_style_images(workspace_name)
+            if code >= 400:
+                return content, code
+        styles, code = self.src_instance.get_styles(workspace_name)
+        if isinstance(styles, str):
+            return styles, code
+        for style in styles.aslist():
+            content, code = self.copy_style(style, workspace_name)
+            if code >= 400:
+                return content, code
+        return content, code
 
     def copy_style(
-        self, style_name: str, workspace_name: str | None = None
+        self,
+        style_name: str,
+        workspace_name: str | None = None,
     ) -> tuple[str, int]:
         """
         Copy a style from source to destination GeoServer instance
@@ -60,3 +91,45 @@ def copy_style(
         if isinstance(style, str):
             return style, code
         return self.dst_instance.create_style(style_name, style, workspace_name)
+
+    def copy_style_images(self, workspace_name: str | None = None) -> tuple[str, int]:
+        """
+        Copy all images in a workspace's style directory, or all global style images if no workspace is provided
+        """
+        resource_dir, code = self.src_instance.get_resource_directory(
+            path="styles", workspace_name=workspace_name
+        )
+        if isinstance(resource_dir, str):
+            return resource_dir, code
+        for child in resource_dir.children:
+            if child.is_image():
+                content, code = self.copy_resource(
+                    resource_dir="styles",
+                    resource_name=child.name,
+                    content_type=child.type,
+                    workspace_name=workspace_name,
+                )
+        return content, code
+
+    def copy_resource(
+        self,
+        resource_dir: str,
+        resource_name: str,
+        content_type: str,
+        workspace_name: str | None = None,
+    ) -> tuple[str, int]:
+        """
+        Copy a resource from source to destination GeoServer instance
+        """
+        resource, code = self.src_instance.get_resource(
+            resource_dir, resource_name, workspace_name
+        )
+        if code >= 400:
+            return resource.decode(), code
+        return self.dst_instance.put_resource(
+            path=resource_dir,
+            name=resource_name,
+            workspace_name=workspace_name,
+            content_type=content_type,
+            data=resource,
+        )

geoservercloud/models/resourcedirectory.py

@@ -0,0 +1,44 @@
+from typing import Any
+
+from geoservercloud.models.common import EntityModel
+
+
+class Resource(EntityModel):
+    def __init__(
+        self,
+        name: str,
+        href: str,
+        type: str,
+    ) -> None:
+        self.name: str = name
+        self.href: str = href
+        self.type: str = type
+
+    def is_image(self) -> bool:
+        return self.type.startswith("image")
+
+
+class ResourceDirectory(EntityModel):
+    def __init__(self, name: str, parent: Resource, children: list[Resource]) -> None:
+        self.name: str = name
+        self.parent: Resource = parent
+        self.children: list[Resource] = children
+
+    @classmethod
+    def from_get_response_payload(cls, payload: dict[str, Any]) -> "ResourceDirectory":
+        resource_directory = payload["ResourceDirectory"]
+        parent = resource_directory["parent"]
+        parent = Resource(
+            name=parent["path"],
+            href=parent["link"]["href"],
+            type=parent["link"]["type"],
+        )
+        children = [
+            Resource(child["name"], child["link"]["href"], child["link"]["type"])
+            for child in resource_directory.get("children", {}).get("child", [])
+        ]
+        return cls(
+            name=resource_directory["name"],
+            parent=parent,
+            children=children,
+        )

geoservercloud/services/restservice.py

@@ -11,6 +11,7 @@
 from geoservercloud.models.featuretype import FeatureType
 from geoservercloud.models.featuretypes import FeatureTypes
 from geoservercloud.models.layer import Layer
+from geoservercloud.models.resourcedirectory import ResourceDirectory
 from geoservercloud.models.style import Style
 from geoservercloud.models.styles import Styles
 from geoservercloud.models.workspace import Workspace
@@ -562,6 +563,39 @@ def delete_all_acl_rules(self) -> tuple[str, int]:
         response: Response = self.rest_client.delete(self.acl_endpoints.rules())
         return response.content.decode(), response.status_code
 
+    def get_resource_directory(
+        self, path: str, workspace_name: str | None = None
+    ) -> tuple[ResourceDirectory | str, int]:
+        response: Response = self.rest_client.get(
+            self.rest_endpoints.resource_directory(path, workspace_name),
+            headers={"Accept": "application/json"},
+        )
+        return self.deserialize_response(response, ResourceDirectory)
+
+    def get_resource(
+        self, path: str, resource_name: str, workspace_name: str | None = None
+    ) -> tuple[bytes, int]:
+        response: Response = self.rest_client.get(
+            self.rest_endpoints.resource(path, resource_name, workspace_name),
+        )
+        return response.content, response.status_code
+
+    def put_resource(
+        self,
+        path: str,
+        name: str,
+        content_type: str,
+        data: bytes,
+        workspace_name: str | None = None,
+    ) -> tuple[str, int]:
+        headers = {"Content-Type": content_type}
+        response: Response = self.rest_client.put(
+            path=self.rest_endpoints.resource(path, name, workspace_name),
+            headers=headers,
+            data=data,
+        )
+        return response.content.decode(), response.status_code
+
     @staticmethod
     def get_wmts_layer_bbox(
         url: str, layer_name: str
@@ -741,3 +775,22 @@ def role_user(self, role_name: str, username: str) -> str:
             return (
                 f"{self.base_url}/security/roles/role/{role_name}/user/{username}.json"
             )
+
+        def resource_directory(
+            self, relative_path: str, workspace_name: str | None = None
+        ) -> str:
+            if not workspace_name:
+                return f"{self.base_url}/resource/{relative_path}"
+            return (
+                f"{self.base_url}/resource/workspaces/{workspace_name}/{relative_path}"
+            )
+
+        def resource(
+            self,
+            relative_path: str,
+            resource_name: str,
+            workspace_name: str | None = None,
+        ) -> str:
+            if not workspace_name:
+                return f"{self.base_url}/resource/{relative_path}/{resource_name}"
+            return f"{self.base_url}/resource/workspaces/{workspace_name}/{relative_path}/{resource_name}"

tests/models/test_resourcedirectory.py

@@ -0,0 +1,46 @@
+from geoservercloud.models.resourcedirectory import Resource, ResourceDirectory
+
+
+def test_from_get_response_payload():
+    payload = {
+        "ResourceDirectory": {
+            "name": "test_resource_directory",
+            "parent": {
+                "path": "workspace/parent",
+                "name": "parent",
+                "link": {"href": "http://example.com", "type": "application/json"},
+            },
+            "children": {
+                "child": [
+                    {
+                        "name": "child1.svg",
+                        "link": {
+                            "href": "http://example.com/child1",
+                            "type": "image/svg+xml",
+                        },
+                    },
+                    {
+                        "name": "child2.xml",
+                        "link": {
+                            "href": "http://example.com/child2",
+                            "type": "application/xml",
+                        },
+                    },
+                ]
+            },
+        }
+    }
+
+    resource_directory = ResourceDirectory.from_get_response_payload(payload)
+
+    assert resource_directory.name == "test_resource_directory"
+    assert resource_directory.parent.name == "workspace/parent"
+    assert resource_directory.parent.href == "http://example.com"
+    assert resource_directory.parent.type == "application/json"
+    assert len(resource_directory.children) == 2
+    assert resource_directory.children[0].name == "child1.svg"
+    assert resource_directory.children[0].href == "http://example.com/child1"
+    assert resource_directory.children[0].type == "image/svg+xml"
+    assert resource_directory.children[1].name == "child2.xml"
+    assert resource_directory.children[1].href == "http://example.com/child2"
+    assert resource_directory.children[1].type == "application/xml"

tests/services/test_resource.py

@@ -0,0 +1,74 @@
+import pytest
+import responses
+
+from geoservercloud.models.resourcedirectory import ResourceDirectory
+from geoservercloud.services.restservice import RestService
+
+
+@pytest.fixture
+def rest_service():
+    yield RestService("http://geoserver", auth=("test", "test"))
+
+
+@pytest.fixture
+def resource_dir_get_response():
+    yield {
+        "ResourceDirectory": {
+            "name": "test_resource_directory",
+            "parent": {
+                "path": "workspace/parent",
+                "name": "parent",
+                "link": {"href": "http://example.com", "type": "application/json"},
+            },
+            "children": {
+                "child": [
+                    {
+                        "name": "child1.svg",
+                        "link": {
+                            "href": "http://example.com/child1",
+                            "type": "image/svg+xml",
+                        },
+                    },
+                    {
+                        "name": "child2.xml",
+                        "link": {
+                            "href": "http://example.com/child2",
+                            "type": "application/xml",
+                        },
+                    },
+                ]
+            },
+        }
+    }
+
+
+def test_get_workspace_style_resource_directory(
+    rest_service: RestService, resource_dir_get_response: dict
+):
+    with responses.RequestsMock() as rsps:
+        rsps.get(
+            url=f"{rest_service.url}/rest/resource/workspaces/test/styles",
+            status=200,
+            json=resource_dir_get_response,
+            match=[responses.matchers.header_matcher({"Accept": "application/json"})],
+        )
+        resource_dir, code = rest_service.get_resource_directory(
+            path="styles", workspace_name="test"
+        )
+        assert isinstance(resource_dir, ResourceDirectory)
+        assert len(resource_dir.children) == 2
+
+
+def test_get_global_style_resource_directory(
+    rest_service: RestService, resource_dir_get_response: dict
+):
+    with responses.RequestsMock() as rsps:
+        rsps.get(
+            url=f"{rest_service.url}/rest/resource/styles",
+            status=200,
+            json=resource_dir_get_response,
+            match=[responses.matchers.header_matcher({"Accept": "application/json"})],
+        )
+        resource_dir, code = rest_service.get_resource_directory(path="styles")
+        assert isinstance(resource_dir, ResourceDirectory)
+        assert len(resource_dir.children) == 2

tests/services/test_rest_endpoints.py

@@ -80,3 +80,21 @@ def test_workspace_style_endpoint(
         endpoints.style(style_name="test_style", workspace_name="test", format=format)
         == path
     )
+
+
+def test_resource_directory_endpoint(endpoints: RestService.RestEndpoints):
+    assert (
+        endpoints.resource_directory("styles", "test_workspace")
+        == "/rest/resource/workspaces/test_workspace/styles"
+    )
+    assert endpoints.resource_directory("styles") == "/rest/resource/styles"
+
+
+def test_resource_endpoint(endpoints: RestService.RestEndpoints):
+    assert (
+        endpoints.resource("styles", "image.svg", "test_workspace")
+        == "/rest/resource/workspaces/test_workspace/styles/image.svg"
+    )
+    assert (
+        endpoints.resource("styles", "image.svg") == "/rest/resource/styles/image.svg"
+    )