Various improvements to GeoServer engine (#23)

* Various improvements to GeoServer engine
Reloading individual cluster nodes
More robust reload url formation
Fixed issue with sql view creation
Removed warnings on delete when resource does not exist

* Fix tests and lint

* More lint

* Fix tox-gh-actions

* Add Python 3.11

.github/workflows/unit_tests.yml

@@ -19,6 +19,7 @@ jobs:
           - windows
           - macos
         py:
+          - "3.11"
           - "3.10"
           - "3.9"
           - "3.8"
@@ -37,7 +38,7 @@ jobs:
       - name: Run tests
         run: tox
       - name: Coveralls
-        if: matrix.os == 'ubuntu' && matrix.py == 3.9
+        if: matrix.os == 'ubuntu' && matrix.py == 3.10
         run: coveralls --service=github
         env: 
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

tests/unit_tests/test_geoserver_engine.py

@@ -1756,7 +1756,6 @@ def test_delete_layer_warning(self, mock_delete, mock_logger):
 
         # Execute
         self.engine.delete_layer(layer_name, datastore=self.store_name)
-        mock_logger.warning.assert_called()
 
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.log')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.requests.delete')
@@ -1904,7 +1903,6 @@ def test_delete_coverage_store_with_warning(self, mock_delete, mock_log):
         self.assertIn(url, put_call_args[0][1]['url'])
         self.assertEqual(json, put_call_args[0][1]['params'])
         self.assertEqual({"Content-type": "application/json"}, put_call_args[0][1]['headers'])
-        mock_log.warning.assert_called()
 
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.log')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.requests.delete')
@@ -1983,7 +1981,6 @@ def test_delete_style_warning(self, mock_delete, mock_logger):
 
         # Create feature type call
         mock_delete.assert_called_with(url=url, auth=self.auth, headers=headers, params=params)
-        mock_logger.warning.assert_called()
 
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.log')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.requests.delete')
@@ -3254,11 +3251,12 @@ def test_create_style_overwrite_referenced_by_existing(self, mock_logger):
 
         mock_logger.error.assert_called()
 
+    @mock.patch('tethys_dataset_services.engines.geoserver_engine.GeoServerSpatialDatasetEngine.reload')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.GeoServerSpatialDatasetEngine.get_layer')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.GeoServerSpatialDatasetEngine.update_layer_styles')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.log')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.requests.post')
-    def test_create_sql_view_layer(self, mock_post, mock_logger, mock_update_layer_styles, mock_get_layer):
+    def test_create_sql_view_layer(self, mock_post, mock_logger, mock_update_layer_styles, mock_get_layer, mock_reload):
         mock_post.side_effect = [MockResponse(201), MockResponse(200)]
         store_id = f'{self.workspace_name}:foo'
         layer_name = self.layer_names[0]
@@ -3300,14 +3298,16 @@ def test_create_sql_view_layer(self, mock_post, mock_logger, mock_update_layer_s
             other_styles=None
         )
         mock_get_layer.assert_called()
+        mock_reload.assert_called()
 
+    @mock.patch('tethys_dataset_services.engines.geoserver_engine.GeoServerSpatialDatasetEngine.reload')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.GeoServerSpatialDatasetEngine.get_layer')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.GeoServerSpatialDatasetEngine.update_layer_styles')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.log')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.requests.post')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.GeoServerCatalog.get_default_workspace')
     def test_create_layer_create_feature_type_already_exists(self, mock_workspace, mock_post, mock_logger,
-                                                             mock_update_layer_styles, mock_get_layer):
+                                                             mock_update_layer_styles, mock_get_layer, mock_reload):
         mock_post.side_effect = [MockResponse(500, 'already exists'), MockResponse(200)]
         mock_workspace().name = self.workspace_name
         store_id = 'foo'
@@ -3350,6 +3350,7 @@ def test_create_layer_create_feature_type_already_exists(self, mock_workspace, m
             other_styles=None
         )
         mock_get_layer.assert_called()
+        mock_reload.assert_called()
 
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.log')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.requests.post')
@@ -3372,7 +3373,7 @@ def test_create_layer_create_sql_view_exception(self, mock_post, mock_logger):
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.log')
     @mock.patch('tethys_dataset_services.engines.geoserver_engine.requests.post')
     def test_create_sql_view_layer_gwc_error(self, mock_post, mock_logger, _):
-        mock_post.side_effect = [MockResponse(201)] + [MockResponse(500, 'GWC exception')] * 300
+        mock_post.side_effect = [MockResponse(201)] + [MockResponse(200)] + ([MockResponse(500, 'GWC exception')] * 300)
         store_id = f'{self.workspace_name}:foo'
         layer_name = self.layer_names[0]
         geometry_type = 'Point'

tethys_dataset_services/engines/geoserver_engine.py

@@ -8,6 +8,7 @@
 import requests
 from requests.auth import HTTPBasicAuth
 from io import BytesIO
+from urllib.parse import urlparse
 from xml.etree import ElementTree
 from zipfile import ZipFile, is_zipfile
 
@@ -96,7 +97,7 @@ def catalog(self):
             )
         return self._catalog
 
-    def __init__(self, endpoint, apikey=None, username=None, password=None, public_endpoint=None):
+    def __init__(self, endpoint, apikey=None, username=None, password=None, public_endpoint=None, node_ports=None):
         """
         Default constructor for Dataset Engines.
 
@@ -105,6 +106,7 @@ def __init__(self, endpoint, apikey=None, username=None, password=None, public_e
           apikey (string, optional): API key that will be used to authenticate with the dataset service.
           username (string, optional): Username that will be used to authenticate with the dataset service.
           password (string, optional): Password that will be used to authenticate with the dataset service.
+          node_ports(list<int>, optional): A list of ports of each node in a clustered GeoServer deployment.
         """
         # Set custom property /geoserver/rest/ -> /geoserver/gwc/rest/
         if public_endpoint:
@@ -114,6 +116,8 @@ def __init__(self, endpoint, apikey=None, username=None, password=None, public_e
         else:
             self._gwc_endpoint = endpoint.replace('rest', 'gwc/rest/')
 
+        self.node_ports = node_ports
+
         super(GeoServerSpatialDatasetEngine, self).__init__(
             endpoint=endpoint,
             apikey=apikey,
@@ -248,6 +252,27 @@ def _get_wfs_url(self, resource_id, output_format='GML3'):
 
         return wfs_url
 
+    def _get_node_endpoints(self, ports=None, public=True, gwc=False):
+        node_endpoints = []
+        if not gwc:
+            endpoint = self.public_endpoint if public and hasattr(self, 'public_endpoint') else self.endpoint
+        else:
+            endpoint = self.get_gwc_endpoint(public=public)
+
+        endpoint = f'{endpoint}/' if not endpoint.endswith('/') else endpoint
+
+        if ports is None:
+            ports = self.node_ports
+        log.debug(f"GeoServer Node Ports: {ports}")
+
+        if ports is not None:
+            gs_url = urlparse(endpoint)
+            for port in ports:
+                node_endpoints.append(f"{gs_url.scheme}://{gs_url.hostname}:{port}{gs_url.path}")
+        else:
+            node_endpoints.append(endpoint)
+        return node_endpoints
+
     @staticmethod
     def _handle_debug(return_object, debug):
         """
@@ -604,7 +629,7 @@ def get_gwc_endpoint(self, public=True):
             gs_endpoint = self._gwc_endpoint
 
         # Add trailing slash for consistency.
-        if gs_endpoint[-1] != '/':
+        if not gs_endpoint.endswith('/'):
             gs_endpoint += '/'
 
         return gs_endpoint
@@ -621,7 +646,7 @@ def get_ows_endpoint(self, workspace, public=True):
         gs_endpoint = gs_endpoint.replace('rest', '{0}/ows'.format(workspace))
 
         # Add trailing slash for consistency.
-        if gs_endpoint[-1] != '/':
+        if not gs_endpoint.endswith('/'):
             gs_endpoint += '/'
         return gs_endpoint
 
@@ -636,7 +661,7 @@ def get_wms_endpoint(self, public=True):
         gs_endpoint = gs_endpoint.replace('rest', 'wms')
 
         # Add trailing slash for consistency.
-        if gs_endpoint[-1] != '/':
+        if not gs_endpoint.endswith('/'):
             gs_endpoint += '/'
         return gs_endpoint
 
@@ -652,22 +677,13 @@ def reload(self, ports=None, public=True):
                               GeoServer are running in a clustered GeoServer configuration.
             public (bool): Use the public geoserver endpoint if True, otherwise use the internal endpoint.
         """
-        urls = []
-        gs_endpoint = self.public_endpoint if public and hasattr(self, 'public_endpoint') else self.endpoint
-
-        if ports is not None:
-            gs_endpoint_template = gs_endpoint.replace('8181', '{0}')
-            for port in ports:
-                urls.append(gs_endpoint_template.format(port) + 'reload')
-        else:
-            urls.append(gs_endpoint + 'reload')
-
-        log.debug("Catalog Reload URLS: {0}".format(urls))
+        node_endpoints = self._get_node_endpoints(ports=ports, public=public)
+        log.debug("Catalog Reload URLS: {0}".format(node_endpoints))
 
         response_dict = {'success': True, 'result': None, 'error': []}
-        for url in urls:
+        for endpoint in node_endpoints:
             try:
-                response = requests.post(url, auth=(self.username, self.password))
+                response = requests.post(f'{endpoint}reload', auth=(self.username, self.password))
 
                 if response.status_code != 200:
                     msg = "Catalog Reload Status Code {0}: {1}".format(response.status_code, response.text)
@@ -683,32 +699,23 @@ def reload(self, ports=None, public=True):
 
     def gwc_reload(self, ports=None, public=True):
         """
-                Reload the GeoWebCache configuration from disk.
-
-                Args:
-                    ports (iterable): A tuple or list of integers representing the ports on which different instances of
-                                      GeoServer are running in a clustered GeoServer configuration.
-                    public (bool): Use the public geoserver endpoint if True, otherwise use the internal
-                                            endpoint.
-                """
-        urls = []
-        gwc_endpoint = self.get_gwc_endpoint(public=public)
-
-        if ports is not None:
-            gs_endpoint_template = gwc_endpoint.replace('8181', '{0}')
-            for port in ports:
-                urls.append(gs_endpoint_template.format(port) + 'reload')
-        else:
-            urls.append(gwc_endpoint + 'reload')
+        Reload the GeoWebCache configuration from disk.
 
-        log.debug("GeoWebCache Reload URLS: {0}".format(urls))
+        Args:
+            ports (iterable): A tuple or list of integers representing the ports on which different instances of
+                                GeoServer are running in a clustered GeoServer configuration.
+            public (bool): Use the public geoserver endpoint if True, otherwise use the internal
+                                    endpoint.
+        """
+        node_endpoints = self._get_node_endpoints(ports=ports, public=public, gwc=True)
+        log.debug("GeoWebCache Reload URLS: {0}".format(node_endpoints))
 
         response_dict = {'success': True, 'result': None, 'error': []}
-        for url in urls:
+        for endpoint in node_endpoints:
             retries_remaining = 3
             while retries_remaining > 0:
                 try:
-                    response = requests.post(url, auth=(self.username, self.password))
+                    response = requests.post(f'{endpoint}reload', auth=(self.username, self.password))
 
                     if response.status_code != 200:
                         msg = "GeoWebCache Reload Status Code {0}: {1}".format(response.status_code, response.text)
@@ -1468,6 +1475,9 @@ def create_sql_view_layer(self, store_id, layer_name, geometry_type, srid, sql,
                     log.error(exception)
                     raise exception
 
+        # Reload before attempting to update styles to avoid issues
+        self.reload()
+
         # Add styles to new layer
         self.update_layer_styles(
             layer_id=layer_id,
@@ -2416,9 +2426,7 @@ def delete_layer(self, layer_id, datastore, recurse=False):
         # Raise an exception if status code is not what we expect
         if response.status_code != 200:
             if response.status_code in self.WARNING_STATUS_CODES:
-                msg = "Delete Layer Status Code {0}: {1}".format(response.status_code, response.text)
-                exception = requests.RequestException(msg, response=response)
-                log.warning(exception)
+                pass
             else:
                 msg = "Delete Layer Status Code {0}: {1}".format(response.status_code, response.text)
                 exception = requests.RequestException(msg, response=response)
@@ -2445,7 +2453,7 @@ def delete_layer_group(self, layer_group_id):
         response = requests.delete(url, auth=(self.username, self.password))
         if response.status_code != 200:
             if response.status_code == 404 and "No such layer group" in response.text:
-                return
+                pass
             else:
                 msg = "Delete Layer Group Status Code {0}: {1}".format(response.status_code, response.text)
                 exception = requests.RequestException(msg, response=response)
@@ -2558,9 +2566,7 @@ def delete_coverage_store(self, store_id, recurse=True, purge=True):
 
         if response.status_code != 200:
             if response.status_code in self.WARNING_STATUS_CODES:
-                msg = "Delete Coverage Store Status Code {0}: {1}".format(response.status_code, response.text)
-                exception = requests.RequestException(msg, response=response)
-                log.warning(exception)
+                pass
             else:
                 msg = "Delete Coverage Store Status Code {0}: {1}".format(response.status_code, response.text)
                 exception = requests.RequestException(msg, response=response)
@@ -2605,9 +2611,7 @@ def delete_style(self, style_id, purge=False):
         # Raise an exception if status code is not what we expect
         if response.status_code != 200:
             if response.status_code in self.WARNING_STATUS_CODES:
-                msg = "Delete Style Status Code {0}: {1}".format(response.status_code, response.text)
-                exception = requests.RequestException(msg, response=response)
-                log.warning(exception)
+                pass
             else:
                 msg = "Delete Style Status Code {0}: {1}".format(response.status_code, response.text)
                 exception = requests.RequestException(msg, response=response)

tox.ini

@@ -1,13 +1,14 @@
 [tox]
 isolated_build = True
-envlist = py39, flake8, clean
+envlist = py37, py38, py39, py310, py311, flake8, clean
 
 [gh-actions]
 python =
     3.7: py37
     3.8: py38
     3.9: py39
     3.10: py310
+    3.11: py311
 
 [testenv]
 deps =