[Closes GeoNode#3957] Increase "geoserver" tests coverage

codecov.yml

@@ -1,33 +1,34 @@
-codecov:
-  notify:
-    require_ci_to_pass: yes
-
-coverage:
-  precision: 2
-  round: down
-  range: "70...100"
-
-  status:
-    project: yes
-    patch: yes
-    changes: no
-
-parsers:
-  gcov:
-    branch_detection:
-      conditional: yes
-      loop: yes
-      method: no
-      macro: no
-
-comment:
-  layout: "header, diff"
-  behavior: default
-  require_changes: no
-
-ignore:
-  - */management/*
-  - geonode/contrib/*
-  - */test*
-  - */wsgi*
-  - */middleware*
+codecov:
+  notify:
+    require_ci_to_pass: yes
+
+coverage:
+  precision: 2
+  round: down
+  range: "70...100"
+
+  status:
+    project: yes
+    patch: yes
+    changes: no
+
+parsers:
+  gcov:
+    branch_detection:
+      conditional: yes
+      loop: yes
+      method: no
+      macro: no
+
+comment:
+  layout: "header, diff"
+  behavior: default
+  require_changes: no
+
+ignore:
+  - */management/*
+  - geonode/contrib/*
+  - */test*
+  - */wsgi*
+  - */middleware*
+  - geonode/qgis_server/*

geonode/geoserver/helpers.py

@@ -85,10 +85,11 @@ def check_geoserver_is_up():
     """Verifies all geoserver is running,
        this is needed to be able to upload.
     """
-    url = "%sweb/" % ogc_server_settings.LOCATION
+    url = "%s" % ogc_server_settings.LOCATION
     resp, content = http_client.request(url, "GET")
     msg = ('Cannot connect to the GeoServer at %s\nPlease make sure you '
-           'have started it.' % ogc_server_settings.LOCATION)
+           'have started it.' % url)
+    logger.debug(resp)
     assert resp['status'] == '200', msg
 
 

geonode/geoserver/tests.py

@@ -42,6 +42,9 @@
 from geonode.layers.populate_layers_data import create_layer_data
 from geonode.layers.models import Layer
 
+import logging
+logger = logging.getLogger(__name__)
+
 san_andres_y_providencia_sld = """<?xml version="1.0" encoding="UTF-8"?>
 <sld:StyledLayerDescriptor xmlns:sld="http://www.opengis.net/sld"
   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
@@ -901,6 +904,99 @@ def test_ogc_server_defaults(self):
         # Make sure we get None vs a KeyError when the key does not exist
         self.assertIsNone(ogc_settings.SFDSDFDSF)
 
+        # Testing OWS endpoints
+        from urlparse import urljoin
+        from django.core.urlresolvers import reverse
+        from .ows import _wcs_get_capabilities, _wfs_get_capabilities, _wms_get_capabilities
+        wcs = _wcs_get_capabilities()
+        logger.debug(wcs)
+        self.assertIsNotNone(wcs)
+
+        try:
+            wcs_url = urljoin(settings.SITEURL, reverse('wcs_endpoint'))
+        except BaseException:
+            wcs_url = urljoin(ogc_settings.PUBLIC_LOCATION, 'ows')
+        self.assertEquals(wcs,
+                          '%s?version=2.0.1&request=GetCapabilities&service=WCS' % wcs_url)
+
+        wfs = _wfs_get_capabilities()
+        logger.debug(wfs)
+        self.assertIsNotNone(wfs)
+
+        try:
+            wfs_url = urljoin(settings.SITEURL, reverse('wfs_endpoint'))
+        except BaseException:
+            wfs_url = urljoin(ogc_settings.PUBLIC_LOCATION, 'ows')
+        self.assertEquals(wfs,
+                          '%s?version=1.1.0&request=GetCapabilities&service=WFS' % wfs_url)
+
+        wms = _wms_get_capabilities()
+        logger.debug(wms)
+        self.assertIsNotNone(wms)
+
+        try:
+            wms_url = urljoin(settings.SITEURL, reverse('wms_endpoint'))
+        except BaseException:
+            wms_url = urljoin(ogc_settings.PUBLIC_LOCATION, 'ows')
+        self.assertEquals(wms,
+                          '%s?version=1.1.1&request=GetCapabilities&service=WMS' % wms_url)
+
+        # Test OWS Download Links
+        import urllib
+        from geonode.geoserver.ows import wcs_links, wfs_links, wms_links
+        instance = Layer.objects.all()[0]
+        bbox = instance.bbox
+        srid = instance.srid
+        height = 512
+        width = 512
+
+        # WMS Links
+        wms_links = wms_links(ogc_settings.public_url + 'wms?',
+                              instance.alternate.encode('utf-8'),
+                              bbox,
+                              srid,
+                              height,
+                              width)
+        self.assertIsNotNone(wms_links)
+        self.assertEquals(len(wms_links), 3)
+        wms_url = urljoin(ogc_settings.PUBLIC_LOCATION, 'wms')
+        identifier = urllib.urlencode({'layers': instance.alternate.encode('utf-8')})
+        for _link in wms_links:
+            logger.debug('%s --> %s' % (wms_url, _link[3]))
+            self.assertTrue(wms_url in _link[3])
+            logger.debug('%s --> %s' % (identifier, _link[3]))
+            self.assertTrue(identifier in _link[3])
+
+        # WFS Links
+        wfs_links = wfs_links(ogc_settings.public_url + 'wfs?',
+                              instance.alternate.encode('utf-8'),
+                              bbox,
+                              srid)
+        self.assertIsNotNone(wfs_links)
+        self.assertEquals(len(wfs_links), 6)
+        wfs_url = urljoin(ogc_settings.PUBLIC_LOCATION, 'wfs')
+        identifier = urllib.urlencode({'typename': instance.alternate.encode('utf-8')})
+        for _link in wfs_links:
+            logger.debug('%s --> %s' % (wfs_url, _link[3]))
+            self.assertTrue(wfs_url in _link[3])
+            logger.debug('%s --> %s' % (identifier, _link[3]))
+            self.assertTrue(identifier in _link[3])
+
+        # WCS Links
+        wcs_links = wcs_links(ogc_settings.public_url + 'wcs?',
+                              instance.alternate.encode('utf-8'),
+                              bbox,
+                              srid)
+        self.assertIsNotNone(wcs_links)
+        self.assertEquals(len(wcs_links), 2)
+        wcs_url = urljoin(ogc_settings.PUBLIC_LOCATION, 'wcs')
+        identifier = urllib.urlencode({'coverageid': instance.alternate.encode('utf-8')})
+        for _link in wcs_links:
+            logger.debug('%s --> %s' % (wcs_url, _link[3]))
+            self.assertTrue(wcs_url in _link[3])
+            logger.debug('%s --> %s' % (identifier, _link[3]))
+            self.assertTrue(identifier in _link[3])
+
     @on_ogc_backend(geoserver.BACKEND_PACKAGE)
     def test_importer_configuration(self):
         """

geonode/security/tests.py

@@ -600,3 +600,56 @@ def test_map_download(self):
 
         response = self.client.get(reverse('map_download', args=(the_map.id,)))
         self.assertTrue('Could not find downloadable layers for this map' in response.content)
+
+
+class GisBackendSignalsTests(ResourceTestCaseMixin, GeoNodeBaseTestSupport):
+
+    def setUp(self):
+        super(GisBackendSignalsTests, self).setUp()
+        if check_ogc_backend(geoserver.BACKEND_PACKAGE):
+            settings.OGC_SERVER['default']['GEOFENCE_SECURITY_ENABLED'] = True
+
+        self.user = 'admin'
+        self.passwd = 'admin'
+        self.list_url = reverse(
+            'api_dispatch_list',
+            kwargs={
+                'api_name': 'api',
+                'resource_name': 'layers'})
+        self.bulk_perms_url = reverse('bulk_permissions')
+        all_public()
+        self.perm_spec = {
+            "users": {"admin": ["view_resourcebase"]}, "groups": {}}
+
+    def test_save_and_delete_signals(self):
+        """Test that GeoServer Signals methods work as espected"""
+
+        layers = Layer.objects.all()[:2].values_list('id', flat=True)
+        test_perm_layer = Layer.objects.get(id=layers[0])
+
+        self.client.login(username='admin', password='admin')
+
+        if check_ogc_backend(geoserver.BACKEND_PACKAGE):
+            from geonode.geoserver.signals import (geoserver_pre_delete,
+                                                   geoserver_post_save,
+                                                   geoserver_post_save_local)
+            # Handle Layer Save and Upload Signals
+            geoserver_post_save(test_perm_layer, sender=Layer)
+            geoserver_post_save_local(test_perm_layer)
+
+            # Check instance bbox and links
+            self.assertIsNotNone(test_perm_layer.bbox)
+            self.assertIsNotNone(test_perm_layer.srid)
+            self.assertIsNotNone(test_perm_layer.link_set)
+            self.assertEquals(len(test_perm_layer.link_set.all()), 7)
+
+            # Layer Manipulation
+            from geonode.geoserver.helpers import check_geoserver_is_up
+            check_geoserver_is_up()
+
+            # Handle Layer Delete Signals
+            geoserver_pre_delete(test_perm_layer, sender=Layer)
+
+            # Check instance has been removed from GeoServer also
+            from geonode.geoserver.views import get_layer_capabilities
+            self.assertIsNone(get_layer_capabilities(test_perm_layer))

geonode/tests/integration.py

@@ -82,6 +82,11 @@
 
 logger = logging.getLogger(__name__)
 
+
+def _log(msg, *args):
+    logger.debug(msg, *args)
+
+
 # Reconnect post_save signals that is disconnected by populate_test_data
 reconnect_signals()
 
@@ -186,6 +191,45 @@ def test_layer_upload(self):
             o = json.loads(r.text)
             self.assertTrue('long-array-array' in o)
             self.assertTrue(len(o['long-array-array']) > 0)
+
+            from geonode.geoserver.helpers import (get_sld_for,
+                                                   fixup_style,
+                                                   set_layer_style,
+                                                   get_store,
+                                                   set_attributes_from_geoserver,
+                                                   set_styles,
+                                                   create_gs_thumbnail)
+
+            _log("0. ------------ %s " % saved_layer)
+            self.assertIsNotNone(saved_layer)
+            workspace, name = saved_layer.alternate.split(':')
+            self.assertIsNotNone(workspace)
+            self.assertIsNotNone(name)
+            ws = gs_catalog.get_workspace(workspace)
+            self.assertIsNotNone(ws)
+            store = get_store(gs_catalog, saved_layer.store, workspace=ws)
+            _log("1. ------------ %s " % store)
+            self.assertIsNotNone(store)
+
+            # Save layer attributes
+            set_attributes_from_geoserver(saved_layer)
+
+            # Save layer styles
+            set_styles(saved_layer, gs_catalog)
+
+            # set SLD
+            sld = saved_layer.default_style.sld_body if saved_layer.default_style else None
+            self.assertIsNotNone(sld)
+            _log("2. ------------ %s " % sld)
+            set_layer_style(saved_layer, saved_layer.alternate, sld)
+
+            fixup_style(gs_catalog, saved_layer.alternate, None)
+            self.assertIsNone(get_sld_for(gs_catalog, saved_layer))
+            _log("3. ------------ %s " % get_sld_for(gs_catalog, saved_layer))
+
+            create_gs_thumbnail(saved_layer, overwrite=True)
+            _log(saved_layer.get_thumbnail_url())
+            _log(saved_layer.has_thumbnail())
         try:
             saved_layer.set_default_permissions()
             url = reverse('layer_metadata', args=[saved_layer.service_typename])
@@ -194,6 +238,9 @@ def test_layer_upload(self):
         finally:
             # Clean up and completely delete the layer
             saved_layer.delete()
+            if check_ogc_backend(geoserver.BACKEND_PACKAGE):
+                from geonode.geoserver.helpers import cleanup
+                cleanup(saved_layer.name, saved_layer.uuid)
 
 
 @override_settings(SITEURL='http://localhost:8001/')

pavement.py

@@ -905,7 +905,7 @@ def run_tests(options):
     Executes the entire test suite.
     """
     if options.get('coverage'):
-        prefix = 'coverage run --branch --source=geonode --omit="*/management/*,geonode/contrib/*,*/test*,*/wsgi*,*/middleware*"'
+        prefix = 'coverage run --branch --source=geonode --omit="*/management/*,geonode/contrib/*,*/test*,*/wsgi*,*/middleware*,geonode/qgis_server/*"'
     else:
         prefix = 'python'
     local = options.get('local', 'false')  # travis uses default to false