diff --git a/geoservercloud/geoservercloud.py b/geoservercloud/geoservercloud.py index d092c68..0ddca8b 100644 --- a/geoservercloud/geoservercloud.py +++ b/geoservercloud/geoservercloud.py @@ -11,6 +11,7 @@ from geoservercloud.models.featuretype import FeatureType from geoservercloud.models.layer import Layer from geoservercloud.models.style import Style +from geoservercloud.models.wmssettings import WmsSettings from geoservercloud.models.workspace import Workspace from geoservercloud.services import OwsService, RestService from geoservercloud.templates import Templates @@ -120,11 +121,90 @@ def recreate_workspace( set_default_workspace=set_default_workspace, ) - def publish_workspace(self, workspace_name) -> tuple[str, int]: + def get_workspace_wms_settings( + self, workspace_name: str + ) -> tuple[dict[str, Any] | str, int]: + """ + Get the WMS settings for a given workspace + """ + wms_settings, status_code = self.rest_service.get_workspace_wms_settings( + workspace_name + ) + if isinstance(wms_settings, str): + return wms_settings, status_code + return wms_settings.asdict(), status_code + + def publish_workspace( + self, + workspace_name: str, + versions: list[str] = ["1.1.1", "1.3.0"], + cite_compliant: bool = False, + schema_base_url: str = "http://schemas.opengis.net", + verbose: bool = False, + bbox_for_each_crs: bool = False, + watermark: dict = { + "enabled": False, + "position": "BOT_RIGHT", + "transparency": 100, + }, + interpolation: str = "Nearest", + get_feature_info_mime_type_checking_enabled: bool = False, + get_map_mime_type_checking_enabled: bool = False, + dynamic_styling_disabled: bool = False, + features_reprojection_disabled: bool = False, + max_buffer: int = 0, + max_request_memory: int = 0, + max_rendering_time: int = 0, + max_rendering_errors: int = 0, + max_requested_dimension_values: int = 100, + cache_configuration: dict = { + "enabled": False, + "maxEntries": 1000, + "maxEntrySize": 51200, + }, + remote_style_max_request_time: int = 60000, + remote_style_timeout: int = 30000, + default_group_style_enabled: bool = True, + transform_feature_info_disabled: bool = False, + auto_escape_template_values: bool = False, + ) -> tuple[str, int]: """ Publish the WMS service for a given workspace """ - return self.rest_service.publish_workspace(Workspace(workspace_name)) + wms_settings = WmsSettings( + workspace_name=workspace_name, + name="WMS", + enabled=True, + versions={ + "org.geotools.util.Version": [ + {"version": version} for version in versions + ] + }, + cite_compliant=cite_compliant, + schema_base_url=schema_base_url, + verbose=verbose, + bbox_for_each_crs=bbox_for_each_crs, + watermark=watermark, + interpolation=interpolation, + get_feature_info_mime_type_checking_enabled=get_feature_info_mime_type_checking_enabled, + get_map_mime_type_checking_enabled=get_map_mime_type_checking_enabled, + dynamic_styling_disabled=dynamic_styling_disabled, + features_reprojection_disabled=features_reprojection_disabled, + max_buffer=max_buffer, + max_request_memory=max_request_memory, + max_rendering_time=max_rendering_time, + max_rendering_errors=max_rendering_errors, + max_requested_dimension_values=max_requested_dimension_values, + cache_configuration=cache_configuration, + remote_style_max_request_time=remote_style_max_request_time, + remote_style_timeout=remote_style_timeout, + default_group_style_enabled=default_group_style_enabled, + transform_feature_info_disabled=transform_feature_info_disabled, + auto_escape_template_values=auto_escape_template_values, + ) + return self.rest_service.put_workspace_wms_settings( + workspace_name, wms_settings + ) def set_default_locale_for_service( self, workspace_name: str, locale: str | None @@ -132,8 +212,9 @@ def set_default_locale_for_service( """ Set a default language for localized WMS requests """ - return self.rest_service.set_default_locale_for_service( - Workspace(workspace_name), locale + wms_settings = WmsSettings(default_locale=locale) + return self.rest_service.put_workspace_wms_settings( + workspace_name, wms_settings ) def unset_default_locale_for_service(self, workspace_name) -> tuple[str, int]: diff --git a/geoservercloud/models/wmssettings.py b/geoservercloud/models/wmssettings.py new file mode 100644 index 0000000..2ac5a13 --- /dev/null +++ b/geoservercloud/models/wmssettings.py @@ -0,0 +1,159 @@ +from typing import Any + +from geoservercloud.models.common import EntityModel, ReferencedObjectModel + + +class WmsSettings(EntityModel): + def __init__( + self, + workspace_name: str | None = None, + name: str | None = None, + enabled: bool | None = None, + default_locale: str | None = None, + auto_escape_template_values: bool | None = None, + bbox_for_each_crs: bool | None = None, + cache_configuration: dict[str, bool | int] | None = None, + cite_compliant: bool | None = None, + default_group_style_enabled: bool | None = None, + dynamic_styling_disabled: bool | None = None, + features_reprojection_disabled: bool | None = None, + get_feature_info_mime_type_checking_enabled: bool | None = None, + get_map_mime_type_checking_enabled: bool | None = None, + interpolation: str | None = None, + max_buffer: int | None = None, + max_rendering_errors: int | None = None, + max_rendering_time: int | None = None, + max_request_memory: int | None = None, + max_requested_dimension_values: int | None = None, + remote_style_max_request_time: int | None = None, + remote_style_timeout: int | None = None, + schema_base_url: str | None = None, + transform_feature_info_disabled: bool | None = None, + verbose: bool | None = None, + versions: dict[str, list[dict[str, str]]] | None = None, + watermark: dict[str, bool | int | str] | None = None, + ) -> None: + self.workspace: ReferencedObjectModel | None = ( + ReferencedObjectModel(workspace_name) if workspace_name else None + ) + self.name: str | None = name + self.enabled: bool | None = enabled + self.default_locale: str | None = default_locale + self.auto_escape_template_values: bool | None = auto_escape_template_values + self.bbox_for_each_crs: bool | None = bbox_for_each_crs + self.cache_configuration: dict[str, bool | int] | None = cache_configuration + self.cite_compliant: bool | None = cite_compliant + self.default_group_style_enabled: bool | None = default_group_style_enabled + self.dynamic_styling_disabled: bool | None = dynamic_styling_disabled + self.features_reprojection_disabled: bool | None = ( + features_reprojection_disabled + ) + self.get_feature_info_mime_type_checking_enabled: bool | None = ( + get_feature_info_mime_type_checking_enabled + ) + self.get_map_mime_type_checking_enabled: bool | None = ( + get_map_mime_type_checking_enabled + ) + self.interpolation: str | None = interpolation + self.max_buffer: int | None = max_buffer + self.max_rendering_errors: int | None = max_rendering_errors + self.max_rendering_time: int | None = max_rendering_time + self.max_request_memory: int | None = max_request_memory + self.max_requested_dimension_values: int | None = max_requested_dimension_values + self.remote_style_max_request_time: int | None = remote_style_max_request_time + self.remote_style_timeout: int | None = remote_style_timeout + self.schema_base_url: str | None = schema_base_url + self.transform_feature_info_disabled: bool | None = ( + transform_feature_info_disabled + ) + self.verbose: bool | None = verbose + self.versions: dict[str, list[dict[str, str]]] | None = versions + self.watermark: dict[str, bool | int | str] | None = watermark + + @property + def workspace_name(self) -> str | None: + return self.workspace.name if self.workspace else None + + def asdict(self) -> dict[str, str]: + return EntityModel.add_items_to_dict( + {}, + { + "workspace": self.workspace_name, + "name": self.name, + "enabled": self.enabled, + "defaultLocale": self.default_locale, + "autoEscapeTemplateValues": self.auto_escape_template_values, + "bboxForEachCRS": self.bbox_for_each_crs, + "cacheConfiguration": self.cache_configuration, + "citeCompliant": self.cite_compliant, + "defaultGroupStyleEnabled": self.default_group_style_enabled, + "dynamicStylingDisabled": self.dynamic_styling_disabled, + "featuresReprojectionDisabled": self.features_reprojection_disabled, + "getFeatureInfoMimeTypeCheckingEnabled": self.get_feature_info_mime_type_checking_enabled, + "getMapMimeTypeCheckingEnabled": self.get_map_mime_type_checking_enabled, + "interpolation": self.interpolation, + "maxBuffer": self.max_buffer, + "maxRenderingErrors": self.max_rendering_errors, + "maxRenderingTime": self.max_rendering_time, + "maxRequestMemory": self.max_request_memory, + "maxRequestedDimensionValues": self.max_requested_dimension_values, + "remoteStyleMaxRequestTime": self.remote_style_max_request_time, + "remoteStyleTimeout": self.remote_style_timeout, + "schemaBaseURL": self.schema_base_url, + "transformFeatureInfoDisabled": self.transform_feature_info_disabled, + "verbose": self.verbose, + "versions": self.versions, + "watermark": self.watermark, + }, + ) + + def post_payload(self) -> dict[str, Any]: + content: dict[str, Any] = self.asdict() + if self.workspace: + content["workspace"] = self.workspace.asdict() + return {"wms": content} + + def put_payload(self) -> dict[str, Any]: + return self.post_payload() + + @classmethod + def from_get_response_payload(cls, content: dict[str, Any]): + wms_settings = content["wms"] + return cls( + workspace_name=wms_settings.get("workspace", {}).get("name"), + name=wms_settings.get("name"), + enabled=wms_settings.get("enabled"), + default_locale=wms_settings.get("defaultLocale"), + auto_escape_template_values=wms_settings.get("autoEscapeTemplateValues"), + bbox_for_each_crs=wms_settings.get("bboxForEachCRS"), + cache_configuration=wms_settings.get("cacheConfiguration"), + cite_compliant=wms_settings.get("citeCompliant"), + default_group_style_enabled=wms_settings.get("defaultGroupStyleEnabled"), + dynamic_styling_disabled=wms_settings.get("dynamicStylingDisabled"), + features_reprojection_disabled=wms_settings.get( + "featuresReprojectionDisabled" + ), + get_feature_info_mime_type_checking_enabled=wms_settings.get( + "getFeatureInfoMimeTypeCheckingEnabled" + ), + get_map_mime_type_checking_enabled=wms_settings.get( + "getMapMimeTypeCheckingEnabled" + ), + interpolation=wms_settings.get("interpolation"), + max_buffer=wms_settings.get("maxBuffer"), + max_rendering_errors=wms_settings.get("maxRenderingErrors"), + max_rendering_time=wms_settings.get("maxRenderingTime"), + max_request_memory=wms_settings.get("maxRequestMemory"), + max_requested_dimension_values=wms_settings.get( + "maxRequestedDimensionValues" + ), + remote_style_max_request_time=wms_settings.get("remoteStyleMaxRequestTime"), + remote_style_timeout=wms_settings.get("remoteStyleTimeout"), + schema_base_url=wms_settings.get("schemaBaseURL"), + transform_feature_info_disabled=wms_settings.get( + "transformFeatureInfoDisabled" + ), + verbose=wms_settings.get("verbose"), + versions=wms_settings.get("versions"), + watermark=wms_settings.get("watermark"), + ) diff --git a/geoservercloud/services/restservice.py b/geoservercloud/services/restservice.py index 7f69d83..9c5fd37 100644 --- a/geoservercloud/services/restservice.py +++ b/geoservercloud/services/restservice.py @@ -14,6 +14,7 @@ from geoservercloud.models.resourcedirectory import ResourceDirectory from geoservercloud.models.style import Style from geoservercloud.models.styles import Styles +from geoservercloud.models.wmssettings import WmsSettings from geoservercloud.models.workspace import Workspace from geoservercloud.models.workspaces import Workspaces from geoservercloud.services.restclient import RestClient @@ -62,23 +63,20 @@ def delete_workspace(self, workspace: Workspace) -> tuple[str, int]: response: Response = self.rest_client.delete(path, params=params) return response.content.decode(), response.status_code - def publish_workspace(self, workspace: Workspace) -> tuple[str, int]: - data: dict[str, dict[str, Any]] = Templates.workspace_wms(workspace.name) - response: Response = self.rest_client.put( - self.rest_endpoints.workspace_wms_settings(workspace.name), json=data + def get_workspace_wms_settings( + self, workspace_name: str + ) -> tuple[WmsSettings | str, int]: + response: Response = self.rest_client.get( + self.rest_endpoints.workspace_wms_settings(workspace_name) ) - return response.content.decode(), response.status_code + return self.deserialize_response(response, WmsSettings) - def set_default_locale_for_service( - self, workspace: Workspace, locale: str | None + def put_workspace_wms_settings( + self, workspace_name: str, wms_settings: WmsSettings ) -> tuple[str, int]: - data: dict[str, dict[str, Any]] = { - "wms": { - "defaultLocale": locale, - } - } response: Response = self.rest_client.put( - self.rest_endpoints.workspace_wms_settings(workspace.name), json=data + self.rest_endpoints.workspace_wms_settings(workspace_name), + json=wms_settings.put_payload(), ) return response.content.decode(), response.status_code diff --git a/geoservercloud/templates.py b/geoservercloud/templates.py index 481ced6..98a4ae3 100644 --- a/geoservercloud/templates.py +++ b/geoservercloud/templates.py @@ -4,51 +4,6 @@ class Templates: - @staticmethod - def workspace_wms(workspace: str) -> dict[str, dict[str, Any]]: - return { - "wms": { - "workspace": {"name": workspace}, - "enabled": True, - "name": "WMS", - "versions": { - "org.geotools.util.Version": [ - {"version": "1.1.1"}, - {"version": "1.3.0"}, - ] - }, - "citeCompliant": False, - "schemaBaseURL": "http://schemas.opengis.net", - "verbose": False, - "bboxForEachCRS": False, - "watermark": { - "enabled": False, - "position": "BOT_RIGHT", - "transparency": 100, - }, - "interpolation": "Nearest", - "getFeatureInfoMimeTypeCheckingEnabled": False, - "getMapMimeTypeCheckingEnabled": False, - "dynamicStylingDisabled": False, - "featuresReprojectionDisabled": False, - "maxBuffer": 0, - "maxRequestMemory": 0, - "maxRenderingTime": 0, - "maxRenderingErrors": 0, - "maxRequestedDimensionValues": 100, - "cacheConfiguration": { - "enabled": False, - "maxEntries": 1000, - "maxEntrySize": 51200, - }, - "remoteStyleMaxRequestTime": 60000, - "remoteStyleTimeout": 30000, - "defaultGroupStyleEnabled": True, - "transformFeatureInfoDisabled": False, - "autoEscapeTemplateValues": False, - } - } - @staticmethod def wmts_store( workspace: str, name: str, capabilities: str diff --git a/tests/models/test_wmssettings.py b/tests/models/test_wmssettings.py new file mode 100644 index 0000000..3666257 --- /dev/null +++ b/tests/models/test_wmssettings.py @@ -0,0 +1,64 @@ +import pytest + +from geoservercloud.models.wmssettings import WmsSettings + + +@pytest.fixture +def mock_wms_settings(): + return { + "autoEscapeTemplateValues": False, + "bboxForEachCRS": False, + "cacheConfiguration": { + "enabled": False, + "maxEntries": 1000, + "maxEntrySize": 51200, + }, + "citeCompliant": False, + "defaultGroupStyleEnabled": True, + "dynamicStylingDisabled": False, + "enabled": True, + "featuresReprojectionDisabled": False, + "getFeatureInfoMimeTypeCheckingEnabled": False, + "getMapMimeTypeCheckingEnabled": False, + "interpolation": "Nearest", + "maxBuffer": 0, + "maxRenderingErrors": 0, + "maxRenderingTime": 0, + "maxRequestMemory": 0, + "maxRequestedDimensionValues": 100, + "name": "WMS", + "remoteStyleMaxRequestTime": 60000, + "remoteStyleTimeout": 30000, + "schemaBaseURL": "http://schemas.opengis.net", + "transformFeatureInfoDisabled": False, + "verbose": False, + "versions": { + "org.geotools.util.Version": [ + {"version": "1.1.1"}, + {"version": "1.3.0"}, + ] + }, + "watermark": { + "enabled": False, + "position": "BOT_RIGHT", + "transparency": 100, + }, + "workspace": {"name": "test_workspace"}, + } + + +def test_wms_settings_initialization(): + wms_settings = WmsSettings(workspace_name="test_workspace") + assert wms_settings.asdict() == {"workspace": "test_workspace"} + assert wms_settings.post_payload() == { + "wms": {"workspace": {"name": "test_workspace"}} + } + assert wms_settings.put_payload() == { + "wms": {"workspace": {"name": "test_workspace"}} + } + + +def test_wms_settings(mock_wms_settings): + wms_settings = WmsSettings.from_get_response_payload({"wms": mock_wms_settings}) + assert wms_settings.post_payload().get("wms") == mock_wms_settings + assert wms_settings.put_payload().get("wms") == mock_wms_settings diff --git a/tests/test_workspace.py b/tests/test_workspace.py index 87fb7be..a92788a 100644 --- a/tests/test_workspace.py +++ b/tests/test_workspace.py @@ -1,8 +1,56 @@ +from typing import Any + +import pytest import responses from responses import matchers from geoservercloud import GeoServerCloud -from tests.conftest import GEOSERVER_URL + + +@pytest.fixture +def mock_wms_settings(): + return { + "wms": { + "workspace": {"name": "test_workspace"}, + "enabled": True, + "name": "WMS", + "versions": { + "org.geotools.util.Version": [ + {"version": "1.1.1"}, + {"version": "1.3.0"}, + ] + }, + "citeCompliant": False, + "schemaBaseURL": "http://schemas.opengis.net", + "verbose": False, + "bboxForEachCRS": False, + "watermark": { + "enabled": False, + "position": "BOT_RIGHT", + "transparency": 100, + }, + "interpolation": "Nearest", + "getFeatureInfoMimeTypeCheckingEnabled": False, + "getMapMimeTypeCheckingEnabled": False, + "dynamicStylingDisabled": False, + "featuresReprojectionDisabled": False, + "maxBuffer": 0, + "maxRequestMemory": 0, + "maxRenderingTime": 0, + "maxRenderingErrors": 0, + "maxRequestedDimensionValues": 100, + "cacheConfiguration": { + "enabled": False, + "maxEntries": 1000, + "maxEntrySize": 51200, + }, + "remoteStyleMaxRequestTime": 60000, + "remoteStyleTimeout": 30000, + "defaultGroupStyleEnabled": True, + "transformFeatureInfoDisabled": False, + "autoEscapeTemplateValues": False, + } + } def test_list_workspaces(geoserver: GeoServerCloud) -> None: @@ -14,7 +62,7 @@ def test_list_workspaces(geoserver: GeoServerCloud) -> None: ] with responses.RequestsMock() as rsps: rsps.get( - url=f"{GEOSERVER_URL}/rest/workspaces.json", + url=f"{geoserver.url}/rest/workspaces.json", status=200, json={"workspaces": {"workspace": workspaces}}, ) @@ -26,7 +74,7 @@ def test_get_workspace_ok(geoserver: GeoServerCloud) -> None: workspace = {"name": workspace_name, "isolated": False} with responses.RequestsMock() as rsps: rsps.get( - url=f"{GEOSERVER_URL}/rest/workspaces/{workspace_name}.json", + url=f"{geoserver.url}/rest/workspaces/{workspace_name}.json", status=200, json={"workspace": workspace}, ) @@ -37,7 +85,7 @@ def test_get_workspace_not_found(geoserver: GeoServerCloud) -> None: workspace_name = "test_workspace" with responses.RequestsMock() as rsps: rsps.get( - url=f"{GEOSERVER_URL}/rest/workspaces/{workspace_name}.json", + url=f"{geoserver.url}/rest/workspaces/{workspace_name}.json", status=404, body=b"No such workspace: 'test_workspace' found", ) @@ -53,7 +101,7 @@ def test_create_workspace(geoserver: GeoServerCloud) -> None: with responses.RequestsMock() as rsps: rsps.post( - url=f"{GEOSERVER_URL}/rest/workspaces.json", + url=f"{geoserver.url}/rest/workspaces.json", status=201, body=b"test_workspace", match=[ @@ -79,7 +127,7 @@ def test_update_workspace(geoserver: GeoServerCloud) -> None: with responses.RequestsMock() as rsps: rsps.post( - url=f"{GEOSERVER_URL}/rest/workspaces.json", + url=f"{geoserver.url}/rest/workspaces.json", status=409, match=[ matchers.json_params_matcher( @@ -93,7 +141,7 @@ def test_update_workspace(geoserver: GeoServerCloud) -> None: ], ) rsps.put( - url=f"{GEOSERVER_URL}/rest/workspaces/{workspace_name}.json", + url=f"{geoserver.url}/rest/workspaces/{workspace_name}.json", match=[ matchers.json_params_matcher( { @@ -119,7 +167,7 @@ def test_delete_workspace(geoserver: GeoServerCloud) -> None: with responses.RequestsMock() as rsps: rsps.delete( - url=f"{GEOSERVER_URL}/rest/workspaces/{workspace}.json", + url=f"{geoserver.url}/rest/workspaces/{workspace}.json", status=200, body=b"", ) @@ -134,12 +182,12 @@ def test_recreate_workspace(geoserver: GeoServerCloud) -> None: with responses.RequestsMock() as rsps: rsps.delete( - url=f"{GEOSERVER_URL}/rest/workspaces/{workspace_name}.json", + url=f"{geoserver.url}/rest/workspaces/{workspace_name}.json", status=200, body=b"", ) rsps.post( - url=f"{GEOSERVER_URL}/rest/workspaces.json", + url=f"{geoserver.url}/rest/workspaces.json", status=201, body=b"test_workspace", match=[ @@ -159,61 +207,57 @@ def test_recreate_workspace(geoserver: GeoServerCloud) -> None: assert status_code == 201 -def test_publish_workspace(geoserver: GeoServerCloud) -> None: +def test_get_workspace_wms_settings( + geoserver: GeoServerCloud, mock_wms_settings: dict[str, Any] +) -> None: + workspace = "test_workspace" + with responses.RequestsMock() as rsps: + rsps.get( + url=f"{geoserver.url}/rest/services/wms/workspaces/{workspace}/settings.json", + status=200, + json=mock_wms_settings, + ) + content, status_code = geoserver.get_workspace_wms_settings(workspace) + assert isinstance(content, dict) + assert content.get("workspace") == "test_workspace" + assert status_code == 200 + + +def test_publish_workspace( + geoserver: GeoServerCloud, mock_wms_settings: dict[str, Any] +) -> None: workspace = "test_workspace" with responses.RequestsMock() as rsps: rsps.put( - url=f"{GEOSERVER_URL}/rest/services/wms/workspaces/{workspace}/settings.json", + url=f"{geoserver.url}/rest/services/wms/workspaces/{workspace}/settings.json", + status=200, + match=[matchers.json_params_matcher(mock_wms_settings)], + ) + + content, status_code = geoserver.publish_workspace(workspace) + assert content == "" + assert status_code == 200 + + +def test_set_service_locale(geoserver: GeoServerCloud) -> None: + workspace = "test_workspace" + + with responses.RequestsMock() as rsps: + rsps.put( + url=f"{geoserver.url}/rest/services/wms/workspaces/{workspace}/settings.json", status=200, match=[ matchers.json_params_matcher( { "wms": { - "workspace": {"name": workspace}, - "enabled": True, - "name": "WMS", - "versions": { - "org.geotools.util.Version": [ - {"version": "1.1.1"}, - {"version": "1.3.0"}, - ] - }, - "citeCompliant": False, - "schemaBaseURL": "http://schemas.opengis.net", - "verbose": False, - "bboxForEachCRS": False, - "watermark": { - "enabled": False, - "position": "BOT_RIGHT", - "transparency": 100, - }, - "interpolation": "Nearest", - "getFeatureInfoMimeTypeCheckingEnabled": False, - "getMapMimeTypeCheckingEnabled": False, - "dynamicStylingDisabled": False, - "featuresReprojectionDisabled": False, - "maxBuffer": 0, - "maxRequestMemory": 0, - "maxRenderingTime": 0, - "maxRenderingErrors": 0, - "maxRequestedDimensionValues": 100, - "cacheConfiguration": { - "enabled": False, - "maxEntries": 1000, - "maxEntrySize": 51200, - }, - "remoteStyleMaxRequestTime": 60000, - "remoteStyleTimeout": 30000, - "defaultGroupStyleEnabled": True, - "transformFeatureInfoDisabled": False, - "autoEscapeTemplateValues": False, + "defaultLocale": "fr", } } ) ], ) - content, status_code = geoserver.publish_workspace(workspace) + content, status_code = geoserver.set_default_locale_for_service(workspace, "fr") assert content == "" assert status_code == 200