From fc0f984a15381561341ed5f88ce65028f6c99439 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?C=C3=A9cile=20Vuilleumier?= Date: Mon, 11 Nov 2024 10:27:08 +0100 Subject: [PATCH] Copy style images --- geoservercloud/geoservercloudsync.py | 81 ++++++++++++++++++++-- geoservercloud/models/resourcedirectory.py | 44 ++++++++++++ geoservercloud/services/restservice.py | 53 ++++++++++++++ tests/models/test_resourcedirectory.py | 46 ++++++++++++ tests/services/test_resource.py | 74 ++++++++++++++++++++ tests/services/test_rest_endpoints.py | 18 +++++ 6 files changed, 312 insertions(+), 4 deletions(-) create mode 100644 geoservercloud/models/resourcedirectory.py create mode 100644 tests/models/test_resourcedirectory.py create mode 100644 tests/services/test_resource.py diff --git a/geoservercloud/geoservercloudsync.py b/geoservercloud/geoservercloudsync.py index df64e52..ff49430 100644 --- a/geoservercloud/geoservercloudsync.py +++ b/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, + ) diff --git a/geoservercloud/models/resourcedirectory.py b/geoservercloud/models/resourcedirectory.py new file mode 100644 index 0000000..83aefb4 --- /dev/null +++ b/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, + ) diff --git a/geoservercloud/services/restservice.py b/geoservercloud/services/restservice.py index 5c2206b..ec4de21 100644 --- a/geoservercloud/services/restservice.py +++ b/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}" diff --git a/tests/models/test_resourcedirectory.py b/tests/models/test_resourcedirectory.py new file mode 100644 index 0000000..aeaacaf --- /dev/null +++ b/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" diff --git a/tests/services/test_resource.py b/tests/services/test_resource.py new file mode 100644 index 0000000..10ef7a0 --- /dev/null +++ b/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 diff --git a/tests/services/test_rest_endpoints.py b/tests/services/test_rest_endpoints.py index 8074c61..6fb0c3e 100644 --- a/tests/services/test_rest_endpoints.py +++ b/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" + )