[Fixes #13833] Improve modularization for better encapsulation

geonode/base/models.py

@@ -62,6 +62,7 @@
 from geonode import GeoNodeException
 
 from geonode.base import enumerations
+from geonode.geoserver.ows import _wms_link
 from geonode.singleton import SingletonModel
 from geonode.groups.conf import settings as groups_settings
 from geonode.base.bbox_utils import BBOXHelper, polygon_from_bbox
@@ -952,6 +953,22 @@ def compact_permission_labels(cls):
     def raw_abstract(self):
         return self._remove_html_tags(self.abstract)
 
+    @property
+    def can_be_downloaded(self):
+        return self.subtype in {"vector", "raster", "vector_time"}
+
+    @property
+    def can_have_wfs_links(self):
+        return self.subtype == "vector"
+
+    @property
+    def can_have_wps_links(self):
+        return self.subtype in {"vector", "tileStore", "remote", "wmsStore", "vector_time"}
+
+    @property
+    def can_have_style(self):
+        return self.subtype not in {"tileStore", "remote"}
+
     @property
     def raw_purpose(self):
         return self._remove_html_tags(self.purpose)
@@ -1509,7 +1526,7 @@ def get_legend_url(self, style_name=None):
     def get_ows_url(self):
         """Return URL for OGC WMS server None if it does not exist."""
         try:
-            ows_link = self.link_set.get(name="OGC:WMS")
+            ows_link = self.link_set.get(name="OGC:WMS") or self.link_set.get(name="OGC:WFS")
         except Link.DoesNotExist:
             return None
         else:
@@ -1914,6 +1931,18 @@ def get_linked_resources(self, as_target: bool = False):
             else LinkedResource.get_linked_resources(source=self)
         )
 
+    def prepare_wms_links(self, wms_url, identifier, bbox, srid, height, width):
+        types = [
+            ("jpg", _("JPEG"), "image/jpeg"),
+            ("pdf", _("PDF"), "application/pdf"),
+            ("png", _("PNG"), "image/png"),
+        ]
+        output = []
+        for ext, name, mime in types:
+            url = _wms_link(wms_url, identifier, mime, height, width, srid, bbox)
+            output.append((ext, name, mime, url))
+        return output
+
 
 class LinkManager(models.Manager):
     """Helper class to access links grouped by type"""

geonode/geoserver/helpers.py

@@ -1012,7 +1012,7 @@ def set_attributes_from_geoserver(layer, overwrite=False):
                 f"Error while retrieving info for {layer.subtype} '{layer.alternate or layer.typename}'", exc_info=True
             )
             attribute_map = []
-    elif layer.subtype in {"vector", "tileStore", "remote", "wmsStore", "vector_time"}:
+    elif layer.can_have_wps_links:
         typename = layer.alternate if layer.alternate else layer.typename
         logger.info(f"Getting WFS info for {layer.subtype} '{typename}'")
         dft_url_path = re.sub(r"\/wms\/?$", "/", server_url)

geonode/geoserver/ows.py

@@ -354,16 +354,3 @@ def _wms_link(wms_url, identifier, mime, height, width, srid=None, bbox=None):
         wms_params["bbox"] = bbox
     _query_separator = "?" if not wms_url.endswith("?") else ""
     return f"{wms_url}{_query_separator}{urlencode(wms_params)}"
-
-
-def wms_links(wms_url, identifier, bbox, srid, height, width):
-    types = [
-        ("jpg", _("JPEG"), "image/jpeg"),
-        ("pdf", _("PDF"), "application/pdf"),
-        ("png", _("PNG"), "image/png"),
-    ]
-    output = []
-    for ext, name, mime in types:
-        url = _wms_link(wms_url, identifier, mime, height, width, srid, bbox)
-        output.append((ext, name, mime, url))
-    return output

geonode/geoserver/tests/test_server.py

@@ -1034,7 +1034,7 @@ def test_ogc_server_defaults(self):
         self.assertIn("version=1.3.0", wms)
 
         # Test OWS Download Links
-        from geonode.geoserver.ows import wcs_links, wfs_links, wms_links
+        from geonode.geoserver.ows import wcs_links, wfs_links
 
         instance = create_single_dataset("san_andres_y_providencia_water")
         instance.name = "san_andres_y_providencia_water"
@@ -1097,7 +1097,9 @@ def test_ogc_server_defaults(self):
         self.assertIsNotNone(instance.default_style.name)
 
         # WMS Links
-        wms_links = wms_links(f"{ogc_settings.public_url}wms?", instance.alternate, bbox, srid, height, width)
+        wms_links = instance.prepare_wms_links(
+            f"{ogc_settings.public_url}wms?", instance.alternate, bbox, srid, height, width
+        )
         self.assertIsNotNone(wms_links)
         self.assertEqual(len(wms_links), 3)
         wms_url = urljoin(ogc_settings.PUBLIC_LOCATION, "wms")

geonode/layers/download_handler.py

@@ -75,7 +75,7 @@ def download_url(self):
         resource = self.get_resource()
         if not resource:
             return None
-        if resource.subtype not in ["vector", "raster", "vector_time"]:
+        if not resource.can_be_downloaded:
             logger.info("Download URL is available only for datasets that have been harvested and copied locally")
             return None
 
@@ -117,7 +117,7 @@ def process_dowload(self, resource=None):
             logger.error("The format provided is not valid for the selected resource")
             return JsonResponse({"error": "The format provided is not valid for the selected resource"}, status=500)
 
-        _format = "application/zip" if resource.is_vector() else "image/tiff"
+        _format = resource.download_format
         # getting default payload
         tpl = get_template("geoserver/dataset_download.xml")
         ctx = {"alternate": resource.alternate, "download_format": download_format or _format}

geonode/layers/models.py

@@ -394,11 +394,15 @@ def get_linked_resources(self, as_target: bool = False):
 
     @property
     def download_url(self):
-        if self.subtype not in ["vector", "raster", "vector_time"]:
+        if self.can_be_downloaded:
             logger.info("Download URL is available only for datasets that have been harvested and copied locally")
             return None
         return build_absolute_uri(reverse("dataset_download", args=(self.alternate,)))
 
+    @property
+    def download_format(self):
+        return "application/zip" if self.is_vector() else "image/tiff"
+
     @property
     def maplayers(self):
         from geonode.maps.models import MapLayer

geonode/utils.py

@@ -1211,7 +1211,7 @@ def set_resource_default_links(instance, layer, prune=False, **kwargs):
         logger.debug(" -- Resource Links[Prune old links]...done!")
 
     if check_ogc_backend(geoserver.BACKEND_PACKAGE):
-        from geonode.geoserver.ows import wcs_links, wfs_links, wms_links
+        from geonode.geoserver.ows import wcs_links, wfs_links
         from geonode.geoserver.helpers import ogc_server_settings, gs_catalog
 
         # Compute parameters for the new links
@@ -1273,7 +1273,7 @@ def set_resource_default_links(instance, layer, prune=False, **kwargs):
         # Set download links for WMS, WCS or WFS and KML
         logger.debug(" -- Resource Links[Set download links for WMS, WCS or WFS and KML]...")
         instance_ows_url = f"{instance.ows_url}?" if instance.ows_url else f"{ogc_server_settings.public_url}ows?"
-        links = wms_links(instance_ows_url, instance.alternate, bbox, srid, height, width)
+        links = instance.prepare_wms_links(instance_ows_url, instance.alternate, bbox, srid, height, width)
 
         for ext, name, mime, wms_url in links:
             try:
@@ -1293,7 +1293,7 @@ def set_resource_default_links(instance, layer, prune=False, **kwargs):
                     resource=instance.resourcebase_ptr, name=gettext_lazy(name), link_type="image"
                 ).update(**_d)
 
-        if instance.subtype == "vector":
+        if instance.can_have_wfs_links:
             links = wfs_links(
                 instance_ows_url,
                 instance.alternate,
@@ -1372,7 +1372,7 @@ def set_resource_default_links(instance, layer, prune=False, **kwargs):
         # Legend link
         logger.debug(" -- Resource Links[Legend link]...")
         try:
-            if instance.subtype not in ["tileStore", "remote"]:
+            if instance.can_have_style:
                 for style in set(
                     list(instance.styles.all())
                     + [
@@ -1452,7 +1452,7 @@ def set_resource_default_links(instance, layer, prune=False, **kwargs):
                         ),
                     )
 
-                if instance.subtype == "vector":
+                if instance.can_have_wfs_links:
                     ogc_wfs_url = instance.ows_url or urljoin(ogc_server_settings.public_url, "ows")
                     ogc_wfs_name = f"OGC WFS: {instance.workspace} Service"
                     if (