Skip to content
Permalink
Browse files
Backport of the new test pick a free port strategy
  • Loading branch information
elpaso committed Oct 4, 2016
1 parent a6bb8a9 commit fd77ee2ed94056de021e7d53d79fa6b79a4b7dc0
@@ -62,7 +62,7 @@ def _setUp(self):
def _tearDown(self):
"""Called by tearDown: run after each test."""
# Clear test layers
self._clearLayer('test_point')
self._clearLayer(self._getLayer('test_point'))

@classmethod
def _compareFeature(cls, layer, attributes):
@@ -71,11 +71,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
"""
layer = cls._getLayer(layer_name)
layer.startEditing()
layer.deleteFeatures([f.id() for f in layer.getFeatures()])
layer.commitChanges()
@@ -3,7 +3,8 @@
QGIS Server HTTP wrapper
This script launches a QGIS Server listening on port 8081 or on the port
specified on the environment variable QGIS_SERVER_DEFAULT_PORT
specified on the environment variable QGIS_SERVER_PORT.
QGIS_SERVER_HOST (defaults to 127.0.0.1)
For testing purposes, HTTP Basic can be enabled by setting the following
environment variables:
@@ -29,19 +30,17 @@


import os
import sys
import urllib.parse
from http.server import BaseHTTPRequestHandler, HTTPServer
from qgis.server import QgsServer, QgsServerFilter

try:
QGIS_SERVER_DEFAULT_PORT = int(os.environ['QGIS_SERVER_DEFAULT_PORT'])
except KeyError:
QGIS_SERVER_DEFAULT_PORT = 8081
QGIS_SERVER_PORT = int(os.environ.get('QGIS_SERVER_PORT', '8081'))
QGIS_SERVER_HOST = os.environ.get('QGIS_SERVER_HOST', '127.0.0.1')

qgs_server = QgsServer()

if os.environ.get('QGIS_SERVER_HTTP_BASIC_AUTH') is not None:
print('HTTP Basic Authorization Required username:%s password:%s' % (os.environ.get('QGIS_SERVER_USERNAME', 'username'), os.environ.get('QGIS_SERVER_PASSWORD', 'password')))
import base64

class HTTPBasicFilter(QgsServerFilter):
@@ -96,7 +95,8 @@ def do_POST(self):


if __name__ == '__main__':
server = HTTPServer(('localhost', QGIS_SERVER_DEFAULT_PORT), Handler)
print('Starting server on localhost:%s, use <Ctrl-C> to stop' %
QGIS_SERVER_DEFAULT_PORT)
server = HTTPServer((QGIS_SERVER_HOST, QGIS_SERVER_PORT), Handler)
print('Starting server on %s:%s, use <Ctrl-C> to stop' %
(QGIS_SERVER_HOST, server.server_port))
sys.stdout.flush()
server.serve_forever()
@@ -17,6 +17,7 @@
"""
import os
import sys
import re
import subprocess
import tempfile
import random
@@ -29,10 +30,9 @@
# This will get replaced with a git SHA1 when you do a git archive
__revision__ = '$Format:%H$'

from time import sleep
from shutil import rmtree

from utilities import unitTestDataPath
from utilities import unitTestDataPath, waitServer
from qgis.core import (
QgsAuthManager,
QgsAuthMethodConfig,
@@ -45,13 +45,10 @@
)

try:
QGIS_SERVER_AUTHMANAGER_DEFAULT_PORT = os.environ['QGIS_SERVER_AUTHMANAGER_DEFAULT_PORT']
QGIS_SERVER_ENDPOINT_PORT = os.environ['QGIS_SERVER_ENDPOINT_PORT']
except:
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("", 0))
QGIS_SERVER_AUTHMANAGER_DEFAULT_PORT = s.getsockname()[1]
s.close()
QGIS_SERVER_ENDPOINT_PORT = '0' # Auto


QGIS_AUTH_DB_DIR_PATH = tempfile.mkdtemp()

@@ -66,7 +63,7 @@ class TestAuthManager(unittest.TestCase):
def setUpClass(cls):
"""Run before all tests:
Creates an auth configuration"""
cls.port = QGIS_SERVER_AUTHMANAGER_DEFAULT_PORT
cls.port = QGIS_SERVER_ENDPOINT_PORT
# Clean env just to be sure
env_vars = ['QUERY_STRING', 'QGIS_PROJECT_FILE']
for ev in env_vars:
@@ -91,12 +88,17 @@ def setUpClass(cls):
os.environ['QGIS_SERVER_HTTP_BASIC_AUTH'] = '1'
os.environ['QGIS_SERVER_USERNAME'] = cls.username
os.environ['QGIS_SERVER_PASSWORD'] = cls.password
os.environ['QGIS_SERVER_DEFAULT_PORT'] = str(cls.port)
os.environ['QGIS_SERVER_PORT'] = str(cls.port)
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)
env=os.environ, stdout=subprocess.PIPE)

line = cls.server.stdout.readline()
cls.port = int(re.findall(b':(\d+)', line)[0])
assert cls.port != 0
# Wait for the server process to start
assert waitServer('http://127.0.0.1:%s' % cls.port), "Server is not responding! http://127.0.0.1:%s" % cls.port

@classmethod
def tearDownClass(cls):
@@ -1,20 +1,14 @@
# -*- coding: utf-8 -*-
"""
Offline editing Tests.
WFS-T tests need using QGIS Server through
qgis_wrapped_server.py.
This is an integration test for QGIS Desktop WFS-T provider and QGIS Server
WFS-T that check if QGIS offline editing works with a WFS-T endpoint.
The test uses testdata/wfs_transactional/wfs_transactional.qgs and three
initially empty shapefiles layers with points, lines and polygons.
The point layer is used in the test
From build dir, run: ctest -R PyQgsOfflineEditingWFS -V
.. 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
@@ -31,15 +25,13 @@

import os
import sys
import re
import subprocess
from shutil import copytree, rmtree
import tempfile
from time import sleep
from utilities import unitTestDataPath
from qgis.core import (
QgsVectorLayer,
QgsProject,
)
from utilities import unitTestDataPath, waitServer
from qgis.core import QgsVectorLayer

from qgis.testing import (
start_app,
@@ -48,27 +40,24 @@

from offlineditingtestbase import OfflineTestBase

from qgis.PyQt.QtCore import QFileInfo

try:
QGIS_SERVER_OFFLINE_EDITING_DEFAULT_PORT = os.environ['QGIS_SERVER_OFFLINE_EDITING_DEFAULT_PORT']
QGIS_SERVER_OFFLINE_PORT = os.environ['QGIS_SERVER_OFFLINE_PORT']
except:
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("", 0))
QGIS_SERVER_OFFLINE_EDITING_DEFAULT_PORT = s.getsockname()[1]
s.close()

QGIS_SERVER_OFFLINE_PORT = '0' # Auto

qgis_app = start_app()


class TestWFST(unittest.TestCase, OfflineTestBase):

# To fake the WFS cache!
counter = 0

@classmethod
def setUpClass(cls):
"""Run before all tests"""
cls.port = QGIS_SERVER_OFFLINE_EDITING_DEFAULT_PORT
cls.port = QGIS_SERVER_OFFLINE_PORT
# Create tmp folder
cls.temp_path = tempfile.mkdtemp()
cls.testdata_path = cls.temp_path + '/' + 'wfs_transactional' + '/'
@@ -86,48 +75,57 @@ def setUpClass(cls):
except KeyError:
pass
# Clear all test layers
cls._clearLayer('test_point')
os.environ['QGIS_SERVER_DEFAULT_PORT'] = str(cls.port)
server_path = os.path.dirname(os.path.realpath(__file__)) + \
cls._clearLayer(cls._getLayer('test_point'))
os.environ['QGIS_SERVER_PORT'] = str(cls.port)
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, stdout=subprocess.PIPE)
line = self.server.stdout.readline()
self.port = int(re.findall(b':(\d+)', line)[0])
assert self.port != 0
# Wait for the server process to start
assert waitServer('http://127.0.0.1:%s' % self.port), "Server is not responding!"
self._setUp()

def tearDown(self):
"""Run after each test."""
# Clear test layer
self._clearLayer(self._getOnlineLayer('test_point'))
# Kill the server
self.server.terminate()
self.server.wait()
del self.server
# 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):
def _getOnlineLayer(self, 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' % (self.port,
self.counter,
self.project_path),
'version': 'auto',
'table': '',
#'sql': '',
}
uri = ' '.join([("%s='%s'" % (k, v)) for k, v in parms.items()])
self.counter += 1
uri = ' '.join([("%s='%s'" % (k, v)) for k, v in list(parms.items())])
wfs_layer = QgsVectorLayer(uri, layer_name, 'WFS')
assert wfs_layer.isValid()
return wfs_layer
@@ -1,3 +1,4 @@

# -*- coding: utf-8 -*-
"""
Tests for WFS-T provider using QGIS Server through qgis_wrapped_server.py.
@@ -36,11 +37,12 @@

import os
import sys
import re
import subprocess
from shutil import copytree, rmtree
import tempfile
from time import sleep
from utilities import unitTestDataPath
from utilities import unitTestDataPath, waitServer
from qgis.core import (
QgsVectorLayer,
QgsFeature,
@@ -56,13 +58,9 @@
)

try:
QGIS_SERVER_WFST_DEFAULT_PORT = os.environ['QGIS_SERVER_WFST_DEFAULT_PORT']
QGIS_SERVER_WFST_PORT = os.environ['QGIS_SERVER_WFST_PORT']
except:
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("", 0))
QGIS_SERVER_WFST_DEFAULT_PORT = s.getsockname()[1]
s.close()
QGIS_SERVER_WFST_PORT = '0' # Auto


qgis_app = start_app()
@@ -73,7 +71,7 @@ class TestWFST(unittest.TestCase):
@classmethod
def setUpClass(cls):
"""Run before all tests"""
cls.port = QGIS_SERVER_WFST_DEFAULT_PORT
cls.port = QGIS_SERVER_WFST_PORT
# Create tmp folder
cls.temp_path = tempfile.mkdtemp()
cls.testdata_path = cls.temp_path + '/' + 'wfs_transactional' + '/'
@@ -93,17 +91,22 @@ def setUpClass(cls):
# Clear all test layers
for ln in ['test_point', 'test_polygon', 'test_linestring']:
cls._clearLayer(ln)
os.environ['QGIS_SERVER_DEFAULT_PORT'] = str(cls.port)
os.environ['QGIS_SERVER_PORT'] = str(cls.port)
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)
env=os.environ, stdout=subprocess.PIPE)
line = cls.server.stdout.readline()
cls.port = int(re.findall(b':(\d+)', line)[0])
assert cls.port != 0
# Wait for the server process to start
assert waitServer('http://127.0.0.1:%s' % cls.port), "Server is not responding! http://127.0.0.1:%s" % cls.port

@classmethod
def tearDownClass(cls):
"""Run after all tests"""
cls.server.terminate()
cls.server.wait()
del cls.server
# Clear all test layers
for ln in ['test_point', 'test_polygon', 'test_linestring']:
@@ -156,7 +159,7 @@ def _getWFSLayer(cls, type_name, layer_name=None):
'table': '',
#'sql': '',
}
uri = ' '.join([("%s='%s'" % (k, v)) for k, v in parms.items()])
uri = ' '.join([("%s='%s'" % (k, v)) for k, v in list(parms.items())])
wfs_layer = QgsVectorLayer(uri, layer_name, 'WFS')
assert wfs_layer.isValid()
return wfs_layer
@@ -18,6 +18,7 @@
import glob
import platform
import tempfile
from urllib2 import urlopen, HTTPError

from qgis.PyQt.QtCore import QDir

@@ -815,3 +816,22 @@ def memberIsDocumented(self, member_elem):
if doc is not None and list(doc):
return True
return False


def waitServer(url, timeout=10):
""" Wait for a server to be online and to respond
HTTP errors are ignored
@param timeout: in seconds
@return: True of False
"""
from time import time as now
end = now() + timeout
while True:
try:
urlopen(url, timeout=1)
return True
except HTTPError:
return True
except Exception as e:
if now() > end:
return False

0 comments on commit fd77ee2

Please sign in to comment.