Skip to content

Commit

Permalink
Merge pull request #3553 from elpaso/server-tests-backport
Browse files Browse the repository at this point in the history
[Server] Backport of the test improved stability
  • Loading branch information
elpaso committed Sep 30, 2016
2 parents a8fbf8a + 02dfb4f commit 1737e47
Show file tree
Hide file tree
Showing 13 changed files with 424 additions and 134 deletions.
1 change: 1 addition & 0 deletions ci/travis/linux/qt5/blacklist.txt
Expand Up @@ -6,6 +6,7 @@ PyQgsMapUnitScale
PyQgsPalLabelingServer
PyQgsRelationEditWidget
PyQgsServer
PyQgsAuthManagerEndpointTest
PyQgsServerAccessControl
PyQgsSipCoverage
PyQgsSpatialiteProvider
Expand Down
1 change: 1 addition & 0 deletions tests/src/python/CMakeLists.txt
Expand Up @@ -147,4 +147,5 @@ IF (WITH_SERVER)
ADD_PYTHON_TEST(PyQgsServerAccessControl test_qgsserver_accesscontrol.py)
ADD_PYTHON_TEST(PyQgsServerWFST test_qgsserver_wfst.py)
ADD_PYTHON_TEST(PyQgsOfflineEditingWFS test_offline_editing_wfs.py)
ADD_PYTHON_TEST(PyQgsAuthManagerEndpointTest test_authmanager_endpoint.py)
ENDIF (WITH_SERVER)
51 changes: 46 additions & 5 deletions tests/src/python/qgis_wrapped_server.py
Expand Up @@ -5,6 +5,13 @@
This script launches a QGIS Server listening on port 8081 or on the port
specified on the environment variable QGIS_SERVER_DEFAULT_PORT
For testing purposes, HTTP Basic can be enabled by setting the following
environment variables:
* QGIS_SERVER_HTTP_BASIC_AUTH (default not set, set to anything to enable)
* QGIS_SERVER_USERNAME (default ="username")
* QGIS_SERVER_PASSWORD (default ="password")
.. 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
Expand All @@ -24,22 +31,56 @@
import os
import urllib.parse
from http.server import BaseHTTPRequestHandler, HTTPServer
from qgis.server import QgsServer
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

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):

def responseComplete(self):
request = self.serverInterface().requestHandler()
if self.serverInterface().getEnv('HTTP_AUTHORIZATION'):
username, password = base64.b64decode(self.serverInterface().getEnv('HTTP_AUTHORIZATION')[6:]).split(':')
if (username == os.environ.get('QGIS_SERVER_USERNAME', 'username')
and password == os.environ.get('QGIS_SERVER_PASSWORD', 'password')):
return
# No auth ...
request.clearHeaders()
request.setHeader('Status', '401 Authorization required')
request.setHeader('WWW-Authenticate', 'Basic realm="QGIS Server"')
request.clearBody()
request.appendBody('<h1>Authorization required</h1>')

filter = HTTPBasicFilter(qgs_server.serverInterface())
qgs_server.serverInterface().registerFilter(filter)


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)
Expand Down
180 changes: 180 additions & 0 deletions tests/src/python/test_authmanager_endpoint.py
@@ -0,0 +1,180 @@
# -*- coding: utf-8 -*-
"""
Tests for auth manager WMS/WFS using QGIS Server through HTTP Basic
enabled qgis_wrapped_server.py.
This is an integration test for QGIS Desktop Auth Manager WFS and WMS provider
and QGIS Server WFS/WMS that check if QGIS can use a stored auth manager auth
configuration to access an HTTP Basic protected endpoint.
From build dir, run: ctest -R PyQgsAuthManagerEnpointTest -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
(at your option) any later version.
"""
import os
import sys
import subprocess
import tempfile
import random
import string
import urllib

__author__ = 'Alessandro Pasotti'
__date__ = '18/09/2016'
__copyright__ = 'Copyright 2016, The QGIS Project'
# 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 qgis.core import (
QgsAuthManager,
QgsAuthMethodConfig,
QgsVectorLayer,
QgsRasterLayer,
)
from qgis.testing import (
start_app,
unittest,
)

try:
QGIS_SERVER_AUTHMANAGER_DEFAULT_PORT = os.environ['QGIS_SERVER_AUTHMANAGER_DEFAULT_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_AUTH_DB_DIR_PATH = tempfile.mkdtemp()

os.environ['QGIS_AUTH_DB_DIR_PATH'] = QGIS_AUTH_DB_DIR_PATH

qgis_app = start_app()


class TestAuthManager(unittest.TestCase):

@classmethod
def setUpClass(cls):
"""Run before all tests:
Creates an auth configuration"""
cls.port = QGIS_SERVER_AUTHMANAGER_DEFAULT_PORT
# Clean env just to be sure
env_vars = ['QUERY_STRING', 'QGIS_PROJECT_FILE']
for ev in env_vars:
try:
del os.environ[ev]
except KeyError:
pass
cls.testdata_path = unitTestDataPath('qgis_server') + '/'
cls.project_path = cls.testdata_path + "test_project.qgs"
# Enable auth
#os.environ['QGIS_AUTH_PASSWORD_FILE'] = QGIS_AUTH_PASSWORD_FILE
authm = QgsAuthManager.instance()
assert (authm.setMasterPassword('masterpassword', True))
cls.auth_config = QgsAuthMethodConfig('Basic')
cls.auth_config.setName('test_auth_config')
cls.username = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(6))
cls.password = cls.username[::-1] # reversed
cls.auth_config.setConfig('username', cls.username)
cls.auth_config.setConfig('password', cls.password)
assert (authm.storeAuthenticationConfig(cls.auth_config)[0])

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)
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()
rmtree(QGIS_AUTH_DB_DIR_PATH)
del cls.server

def setUp(self):
"""Run before each test."""
pass

def tearDown(self):
"""Run after each test."""
pass

@classmethod
def _getWFSLayer(cls, type_name, layer_name=None, authcfg=None):
"""
WFS layer factory
"""
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),
'version': 'auto',
'table': '',
}
if authcfg is not None:
parms.update({'authcfg': authcfg})
uri = ' '.join([("%s='%s'" % (k, v.decode('utf-8'))) for k, v in list(parms.items())])
wfs_layer = QgsVectorLayer(uri, layer_name, 'WFS')
return wfs_layer

@classmethod
def _getWMSLayer(cls, layers, layer_name=None, authcfg=None):
"""
WMS layer factory
"""
if layer_name is None:
layer_name = 'wms_' + layers.replace(',', '')
parms = {
'crs': 'EPSG:4326',
'url': 'http://127.0.0.1:%s/?map=%s' % (cls.port, cls.project_path),
'format': 'image/png',
# This is needed because of a really wierd implementation in QGIS Server, that
# replaces _ in the the real layer name with spaces
'layers': urllib.quote(layers.replace('_', ' ')),
'styles': '',
'version': 'auto',
#'sql': '',
}
if authcfg is not None:
parms.update({'authcfg': authcfg})
uri = '&'.join([("%s=%s" % (k, v.replace('=', '%3D'))) for k, v in list(parms.items())])
wms_layer = QgsRasterLayer(uri, layer_name, 'wms')
return wms_layer

def testValidAuthAccess(self):
"""
Access the HTTP Basic protected layer with valid credentials
"""
wfs_layer = self._getWFSLayer('testlayer_èé', authcfg=self.auth_config.id())
self.assertTrue(wfs_layer.isValid())
wms_layer = self._getWMSLayer('testlayer_èé', authcfg=self.auth_config.id())
self.assertTrue(wms_layer.isValid())

def testInvalidAuthAccess(self):
"""
Access the HTTP Basic protected layer with no credentials
"""
wfs_layer = self._getWFSLayer('testlayer èé')
self.assertFalse(wfs_layer.isValid())
wms_layer = self._getWMSLayer('testlayer_èé')
self.assertFalse(wms_layer.isValid())


if __name__ == '__main__':
unittest.main()
10 changes: 7 additions & 3 deletions tests/src/python/test_offline_editing_wfs.py
Expand Up @@ -51,9 +51,13 @@
from qgis.PyQt.QtCore import QFileInfo

try:
QGIS_SERVER_WFST_DEFAULT_PORT = os.environ['QGIS_SERVER_WFST_DEFAULT_PORT']
QGIS_SERVER_OFFLINE_EDITING_DEFAULT_PORT = os.environ['QGIS_SERVER_OFFLINE_EDITING_DEFAULT_PORT']
except:
QGIS_SERVER_WFST_DEFAULT_PORT = 8081
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_app = start_app()
Expand All @@ -64,7 +68,7 @@ class TestWFST(unittest.TestCase, OfflineTestBase):
@classmethod
def setUpClass(cls):
"""Run before all tests"""
cls.port = QGIS_SERVER_WFST_DEFAULT_PORT
cls.port = QGIS_SERVER_OFFLINE_EDITING_DEFAULT_PORT
# Create tmp folder
cls.temp_path = tempfile.mkdtemp()
cls.testdata_path = cls.temp_path + '/' + 'wfs_transactional' + '/'
Expand Down
27 changes: 16 additions & 11 deletions tests/src/python/test_qgsserver.py
Expand Up @@ -25,7 +25,7 @@
import osgeo.gdal

# Strip path and content length because path may vary
RE_STRIP_PATH = r'MAP=[^&]+|Content-Length: \d+'
RE_STRIP_PATH = r'MAP=[^&]+|Content-Length: \d+|<Attribute typeName="[^>]+'


class TestQgsServer(unittest.TestCase):
Expand Down Expand Up @@ -150,28 +150,33 @@ def responseComplete(self):

# WMS tests
def wms_request_compare(self, request, extra=None, reference_file=None):
project = self.testdata_path + "test+project.qgs"
project = self.testdata_path + "test_project.qgs"
assert os.path.exists(project), "Project file not found: " + project

query_string = 'MAP=%s&SERVICE=WMS&VERSION=1.3&REQUEST=%s' % (urllib.quote(project), request)
if extra is not None:
query_string += extra
header, body = [str(_v) for _v in self.server.handleRequest(query_string)]
response = header + body
f = open(self.testdata_path + (request.lower() if not reference_file else reference_file) + '.txt')
reference_path = self.testdata_path + (request.lower() if not reference_file else reference_file) + '.txt'
f = open(reference_path)
expected = f.read()
f.close()
# Store the output for debug or to regenerate the reference documents:
"""
f = open(reference_path, 'wb+')
f.write(response)
f.close()
f = open(os.path.dirname(__file__) + '/expected.txt', 'w+')
f.write(expected)
f.close()
f = open(os.path.dirname(__file__) + '/response.txt', 'w+')
f.write(response)
f.close()
#"""
response = re.sub(RE_STRIP_PATH, '', response)
expected = re.sub(RE_STRIP_PATH, '', expected)
"""
response = re.sub(RE_STRIP_PATH, '*****', response)
expected = re.sub(RE_STRIP_PATH, '*****', expected)

# for older GDAL versions (<2.0), id field will be integer type
if int(osgeo.gdal.VersionInfo()[:1]) < 2:
Expand Down Expand Up @@ -214,7 +219,7 @@ def test_project_wms(self):

def wms_inspire_request_compare(self, request):
"""WMS INSPIRE tests"""
project = self.testdata_path + "test+project_inspire.qgs"
project = self.testdata_path + "test_project_inspire.qgs"
assert os.path.exists(project), "Project file not found: " + project

query_string = 'MAP=%s&SERVICE=WMS&VERSION=1.3.0&REQUEST=%s' % (urllib.quote(project), request)
Expand Down Expand Up @@ -243,7 +248,7 @@ def test_project_wms_inspire(self):

# WFS tests
def wfs_request_compare(self, request):
project = self.testdata_path + "test+project_wfs.qgs"
project = self.testdata_path + "test_project_wfs.qgs"
assert os.path.exists(project), "Project file not found: " + project

query_string = 'MAP=%s&SERVICE=WFS&VERSION=1.0.0&REQUEST=%s' % (urllib.quote(project), request)
Expand Down Expand Up @@ -277,7 +282,7 @@ def test_project_wfs(self):
self.wfs_request_compare(request)

def wfs_getfeature_compare(self, requestid, request):
project = self.testdata_path + "test+project_wfs.qgs"
project = self.testdata_path + "test_project_wfs.qgs"
assert os.path.exists(project), "Project file not found: " + project

query_string = 'MAP=%s&SERVICE=WFS&VERSION=1.0.0&REQUEST=%s' % (urllib.quote(project), request)
Expand Down Expand Up @@ -324,7 +329,7 @@ def test_getfeature(self):
self.wfs_getfeature_compare(id, req)

def wfs_getfeature_post_compare(self, requestid, request):
project = self.testdata_path + "test+project_wfs.qgs"
project = self.testdata_path + "test_project_wfs.qgs"
assert os.path.exists(project), "Project file not found: " + project

query_string = 'MAP={}'.format(urllib.quote(project))
Expand Down Expand Up @@ -369,7 +374,7 @@ def test_getfeature_post(self):
def test_getLegendGraphics(self):
"""Test that does not return an exception but an image"""
parms = {
'MAP': self.testdata_path + "test%2Bproject.qgs",
'MAP': self.testdata_path + "test_project.qgs",
'SERVICE': 'WMS',
'VERSION': '1.0.0',
'REQUEST': 'GetLegendGraphic',
Expand Down

0 comments on commit 1737e47

Please sign in to comment.