diff --git a/geoservercloud/geoservercloud.py b/geoservercloud/geoservercloud.py index fe2e96b..11e99d5 100644 --- a/geoservercloud/geoservercloud.py +++ b/geoservercloud/geoservercloud.py @@ -14,6 +14,8 @@ DataStores, KeyDollarListDict, PostGisDataStore, + Style, + Styles, Workspace, Workspaces, ) @@ -169,6 +171,17 @@ def get_datastores(self, workspace_name: str) -> dict[str, Any]: response = self.get_request(self.rest_endpoints.datastores(workspace_name)) return DataStores.from_response(response).datastores + def get_postgis_datastore( + self, workspace_name: str, datastore_name: str + ) -> dict[str, Any]: + """ + Get a specific datastore + """ + response = self.get_request( + self.rest_endpoints.datastore(workspace_name, datastore_name) + ) + return PostGisDataStore.from_response(response) + def create_pg_datastore( self, workspace_name: str, @@ -444,6 +457,32 @@ def publish_gwc_layer( json=payload, ) + def get_styles(self, workspace_name: str | None = None) -> dict[str, Any]: + """ + Get all styles for a given workspace + """ + path = ( + self.rest_endpoints.styles() + if not workspace_name + else self.rest_endpoints.workspace_styles(workspace_name) + ) + styles = Styles.from_response(self.get_request(path)).styles + return styles + + def get_style( + self, style: str, workspace_name: str | None = None + ) -> dict[str, Any]: + """ + Get a specific style + """ + path = ( + self.rest_endpoints.style(style) + if not workspace_name + else self.rest_endpoints.workspace_style(workspace_name, style) + ) + return Style.from_response(self.get_request(path)) + + # TODO: add a create_style method that takes a Style object as input def create_style_from_file( self, style: str, diff --git a/geoservercloud/models/__init__.py b/geoservercloud/models/__init__.py index 59369c6..47b3230 100644 --- a/geoservercloud/models/__init__.py +++ b/geoservercloud/models/__init__.py @@ -1,6 +1,8 @@ from .common import KeyDollarListDict from .dataStore import PostGisDataStore from .dataStores import DataStores +from .style import Style +from .styles import Styles from .workspace import Workspace from .workspaces import Workspaces @@ -8,6 +10,8 @@ "DataStores", "KeyDollarListDict", "PostGisDataStore", + "Style", + "Styles", "Workspaces", "Workspace", ] diff --git a/geoservercloud/models/dataStore.py b/geoservercloud/models/dataStore.py index fe196d2..f2cd9f4 100644 --- a/geoservercloud/models/dataStore.py +++ b/geoservercloud/models/dataStore.py @@ -1,3 +1,4 @@ +import json import logging from . import KeyDollarListDict @@ -49,7 +50,8 @@ def put_payload(self): @classmethod def from_response(cls, response): - + if response.status_code == 404: + return None json_data = response.json() connection_parameters = cls.parse_connection_parameters(json_data) return cls( @@ -68,3 +70,6 @@ def parse_connection_parameters(cls, json_data): .get("connectionParameters", {}) .get("entry", []) ) + + def __repr__(self): + return json.dumps(self.put_payload(), indent=4) diff --git a/geoservercloud/models/dataStores.py b/geoservercloud/models/dataStores.py index e3c0f82..d0aa3f4 100644 --- a/geoservercloud/models/dataStores.py +++ b/geoservercloud/models/dataStores.py @@ -1,8 +1,5 @@ import logging -import jsonschema -import requests - log = logging.getLogger() diff --git a/geoservercloud/models/style.py b/geoservercloud/models/style.py new file mode 100644 index 0000000..ce3ebb6 --- /dev/null +++ b/geoservercloud/models/style.py @@ -0,0 +1,119 @@ +import json + +import xmltodict + + +class Style: + + def __init__( + self, + name: str, + workspace: str | None = None, + format: str | None = "sld", + language_version: dict | None = {"version": "1.0.0"}, + filename: str | None = None, + date_created: str | None = None, + date_modified: str | None = None, + legend_url: str | None = None, + legend_format: str | None = None, + legend_width: str | None = None, + legend_height: str | None = None, + ) -> None: + self._workspace = workspace + self._name = name + self._format = format + self._language_version = language_version + self._filename = filename + self._date_created = date_created + self._date_modified = date_modified + self.create_legend(legend_url, legend_format, legend_width, legend_height) + + # create one property for each attribute + @property + def workspace(self): + return self._workspace + + @property + def name(self): + return self._name + + @property + def format(self): + return self._format + + @property + def language_version(self): + return self._language_version + + @property + def filename(self): + return self._filename + + @property + def date_created(self): + return self._date_created + + @property + def date_modified(self): + return self._date_modified + + @property + def legend(self): + return self._legend + + def create_legend(self, url, image_format, width, height): + if any([url, image_format, width, height]): + self._legend = {} + if url: + self.legend["onlineResource"] = url + if image_format: + self.legend["format"] = image_format + if width: + self.legend["width"] = width + if height: + self.legend["height"] = height + else: + self._legend = None + + def put_payload(self): + payload = { + "style": { + "name": self.name, + "format": self.format, + "languageVersion": self.language_version, + "filename": self.filename, + } + } + if self.legend: + payload["style"]["legend"] = self.legend + return payload + + def post_payload(self): + return self.put_payload() + + @classmethod + def from_response(cls, response): + json_data = response.json() + style_data = json_data.get("style", {}) + return cls( + workspace=style_data.get("workspace"), + name=style_data.get("name"), + format=style_data.get("format"), + language_version=style_data.get("languageVersion", None), + filename=style_data.get("filename"), + date_created=style_data.get("dateCreated"), + date_modified=style_data.get("dateModified"), + legend_url=style_data.get("legend", {}).get("onlineResource"), + legend_format=style_data.get("legend", {}).get("format"), + legend_width=style_data.get("legend", {}).get("width"), + legend_height=style_data.get("legend", {}).get("height"), + ) + + def xml_post_payload(self): + return xmltodict.unparse(self.post_payload()).split("\n", 1)[1] + + def xml_put_payload(self): + return xmltodict.unparse(self.put_payload()).split("\n", 1)[1] + + def __repr__(self): + return json.dumps(self.put_payload(), indent=4) diff --git a/geoservercloud/models/styles.py b/geoservercloud/models/styles.py new file mode 100644 index 0000000..9112946 --- /dev/null +++ b/geoservercloud/models/styles.py @@ -0,0 +1,28 @@ +class Styles: + + def __init__(self, styles: list[str], workspace: str | None = None) -> None: + self._workspace = workspace + self._styles = styles + + @property + def workspace(self): + return self._workspace + + @property + def styles(self): + return self._styles + + @classmethod + def from_response(cls, response): + json_data = response.json() + styles = [] + try: + workspace = json_data["styles"]["workspace"] + except KeyError: + workspace = None + try: + for style in json_data.get("styles", {}).get("style", []): + styles.append(style["name"]) + except AttributeError: + styles = [] + return cls(styles, workspace) diff --git a/geoservercloud/models/workspace.py b/geoservercloud/models/workspace.py index df833a2..7af82bd 100644 --- a/geoservercloud/models/workspace.py +++ b/geoservercloud/models/workspace.py @@ -1,8 +1,6 @@ +import json import logging -import jsonschema -import requests - log = logging.getLogger() @@ -29,3 +27,6 @@ def from_response(cls, response): json_data.get("workspace", {}).get("isolated", False), ) return cls(json_data.get("workspace", {}).get("name", None)) + + def __repr__(self): + return json.dumps(self.put_payload(), indent=4) diff --git a/geoservercloud/models/workspaces.py b/geoservercloud/models/workspaces.py index d6c4868..05c4fbd 100644 --- a/geoservercloud/models/workspaces.py +++ b/geoservercloud/models/workspaces.py @@ -1,8 +1,5 @@ import logging -import jsonschema -import requests - log = logging.getLogger() diff --git a/tests/test_datastore.py b/tests/test_datastore.py index 4fdd95d..eef45e5 100644 --- a/tests/test_datastore.py +++ b/tests/test_datastore.py @@ -1,3 +1,4 @@ +import json from collections.abc import Generator from typing import Any @@ -6,6 +7,7 @@ from responses import matchers from geoservercloud.geoservercloud import GeoServerCloud +from geoservercloud.models import PostGisDataStore # Ensure this import is correct from tests.conftest import GEOSERVER_URL WORKSPACE = "test_workspace" @@ -105,6 +107,40 @@ def test_get_datastores( assert datastores == ["test_store"] +# Test the get_postgis_datastore method with a valid response +def test_get_postgis_datastore_valid( + 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", + json=pg_payload, + status=200, + ) + result = geoserver.get_postgis_datastore(WORKSPACE, STORE) + assert len(rsps.calls) == 1 + # FIXME: I think the geoserver rest endpoint is wrong, might be a problem with the conftest.py stuff. + # assert rsps.calls[0].request.url == geoserver.rest_endpoints.datastore(WORKSPACE, STORE) + assert json.loads(str(result)) == pg_payload + + +# Test the get_postgis_datastore method with a 404 error +def test_get_postgis_datastore_not_found(geoserver: GeoServerCloud) -> None: + datastore_name = "non_existing_datastore" + with responses.RequestsMock() as rsps: + rsps.get( + url=f"{GEOSERVER_URL}/rest/workspaces/{WORKSPACE}/datastores/{datastore_name}.json", + json={"error": "Datastore not found"}, + status=404, + ) + + not_existing_datastore = geoserver.get_postgis_datastore( + WORKSPACE, datastore_name + ) + assert len(rsps.calls) == 1 + assert not_existing_datastore is None + + def test_create_pg_datastore( geoserver: GeoServerCloud, pg_payload: dict[str, dict[str, Any]] ) -> None: diff --git a/tests/test_models_style.py b/tests/test_models_style.py new file mode 100644 index 0000000..e3ec48a --- /dev/null +++ b/tests/test_models_style.py @@ -0,0 +1,173 @@ +from unittest.mock import Mock + +import pytest + +from geoservercloud.models import Style # Adjust based on the actual import path + + +# Test initialization of Style class +def test_style_initialization(): + style = Style( + name="test_style", + workspace="test_workspace", + format="sld", + language_version={"version": "1.0.0"}, + filename="style.sld", + date_created="2023-10-01", + date_modified="2023-10-02", + legend_url="http://example.com/legend.png", + legend_format="image/png", + legend_width="100", + legend_height="100", + ) + + assert style.name == "test_style" + assert style.workspace == "test_workspace" + assert style.format == "sld" + assert style.language_version == {"version": "1.0.0"} + assert style.filename == "style.sld" + assert style.date_created == "2023-10-01" + assert style.date_modified == "2023-10-02" + assert style.legend == { + "onlineResource": "http://example.com/legend.png", + "format": "image/png", + "width": "100", + "height": "100", + } + + +# Test initialization without a legend +def test_style_initialization_without_legend(): + style = Style( + name="test_style", + workspace="test_workspace", + format="sld", + language_version={"version": "1.0.0"}, + filename="style.sld", + ) + + assert style.legend is None + + +# Test create_legend method +def test_style_create_legend(): + style = Style( + name="test_style", + workspace="test_workspace", + legend_url="http://example.com/legend.png", + legend_format="image/png", + legend_width="100", + legend_height="100", + ) + + assert style.legend == { + "onlineResource": "http://example.com/legend.png", + "format": "image/png", + "width": "100", + "height": "100", + } + + +# Test put_payload method with legend +def test_style_put_payload_with_legend(mocker): + style = Style( + name="test_style", + workspace="test_workspace", + legend_url="http://example.com/legend.png", + legend_format="image/png", + legend_width="100", + legend_height="100", + ) + + expected_payload = { + "style": { + "name": "test_style", + "format": "sld", + "languageVersion": {"version": "1.0.0"}, + "filename": None, + "legend": { + "onlineResource": "http://example.com/legend.png", + "format": "image/png", + "width": "100", + "height": "100", + }, + } + } + + payload = style.put_payload() + assert payload == expected_payload + + +# Test put_payload method without legend +def test_style_put_payload_without_legend(mocker): + style = Style( + name="test_style", + workspace="test_workspace", + ) + + expected_payload = { + "style": { + "name": "test_style", + "format": "sld", + "languageVersion": {"version": "1.0.0"}, + "filename": None, + } + } + + payload = style.put_payload() + assert payload == expected_payload + + +# Test post_payload method +def test_style_post_payload(mocker): + style = Style( + name="test_style", + workspace="test_workspace", + ) + + mock_put_payload = mocker.patch.object( + style, "put_payload", return_value={"style": {}} + ) + + payload = style.post_payload() + + assert payload == {"style": {}} + mock_put_payload.assert_called_once() + + +# Test from_response method +def test_style_from_response(): + mock_response = Mock() + mock_response.json.return_value = { + "style": { + "workspace": "test_workspace", + "name": "test_style", + "format": "sld", + "languageVersion": {"version": "1.0.0"}, + "filename": "style.sld", + "dateCreated": "2023-10-01", + "dateModified": "2023-10-02", + "legend": { + "onlineResource": "http://example.com/legend.png", + "format": "image/png", + "width": "100", + "height": "100", + }, + } + } + + style = Style.from_response(mock_response) + + assert style.name == "test_style" + assert style.workspace == "test_workspace" + assert style.format == "sld" + assert style.language_version == {"version": "1.0.0"} + assert style.filename == "style.sld" + assert style.date_created == "2023-10-01" + assert style.date_modified == "2023-10-02" + assert style.legend == { + "onlineResource": "http://example.com/legend.png", + "format": "image/png", + "width": "100", + "height": "100", + } diff --git a/tests/test_models_styles.py b/tests/test_models_styles.py new file mode 100644 index 0000000..9a8eb4a --- /dev/null +++ b/tests/test_models_styles.py @@ -0,0 +1,69 @@ +from unittest.mock import Mock + +import pytest + +from geoservercloud.models import Styles # Replace with the correct import path + + +# Test initialization of Styles class +def test_styles_initialization(): + workspace = "test_workspace" + styles = ["style1", "style2"] + + styles_instance = Styles(styles, workspace) + + assert styles_instance.workspace == workspace + assert styles_instance.styles == styles + + +# Test the from_response method with a valid response +def test_styles_from_response_valid(): + mock_response = Mock() + mock_response.json.return_value = { + "styles": { + "workspace": "test_workspace", + "style": [{"name": "style1"}, {"name": "style2"}], + } + } + + styles_instance = Styles.from_response(mock_response) + + assert styles_instance.workspace == "test_workspace" + assert styles_instance.styles == ["style1", "style2"] + + +# Test the from_response method when no workspace is provided +def test_styles_from_response_no_workspace(): + mock_response = Mock() + mock_response.json.return_value = { + "styles": {"style": [{"name": "style1"}, {"name": "style2"}]} + } + + styles_instance = Styles.from_response(mock_response) + + assert styles_instance.workspace is None + assert styles_instance.styles == ["style1", "style2"] + + +# Test the from_response method with empty styles list +def test_styles_from_response_empty_styles(): + mock_response = Mock() + mock_response.json.return_value = { + "styles": {"workspace": "test_workspace", "style": []} + } + + styles_instance = Styles.from_response(mock_response) + + assert styles_instance.workspace == "test_workspace" + assert styles_instance.styles == [] + + +# Test the from_response method with no styles section +def test_styles_from_response_no_styles_section(): + mock_response = Mock() + mock_response.json.return_value = {} + + styles_instance = Styles.from_response(mock_response) + + assert styles_instance.workspace is None + assert styles_instance.styles == [] diff --git a/tests/test_style.py b/tests/test_style.py index 20025fd..d844416 100644 --- a/tests/test_style.py +++ b/tests/test_style.py @@ -4,11 +4,79 @@ import responses from geoservercloud.geoservercloud import GeoServerCloud +from geoservercloud.models import Styles from tests.conftest import GEOSERVER_URL STYLE = "test_style" +def test_get_styles_no_workspace(geoserver: GeoServerCloud): + # Mock the self.rest_endpoints.styles() URL + with responses.RequestsMock() as rsps: + rsps.get( + url=f"{GEOSERVER_URL}/rest/styles.json", + status=200, + json={ + "styles": { + "style": [ + { + "name": "style1", + "href": f"{GEOSERVER_URL}/rest/styles/style1.json", + }, + { + "name": "style2", + "href": f"{GEOSERVER_URL}/rest/styles/style2.json", + }, + ] + } + }, + ) + result = geoserver.get_styles() + + assert result == ["style1", "style2"] + + +@responses.activate +def test_get_styles_with_workspace(geoserver: GeoServerCloud): + workspace_name = "test_workspace" + with responses.RequestsMock() as rsps: + rsps.get( + url=f"{GEOSERVER_URL}/rest/workspaces/{workspace_name}/styles.json", + status=200, + json={ + "styles": { + "style": [ + { + "name": "style3", + "href": f"{GEOSERVER_URL}/rest/workspaces/{workspace_name}/styles/style3.json", + }, + { + "name": "style4", + "href": f"{GEOSERVER_URL}/rest/workspaces/{workspace_name}/styles/style4.json", + }, + ] + } + }, + ) + result = geoserver.get_styles(workspace_name) + + assert result == ["style3", "style4"] + + +def test_get_style_no_workspace(geoserver: GeoServerCloud) -> None: + with responses.RequestsMock() as rsps: + rsps.get( + url=f"{GEOSERVER_URL}/rest/styles/{STYLE}.json", + status=200, + json={"style": {"name": STYLE}}, + ) + + style = geoserver.get_style(STYLE) + + assert style.name == STYLE # type: ignore + assert style.workspace is None # type: ignore + + def test_create_style(geoserver: GeoServerCloud) -> None: file_path = (Path(__file__).parent / "resources/style.sld").resolve() with responses.RequestsMock() as rsps: