Skip to content
Permalink
Browse files
Merge pull request #43788 from MorriganR/postgresTestWithRemoteDbV2
Additional test for postgres (ver.2)
  • Loading branch information
m-kuhn committed Jul 8, 2021
2 parents 754328c + b7ef263 commit f3da9993f044c6edce0a2edafe570198ff330402
@@ -20,3 +20,5 @@ services:
- postgres
env_file:
- docker-variables.env
cap_add:
- NET_ADMIN
@@ -125,6 +125,7 @@ FROM binary-for-oracle as binary-only
RUN apt-get update \
&& DEBIAN_FRONTEND=noninteractive apt-get install -y \
grass \
iproute2 \
postgresql-client \
spawn-fcgi \
&& pip3 install \
@@ -382,6 +382,7 @@ if (ENABLE_PGTEST)
ADD_PYTHON_TEST(PyQgsVectorFileWriterPostgres test_qgsvectorfilewriter_postgres.py)
ADD_PYTHON_TEST(PyQgsVectorLayerUtilsPostgres test_qgsvectorlayerutils_postgres.py)
ADD_PYTHON_TEST(PyQgsPostgresProvider test_provider_postgres.py)
ADD_PYTHON_TEST(PyQgsPostgresProviderLatency test_provider_postgres_latency.py TEST_TIMEOUT=600)
ADD_PYTHON_TEST(PyQgsPostgresRasterProvider test_provider_postgresraster.py)
ADD_PYTHON_TEST(PyQgsPostgresDomain test_qgspostgresdomain.py)
ADD_PYTHON_TEST(PyQgsPostgresTransaction test_qgspostgrestransaction.py)
@@ -410,7 +411,7 @@ if (ENABLE_PGTEST)
PyQgsRelationEditWidget PyQgsRelationPostgres PyQgsVectorLayerTools PyQgsProjectStoragePostgres
PyQgsAuthManagerPKIPostgresTest PyQgsAuthManagerPasswordPostgresTest PyQgsAuthManagerOgrPostgresTest
PyQgsDbManagerPostgis PyQgsDatabaseSchemaModel PyQgsDatabaseTableModel PyQgsDatabaseSchemaComboBox PyQgsDatabaseTableComboBox
PyQgsProviderConnectionPostgres
PyQgsProviderConnectionPostgres PyQgsPostgresProviderLatency
PROPERTIES LABELS "POSTGRES")
endif()

@@ -0,0 +1,134 @@
# -*- coding: utf-8 -*-
"""QGIS Unit tests for the remote PostgreSQL server.
Note: The test makes sense if the network latency
between QGIS and the PG server is 100ms.
Run with ctest -V -R PyQgsPostgresProviderLatency
.. 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
(at your option) any later version.
"""
from builtins import next

__author__ = 'Daryna Dyka'
__date__ = '2021-06-13'
__copyright__ = 'Copyright 2021, The QGIS Project'

import qgis # NOQA
import psycopg2

import os
import time

from qgis.core import QgsVectorLayer, QgsFeatureRequest
from qgis.testing import start_app, unittest

QGISAPP = start_app()


class TestPyQgsPostgresProviderLatency(unittest.TestCase):

@classmethod
def setUpClass(cls):
"""Run before all tests"""
cls.dbconn = 'service=qgis_test'
if 'QGIS_PGTEST_DB' in os.environ:
cls.dbconn = os.environ['QGIS_PGTEST_DB']
cls.con = psycopg2.connect(cls.dbconn)

@classmethod
def tearDownClass(cls):
"""Run after all tests"""
os.system('tc qdisc del dev eth0 root')

def setDelay(self, delay_in_ms):
if delay_in_ms == 0:
os.system('tc qdisc del dev eth0 root')
else:
os.system('tc qdisc add dev eth0 root netem delay {}ms'.format(delay_in_ms))

def getDelay(self):
self.assertTrue(self.con)
cur = self.con.cursor()
self.assertTrue(cur)
start_time = time.time()
cur.execute('SELECT 1;')
duration = round(1000 * abs(time.time() - start_time), 1)
cur.close()
self.con.commit()
return duration

def execSQLCommand(self, sql):
self.assertTrue(self.con)
cur = self.con.cursor()
self.assertTrue(cur)
cur.execute(sql)
cur.close()
self.con.commit()

@unittest.skip("Skipping")
def testStatsSetDelayToDB(self):
"""Test Set delay to remote DB"""

delay_set = [0] * 100
delay_get = [0] * 100
delay_delta = [0] * 100
for delay_ms in range(1, 100, 1):
delay_set[delay_ms] = delay_ms
self.setDelay(delay_ms)
delay_get[delay_ms] = self.getDelay()
self.setDelay(0)
delay_delta[delay_ms] = round(delay_get[delay_ms] / 2.0 - delay_set[delay_ms], 1)

self.assertTrue(False, '\nset: {}\nget: {}\nget/2 - set: {}'.format(delay_set, delay_get, delay_delta))

def testSetDelayToDB(self):
"""Test Set delay to remote DB"""

self.setDelay(100)
delay_get = self.getDelay() / 2
self.setDelay(0)
self.assertTrue(delay_get > 90 and delay_get < 110, 'set delay to 100ms - unsuccessful (got: {}ms)'.format(delay_get))

def testSaveChangedGeometryToDB(self):
"""Test Save geometries to remote DB"""

self.execSQLCommand('''
DROP TABLE IF EXISTS qgis_test.speed_test_remote_db CASCADE;
CREATE TABLE qgis_test.speed_test_remote_db (
id_serial serial NOT NULL,
geom geometry(Polygon,3857),
CONSTRAINT speed_test_remote_db_pk PRIMARY KEY (id_serial) );
INSERT INTO qgis_test.speed_test_remote_db(geom)
SELECT
ST_Translate(
ST_GeomFromText(\'POLYGON((3396900.0 6521800.0,3396900.0 6521870.0,
3396830.0 6521870.0,3396830.0 6521800.0,3396900.0 6521800.0))\', 3857 ),
100.0 * dx,
100.0 * dy )
FROM generate_series(1,42) dx, generate_series(1,42) dy;''')

set_new_layer = ' sslmode=disable key=\'id_serial\' srid=3857 type=POLYGON table="qgis_test"."speed_test_remote_db" (geom) sql='
error_string = 'Save geoms to remote DB : expected < 10s, got {}s'

vl = QgsVectorLayer(self.dbconn + set_new_layer, 'test_vl_remote_save', 'postgres')
# fids = [f.id() for f in vl.getFeatures(QgsFeatureRequest().setLimit(1000))]
fids = [f.id() for f in vl.getFeatures(QgsFeatureRequest().setLimit(50))]
self.assertTrue(vl.startEditing())
for f in vl.getFeatures(fids):
self.assertTrue(vl.changeGeometry(f.id(), f.geometry()))

self.setDelay(100)
start_time = time.time()
self.assertTrue(vl.commitChanges())
duration = round(abs(time.time() - start_time), 1)
self.setDelay(0)
self.assertTrue(duration < 10, error_string.format(duration))


if __name__ == '__main__':
unittest.main()

0 comments on commit f3da999

Please sign in to comment.