Merge pull request #101 from Kilometerfresserin/issue_100

fix #100 and add unittest for layergroup methods

.gitignore

@@ -1,7 +1,4 @@
-venv/
-data/
 style.sld
-.vscode/
 FunctionsToImplement.py
 record.txt
 package_test.py
@@ -126,6 +123,9 @@ venv/
 ENV/
 env.bak/
 venv.bak/
+.vscode/
+.direnv/
+venv/
 
 # Spyder project settings
 .spyderproject

docs/source/change_log.rst

@@ -4,6 +4,7 @@ Change Log
 ``Master branch``
 ^^^^^^^^^^^^^^^^^
 
+* Allow deletion of layergroups from workspaces (see issue `#100 <https://github.com/gicait/geoserver-rest/issues/100>`) and add unittests for the layergroup methods.
 * Fix json-bug in create_coveragestore method (see issue `#86 <https://github.com/gicait/geoserver-rest/issues/86>`)
 
 ``[v2.4.0 - 2023-01-10]``

geo/Geoserver.py

@@ -731,7 +731,7 @@ def create_layergroup(
 
             # check if it already exist in Geoserver
             try:
-                existing_layergroup = self.get_layergroup(name)
+                existing_layergroup = self.get_layergroup(name, workspace=workspace)
             except:
                 existing_layergroup = None
 
@@ -945,14 +945,27 @@ def update_layergroup(
     def delete_layergroup(
         self,
         layergroup_name: str,
+        workspace: Optional[str] = None
     ) -> str:
+        """
+        Delete a layer group from the geoserver and raise an exception 
+        in case the layer group does not exist, or the geoserver is unavailable.
+
+        Parameters
+        ----------
+        layergroup_name: str, required The name of the layer group to be deleted
+        workspace: str, optional The workspace the layergroup is located in
+        """
+
+
         try:
-            if self.get_layergroup(layer_name=layergroup_name) is None:
-                raise Exception(
-                    f"Layer group: {layergroup_name} is not a valid layer group in the Geoserver instance"
-                )
+            #raises an exception in case the layer group doesn't exist
+            self.get_layergroup(layer_name=layergroup_name, workspace=workspace) 
 
-            url = f"{self.service_url}/rest/layergroups/{layergroup_name}"
+            if workspace == None:
+                url = f"{self.service_url}/rest/layergroups/{layergroup_name}"
+            else:
+                url = f"{self.service_url}/rest/workspaces/{workspace}/layergroups/{layergroup_name}"
 
             r = self._requests(url=url, method="delete")
             if r.status_code == 200:

requirements_dev.txt

@@ -3,3 +3,5 @@ black
 flake8
 sphinx>=1.7
 pre-commit
+environs
+ddt

tests/.env_template

@@ -0,0 +1,4 @@
+# Login information for geoserver
+GEOSERVER_URL="http://localhost:8000/geoserver"
+GEOSERVER_USER="admin"
+GEOSERVER_PASSWORD="geoserver"

tests/test_layergroup.py

@@ -0,0 +1,207 @@
+import os
+import unittest
+from environs import Env
+
+from unittest.mock import MagicMock, patch #allows replacing methods ans Objects by Mocks
+from ddt import data, ddt, unpack #allows running the same test with different parameters
+
+from geo.Geoserver import Geoserver
+
+@ddt
+class TestLayerGroup(unittest.TestCase):
+    """
+    Tests all layergroup related methods of the geoserver class.
+
+    How to use:
+
+    You need to have a geoserver that you can use for testing.
+    In order to run the test, you need to create an .env file based on the .env_template.
+    Adjust the .env file with the url and login information for the server you're testing against.
+
+    You can run this test by executing:
+    python -m unittest tests.test_layergroup
+
+    The test will temporarily create a new workspace "unittest" on your geoserver. This workspace
+    (or any workspace of that name that previously existed on your geoserver) will be deleted
+    after the test.
+
+    The setup of this test relies on the create_workspace, delete_workspace, 
+    and create_coveragestore methods.
+
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        '''
+        is run once when setting up the test class
+
+        sets up a geoserver instance, builds a workspace, 
+        and uploads 2 example layers that we can later on use to build our layergroups
+        '''
+
+        env = Env()
+        env.read_env()
+
+        cls.geoserver = Geoserver(
+            os.environ["GEOSERVER_URL"], 
+            username= os.environ["GEOSERVER_USER"],
+            password=os.environ["GEOSERVER_PASSWORD"]
+            )
+
+        # set up workspace for testing
+        try:
+            cls.geoserver.create_workspace(workspace="unittest")
+        except:
+            # is raised when the workspace exists already
+            cls.geoserver.delete_workspace(workspace="unittest")
+            cls.geoserver.create_workspace(workspace="unittest")
+
+        # upload sample layers to the testing workspace
+        # credits for the sample file: https://github.com/mommermi/geotiff_sample
+
+        cls.geoserver.create_coveragestore(
+                layer_name="test_layer_1",
+                path="tests/data/sample_geotiff.tif", 
+                workspace="unittest"
+            )
+
+        cls.geoserver.create_coveragestore(
+                layer_name="test_layer_2",
+                path="tests/data/sample_geotiff.tif", 
+                workspace="unittest"
+            )
+
+
+    @classmethod
+    def tearDownClass(cls):
+        '''
+        is run when tearing down test class
+        '''
+        cls.geoserver.delete_workspace(workspace="unittest")
+
+    def setUp(self):
+        '''
+        is run before each individual test method
+        '''
+        pass
+
+    def tearDown(self):
+        '''
+        is run after each individual test method
+        '''
+        pass
+
+    @data(
+        ("NonExistingLayerGroup", None), #layergroup in global workspace
+        ("NonExistingLayerGroup", "unittest") #layergroup in our workspace
+        )
+    @unpack
+    def test_get_layergroup_that_doesnt_exist(self, layergroup_name, workspace):
+
+        with self.assertRaises(Exception):
+
+            self.geoserver.get_layergroup(
+                layer_name=layergroup_name,
+                workspace=workspace
+            )
+
+    @data(
+        ("NonExistingLayerGroup", None), #layergroup in global workspace
+        ("NonExistingLayerGroup", "unittest") #layergroup in our workspace
+        )
+    @unpack
+    def test_delete_layergroup_that_doesnt_exist(self, layergroup_name, workspace):
+
+        with self.assertRaises(Exception):
+
+            self.geoserver.delete_layergroup(
+                layergroup_name=layergroup_name,
+                workspace=workspace
+            )
+
+    @data(
+        (
+            "test-layergroup-name", #name
+            "single", #mode
+            "test-layergroup-title", #title
+            "test-layergroup-abstract-text", #abstractText
+            ["keyword_1", "keyword_2"], #keywords
+            "unittest" #workspace
+        ),
+        (
+            "test-layergroup-name", #name
+            "single", #mode
+            "test-layergroup-title", #title
+            "test-layergroup-abstract-text", #abstractText
+            ["keyword_1", "keyword_2"], #keywords
+            None #workspace
+        ),
+        )
+    @unpack
+    def test_create_and_get_and_delete_layergroup(self, name, mode, title, abstract_text, keywords, workspace):
+
+        self.geoserver.create_layergroup(
+            name = name,
+            mode = mode,
+            title = title,
+            abstract_text = abstract_text,
+            layers = ["test_layer_1", "test_layer_2"],
+            workspace = workspace,
+            keywords = keywords
+        )
+
+        layer_group_dict = self.geoserver.get_layergroup(
+            layer_name = name,
+            workspace=workspace,
+        )
+
+        self.assertIsInstance(layer_group_dict, dict)
+
+        self.assertEqual(
+            layer_group_dict["layerGroup"]["name"], name
+        )
+
+        self.assertEqual(
+            layer_group_dict["layerGroup"]["mode"], mode.upper()
+        )
+
+        self.assertEqual(
+            layer_group_dict["layerGroup"]["title"], title
+        )
+
+        self.assertEqual(
+            layer_group_dict["layerGroup"]["abstractTxt"], abstract_text
+        )
+
+        if workspace is not None:
+            self.assertEqual(
+                layer_group_dict["layerGroup"]["workspace"]["name"],
+                workspace,
+                "layer_group has not been assigned to the right workspace"
+            )
+
+        self.assertEqual(
+            len(layer_group_dict["layerGroup"]["publishables"]["published"]),
+            2,
+            f'{len(layer_group_dict["layerGroup"]["publishables"]["published"])} instead of 2 layers in layergroup'
+        )
+
+        self.assertEqual(
+            layer_group_dict["layerGroup"]["keywords"]["string"],
+            keywords
+        )
+
+        self.geoserver.delete_layergroup(
+            layergroup_name=name,
+            workspace=workspace
+        )
+
+        with self.assertRaises(Exception) as assertion:
+            assertion.msg = "Layer group has not been deleted properly."
+            self.geoserver.get_layergroup(
+            layer_name = name,
+            workspace=workspace,
+        )
+
+if __name__ == '__main__':
+    unittest.main()