feature: add supprot for external | url method when create coveragestore

geo/Geoserver.py

@@ -171,6 +171,8 @@ def _requests(self,
             return requests.put(url, auth=(self.username, self.password), **kwargs, **self.request_options)
         elif method.lower() == "delete":
             return requests.delete(url, auth=(self.username, self.password), **kwargs, **self.request_options)
+        else:
+            raise Exception("unsupported http method name.")
 
     # _______________________________________________________________________________________________
     #
@@ -547,6 +549,7 @@ def create_coveragestore(
         layer_name: Optional[str] = None,
         file_type: str = "GeoTIFF",
         content_type: str = "image/tiff",
+        method: str = "file"
     ):
         """
         Creates the coverage store; Data will be uploaded to the server.
@@ -563,6 +566,8 @@ def create_coveragestore(
             The type of the file.
         content_type : str
             The content type of the file.
+        method : str
+            file | url | external | remote
 
         Returns
         -------
@@ -586,21 +591,30 @@ def create_coveragestore(
                 layer_name = f[0]
 
         file_type = file_type.lower()
+        if file_type == "netcdf":
+            # files such as netcdf contain multiple layers, which means a single coverage name cannot be specified.
+            url = "{0}/rest/workspaces/{1}/coveragestores/{2}/{3}.{4}".format(
+                self.service_url, workspace, layer_name, method, file_type
+            )
+        else:
+            url = "{0}/rest/workspaces/{1}/coveragestores/{2}/{3}.{4}?coverageName={2}".format(
+                self.service_url, workspace, layer_name, method, file_type
+            )
 
-        url = "{0}/rest/workspaces/{1}/coveragestores/{2}/file.{3}?coverageName={2}".format(
-            self.service_url, workspace, layer_name, file_type
-        )
+        if method == 'file':
+            headers = {"content-type": content_type, "Accept": "application/json"}
+            with open(path, "rb") as f:
+                r = self._requests(method="put", url=url, data=f, headers=headers)
+        else:
+            headers = {"content-type": "text/plain", "Accept": "application/json"}
+            r = self._requests(method="put", url=url, data=path, headers=headers)
 
-        headers = {"content-type": content_type, "Accept": "application/json"}
+        if r.status_code == 201:
+            return r.json()
+        else:
+            raise GeoserverException(r.status_code, r.content)
 
-        r = None
-        with open(path, "rb") as f:
-            r = self._requests(method="put", url=url, data=f, headers=headers)
 
-            if r.status_code == 201:
-                return r.json()
-            else:
-                raise GeoserverException(r.status_code, r.content)
 
     def publish_time_dimension_to_coveragestore(
         self,

tests/test_geoserver.py

@@ -548,3 +548,72 @@ def test_delete(self):
 class TestOther:
     def test_classified_xml(self):
         classified_xml("test", "kamal", [4, 5, 3, 12], color_ramp="hot")
+
+
+class TestCoveragestore:
+    def setup_method(self):
+        self.workspace_name = "test_workspace"
+        self.layer_name = "netcdf"
+        self.path = f"{HERE}/data/tos_O1_2001-2002.nc"
+        self.url = "http://localhost:8000/tos_O1_2001-2002.nc"
+        self.type = "NetCDF"
+        try:
+            geo.create_workspace(workspace=self.workspace_name)
+        except:
+            geo.delete_workspace(workspace=self.workspace_name)
+            geo.create_workspace(workspace=self.workspace_name)
+
+    def teardown_method(self):
+        geo.delete_workspace(workspace=self.workspace_name)
+
+    def _verify_coveragestore(self, response):
+        """
+        Helper method to verify coveragestore creation
+        """
+        assert response["coverageStore"]["name"] == self.layer_name
+        coveragestore = geo.get_coveragestore(
+            coveragestore_name=self.layer_name,
+            workspace=self.workspace_name
+        )
+        assert coveragestore["coverageStore"]["name"] == self.layer_name
+        assert coveragestore["coverageStore"]["type"] == self.type
+        assert coveragestore["coverageStore"]["workspace"]["name"] == self.workspace_name
+
+    def _test_create_coveragestore(self, method, path=None):
+        """
+        Helper method to test coveragestore creation with different methods
+        """
+        try:
+            resp = geo.create_coveragestore(
+                path=path or self.path,
+                workspace=self.workspace_name,
+                layer_name=self.layer_name,
+                file_type=self.type,
+                content_type="application/x-netcdf",
+                method=method
+            )
+            self._verify_coveragestore(resp)
+        finally:
+            geo.delete_coveragestore(
+                coveragestore_name=self.layer_name,
+                workspace=self.workspace_name
+            )
+
+    def test_create_coveragestore_using_file_method(self):
+        """
+        Tests that a coveragestore can be created using "file" method
+        """
+        self._test_create_coveragestore("file")
+
+    def test_create_coveragestore_using_external_method(self):
+        """
+        Tests that a coveragestore can be created using "external" method
+        """
+        self._test_create_coveragestore("external")
+
+    @pytest.mark.skip(reason="Only setup for local testing.")
+    def test_create_coveragestore_using_url_method(self):
+        """
+        Tests that a coveragestore can be created using "url" method
+        """
+        self._test_create_coveragestore("url", self.url)