Skip to content
Permalink
Browse files

Merge pull request #3298 from elpaso/offline_editing_tests2

[offline editing] Added insert and update tests
  • Loading branch information
elpaso committed Jul 12, 2016
2 parents 2ccc7e1 + b5f02e3 commit e59fda26e662497179fb1488d370c169730efee8
@@ -1,6 +1,18 @@
# -*- coding: utf-8 -*-
"""QGIS Unit test utils for offline editing tests.
There are three layers referenced through the code:
- the "online_layer" is the layer being edited online (WFS or PostGIS) layer inside
QGIS client
- the "offline_layer" (SQLite)
- the "layer", is the shapefile layer that is served by QGIS Server WFS, in case of
PostGIS, this will be the same layer referenced by online_layer
Each test simulates one working session.
When testing on PostGIS, the first two layers will be exactly the same object.
.. note:: This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
@@ -16,6 +28,7 @@
__revision__ = '$Format:%H$'

from time import sleep
import os

from qgis.core import (
QgsFeature,
@@ -37,6 +50,12 @@
(4, 'name 4', QgsPoint(10, 46.5)),
]

# Additional features for insert test
TEST_FEATURES_INSERT = [
(5, 'name 5', QgsPoint(9.7, 45.7)),
(6, 'name 6', QgsPoint(10.6, 46.6)),
]


class OfflineTestBase(object):

@@ -47,22 +66,25 @@ def _setUp(self):
# Setup: create some features for the test layer
features = []
layer = self._getLayer('test_point')
assert layer.startEditing()
for id, name, geom in TEST_FEATURES:
f = QgsFeature(layer.pendingFields())
f['id'] = id
f['name'] = name
f.setGeometry(QgsGeometry.fromPoint(geom))
features.append(f)
layer.dataProvider().addFeatures(features)
# Add the remote layer
layer.addFeatures(features)
assert layer.commitChanges()
# Add the online layer
self.registry = QgsMapLayerRegistry.instance()
self.registry.removeAllMapLayers()
assert self.registry.addMapLayer(self._getOnlineLayer('test_point')) is not None

def _tearDown(self):
"""Called by tearDown: run after each test."""
# Clear test layers
self._clearLayer('test_point')
# Delete the sqlite db
#os.unlink(os.path.join(self.temp_path, 'offlineDbFile.sqlite'))
pass

@classmethod
def _compareFeature(cls, layer, attributes):
@@ -71,11 +93,10 @@ def _compareFeature(cls, layer, attributes):
return f['name'] == attributes[1] and f.geometry().asPoint().toString() == attributes[2].toString()

@classmethod
def _clearLayer(cls, layer_name):
def _clearLayer(cls, layer):
"""
Delete all features from the backend layer
Delete all features from the given layer
"""
layer = cls._getLayer(layer_name)
layer.startEditing()
layer.deleteFeatures([f.id() for f in layer.getFeatures()])
layer.commitChanges()
@@ -108,19 +129,26 @@ def _getFeatureByAttribute(cls, layer, attr_name, attr_value):
raise Exception("Wrong attributes in WFS layer %s" %
layer.name())

def test_offlineConversion(self):
def _testInit(self):
"""
Preliminary checks for each test
"""
# goes offline
ol = QgsOfflineEditing()
online_layer = list(self.registry.mapLayers().values())[0]
self.assertTrue(online_layer.hasGeometryType())
# Check we have 3 features
# Check we have features
self.assertEqual(len([f for f in online_layer.getFeatures()]), len(TEST_FEATURES))
self.assertTrue(ol.convertToOfflineProject(self.temp_path, 'offlineDbFile.sqlite', [online_layer.id()]))
offline_layer = list(self.registry.mapLayers().values())[0]
self.assertTrue(offline_layer.hasGeometryType())
self.assertTrue(offline_layer.isValid())
self.assertTrue(offline_layer.name().find('(offline)') > -1)
self.assertEqual(len([f for f in offline_layer.getFeatures()]), len(TEST_FEATURES))
return ol, offline_layer

def test_updateFeatures(self):
ol, offline_layer = self._testInit()
# Edit feature 2
feat2 = self._getFeatureByAttribute(offline_layer, 'name', "'name 2'")
self.assertTrue(offline_layer.startEditing())
@@ -131,8 +159,8 @@ def test_offlineConversion(self):
self.assertTrue(ol.isOfflineProject())
# Sync
ol.synchronize()
sleep(2)
# Does anybody know why the sleep is needed? Is that a threaded WFS consequence?
sleep(1)
online_layer = list(self.registry.mapLayers().values())[0]
self.assertTrue(online_layer.isValid())
self.assertFalse(online_layer.name().find('(offline)') > -1)
@@ -176,3 +204,98 @@ def test_offlineConversion(self):
self.assertTrue(self._compareFeature(layer, TEST_FEATURES[1 - 1]))
self.assertTrue(self._compareFeature(layer, TEST_FEATURES[2 - 1]))
self.assertTrue(self._compareFeature(layer, TEST_FEATURES[3 - 1]))

def test_deleteOneFeature(self):
"""
Delete a single feature
"""
ol, offline_layer = self._testInit()
# Delete feature 3
feat3 = self._getFeatureByAttribute(offline_layer, 'name', "'name 3'")
self.assertTrue(offline_layer.startEditing())
self.assertTrue(offline_layer.deleteFeatures([feat3.id()]))
self.assertTrue(offline_layer.commitChanges())
self.assertRaises(Exception, lambda: self._getFeatureByAttribute(offline_layer, 'name', "'name 3'"))
self.assertTrue(ol.isOfflineProject())
# Sync
ol.synchronize()
# Does anybody know why the sleep is needed? Is that a threaded WFS consequence?
sleep(1)
online_layer = list(self.registry.mapLayers().values())[0]
self.assertTrue(online_layer.isValid())
self.assertFalse(online_layer.name().find('(offline)') > -1)
self.assertEqual(len([f for f in online_layer.getFeatures()]), len(TEST_FEATURES) - 1)
# Check that data have changed in the backend (raise exception if not found)
self.assertRaises(Exception, lambda: self._getFeatureByAttribute(online_layer, 'name', "'name 3'"))
# Check that all other features have not changed
layer = self._getLayer('test_point')
self.assertTrue(self._compareFeature(layer, TEST_FEATURES[1 - 1]))
self.assertTrue(self._compareFeature(layer, TEST_FEATURES[2 - 1]))
self.assertTrue(self._compareFeature(layer, TEST_FEATURES[4 - 1]))

def test_deleteMultipleFeatures(self):
"""
Delete a multiple features
"""
ol, offline_layer = self._testInit()
# Delete feature 1 and 3
feat1 = self._getFeatureByAttribute(offline_layer, 'name', "'name 1'")
feat3 = self._getFeatureByAttribute(offline_layer, 'name', "'name 3'")
self.assertTrue(offline_layer.startEditing())
self.assertTrue(offline_layer.deleteFeatures([feat3.id(), feat1.id()]))
self.assertTrue(offline_layer.commitChanges())
self.assertRaises(Exception, lambda: self._getFeatureByAttribute(offline_layer, 'name', "'name 3'"))
self.assertRaises(Exception, lambda: self._getFeatureByAttribute(offline_layer, 'name', "'name 1'"))
self.assertTrue(ol.isOfflineProject())
# Sync
ol.synchronize()
# Does anybody know why the sleep is needed? Is that a threaded WFS consequence?
sleep(1)
online_layer = list(self.registry.mapLayers().values())[0]
self.assertTrue(online_layer.isValid())
self.assertFalse(online_layer.name().find('(offline)') > -1)
self.assertEqual(len([f for f in online_layer.getFeatures()]), len(TEST_FEATURES) - 2)
# Check that data have changed in the backend (raise exception if not found)
self.assertRaises(Exception, lambda: self._getFeatureByAttribute(online_layer, 'name', "'name 3'"))
self.assertRaises(Exception, lambda: self._getFeatureByAttribute(online_layer, 'name', "'name 1'"))
# Check that all other features have not changed
layer = self._getLayer('test_point')
self.assertTrue(self._compareFeature(layer, TEST_FEATURES[2 - 1]))
self.assertTrue(self._compareFeature(layer, TEST_FEATURES[4 - 1]))

def test_InsertFeatures(self):
"""
Insert multiple features
"""
ol, offline_layer = self._testInit()
# Insert feature 5 and 6
self.assertTrue(offline_layer.startEditing())
features = []
for id, name, geom in TEST_FEATURES_INSERT:
f = QgsFeature(offline_layer.pendingFields())
f['id'] = id
f['name'] = name
f.setGeometry(QgsGeometry.fromPoint(geom))
features.append(f)
offline_layer.addFeatures(features)
self.assertTrue(offline_layer.commitChanges())
self._getFeatureByAttribute(offline_layer, 'name', "'name 5'")
self._getFeatureByAttribute(offline_layer, 'name', "'name 6'")
self.assertTrue(ol.isOfflineProject())
# Sync
ol.synchronize()
# Does anybody know why the sleep is needed? Is that a threaded WFS consequence?
sleep(1)
online_layer = list(self.registry.mapLayers().values())[0]
self.assertTrue(online_layer.isValid())
self.assertFalse(online_layer.name().find('(offline)') > -1)
self.assertEqual(len([f for f in online_layer.getFeatures()]), len(TEST_FEATURES) + 2)
# Check that data have changed in the backend (raise exception if not found)
self._getFeatureByAttribute(online_layer, 'name', "'name 5'")
self._getFeatureByAttribute(online_layer, 'name', "'name 6'")
# Check that all other features have not changed
layer = self._getLayer('test_point')
self.assertTrue(self._compareFeature(layer, TEST_FEATURES[1 - 1]))
self.assertTrue(self._compareFeature(layer, TEST_FEATURES[2 - 1]))
self.assertTrue(self._compareFeature(layer, TEST_FEATURES[3 - 1]))
self.assertTrue(self._compareFeature(layer, TEST_FEATURES[4 - 1]))
@@ -31,15 +31,26 @@
except KeyError:
QGIS_SERVER_DEFAULT_PORT = 8081

qgs_server = QgsServer()


class Handler(BaseHTTPRequestHandler):

def do_GET(self):
# CGI vars:
for k, v in self.headers.items():
qgs_server.putenv('HTTP_%s' % k.replace(' ', '-').replace('-', '_').replace(' ', '-').upper(), v)
qgs_server.putenv('SERVER_PORT', str(self.server.server_port))
qgs_server.putenv('SERVER_NAME', self.server.server_name)
qgs_server.putenv('REQUEST_URI', self.path)
parsed_path = urllib.parse.urlparse(self.path)
s = QgsServer()
headers, body = s.handleRequest(parsed_path.query)
self.send_response(200)
for k, v in [h.split(':') for h in headers.decode().split('\n') if h]:
headers, body = qgs_server.handleRequest(parsed_path.query)
headers_dict = dict(h.split(': ', 1) for h in headers.decode().split('\n') if h)
try:
self.send_response(int(headers_dict['Status'].split(' ')[0]))
except:
self.send_response(200)
for k, v in headers_dict.items():
self.send_header(k, v)
self.end_headers()
self.wfile.write(body)
@@ -36,10 +36,7 @@
import tempfile
from time import sleep
from utilities import unitTestDataPath
from qgis.core import (
QgsVectorLayer,
QgsProject,
)
from qgis.core import QgsVectorLayer

from qgis.testing import (
start_app,
@@ -48,19 +45,19 @@

from offlineditingtestbase import OfflineTestBase

from qgis.PyQt.QtCore import QFileInfo

try:
QGIS_SERVER_WFST_DEFAULT_PORT = os.environ['QGIS_SERVER_WFST_DEFAULT_PORT']
except:
QGIS_SERVER_WFST_DEFAULT_PORT = 8081


qgis_app = start_app()


class TestWFST(unittest.TestCase, OfflineTestBase):

# To fake the WFS cache!
counter = 0

@classmethod
def setUpClass(cls):
"""Run before all tests"""
@@ -82,47 +79,55 @@ def setUpClass(cls):
except KeyError:
pass
# Clear all test layers
cls._clearLayer('test_point')
cls._clearLayer(cls._getLayer('test_point'))
os.environ['QGIS_SERVER_DEFAULT_PORT'] = str(cls.port)
server_path = os.path.dirname(os.path.realpath(__file__)) + \
cls.server_path = os.path.dirname(os.path.realpath(__file__)) + \
'/qgis_wrapped_server.py'
cls.server = subprocess.Popen([sys.executable, server_path],
env=os.environ)
sleep(2)

@classmethod
def tearDownClass(cls):
"""Run after all tests"""
cls.server.terminate()
del cls.server
# Clear test layer
cls._clearLayer('test_point')
rmtree(cls.temp_path)

def setUp(self):
"""Run before each test."""
self.server = subprocess.Popen([sys.executable, self.server_path],
env=os.environ)
# Wait for the server process to start
sleep(2)
self._setUp()

def tearDown(self):
"""Run after each test."""
# Clear test layer
self._clearLayer(self._getOnlineLayer('test_point'))
# Kill the server
self.server.terminate()
del self.server
# Wait for the server process to stop
sleep(2)
# Delete the sqlite db
os.unlink(os.path.join(self.temp_path, 'offlineDbFile.sqlite'))
self._tearDown()

@classmethod
def _getOnlineLayer(cls, type_name, layer_name=None):
"""
Layer factory (return the online layer), provider specific
Return a new WFS layer, overriding the WFS cache
"""
if layer_name is None:
layer_name = 'wfs_' + type_name
parms = {
'srsname': 'EPSG:4326',
'typename': type_name,
'url': 'http://127.0.0.1:%s/?map=%s' % (cls.port,
cls.project_path),
'url': 'http://127.0.0.1:%s/%s/?map=%s' % (cls.port,
cls.counter,
cls.project_path),
'version': 'auto',
'table': '',
#'sql': '',
}
cls.counter += 1
uri = ' '.join([("%s='%s'" % (k, v)) for k, v in parms.items()])
wfs_layer = QgsVectorLayer(uri, layer_name, 'WFS')
assert wfs_layer.isValid()

0 comments on commit e59fda2

Please sign in to comment.
You can’t perform that action at this time.