sharpetronics/QGIS
1
0
Fork 0
mirror of https://github.com/qgis/QGIS.git synced 2025-04-13 00:03:09 -04:00
Code Issues Packages Projects Releases Wiki Activity

[server] Python plugins API cleanup part 3

Browse Source 
This removes the handleRequest method that returns the
headers and body as byte array.

This superceeded by the implementation that takes a
request and response instances.
This commit is contained in:
Alessandro Pasotti 2017-04-30 22:21:18 +02:00
parent 2afcad279d
commit 569593b79a
10 changed files with 106 additions and 324 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
python/server/qgsbufferserverrequest.sip
View File

@ -1,7 +1,7 @@
/************************************************************************
* This file has been generated automatically from *
* *
* ../src/server/qgsbufferserverrequest.h *
* src/server/qgsbufferserverrequest.h *
* *
* Do not edit manually ! Edit header and run scripts/sipify.pl again *
************************************************************************/
@ -46,7 +46,7 @@ class QgsBufferServerRequest : QgsServerRequest
/************************************************************************
* This file has been generated automatically from *
* *
* ../src/server/qgsbufferserverrequest.h *
* src/server/qgsbufferserverrequest.h *
* *
* Do not edit manually ! Edit header and run scripts/sipify.pl again *
************************************************************************/

6
python/server/qgsbufferserverresponse.sip
View File

@ -1,7 +1,7 @@
/************************************************************************
* This file has been generated automatically from *
* *
* ../src/server/qgsbufferserverresponse.h *
* src/server/qgsbufferserverresponse.h *
* *
* Do not edit manually ! Edit header and run scripts/sipify.pl again *
************************************************************************/
@ -141,13 +141,13 @@ class QgsBufferServerResponse: QgsServerResponse
private:
QgsBufferServerResponse(const QgsBufferServerResponse &) ;
QgsBufferServerResponse( const QgsBufferServerResponse & ) ;
};
/************************************************************************
* This file has been generated automatically from *
* *
* ../src/server/qgsbufferserverresponse.h *
* src/server/qgsbufferserverresponse.h *
* *
* Do not edit manually ! Edit header and run scripts/sipify.pl again *
************************************************************************/

147
python/server/qgsserver.sip
View File

@ -17,141 +17,6 @@
***************************************************************************/
%MappedType QPair<QByteArray, QByteArray>
{
%TypeHeaderCode
#include <QPair>
#include <QByteArray>
%End
%TypeCode
// Convenience function for converting a QByteArray to a Python str object. (from QtCore/qbytearray.sip)
static PyObject *QByteArrayToPyStr(QByteArray *ba)
{
char *data = ba->data();
if (data)
// QByteArrays may have embedded '\0's so set the size explicitly.
return SIPBytes_FromStringAndSize(data, ba->size());
return SIPBytes_FromString("");
}
%End
%ConvertFromTypeCode
// Create the tuple.
return Py_BuildValue((char *)"OO", QByteArrayToPyStr( &sipCpp->first ), QByteArrayToPyStr( &sipCpp->second ) );
%End
%ConvertToTypeCode
// See if we are just being asked to check the type of the Python
// object.
if (!sipIsErr)
{
// Checking whether or not None has been passed instead of a list
// has already been done.
if (!PyTuple_Check(sipPy) || PyTuple_Size(sipPy) != 2)
return 0;
// Check the type of each element. We specify SIP_NOT_NONE to
// disallow None because it is a list of QPoint, not of a pointer
// to a QPoint, so None isn't appropriate.
for (int i = 0; i < PyTuple_Size(sipPy); ++i)
if (!sipCanConvertToType(PyTuple_GET_ITEM(sipPy, i),
sipType_QByteArray, SIP_NOT_NONE))
return 0;
// The type is valid.
return 1;
}
// Create the instance on the heap.
QPair<QByteArray, QByteArray> *qp = new QPair<QByteArray, QByteArray>;
QByteArray *qba1;
int state;
// Get the address of the element's C++ instance. Note that, in
// this case, we don't apply any ownership changes to the list
// elements, only to the list itself.
qba1 = reinterpret_cast<QByteArray *>(sipConvertToType(
PyTuple_GET_ITEM(sipPy, 0),
sipType_QByteArray, 0,
SIP_NOT_NONE,
&state, sipIsErr));
// Deal with any errors.
if (*sipIsErr)
{
sipReleaseType(qba1, sipType_QByteArray, state);
// Tidy up.
delete qp;
// There is no temporary instance.
return 0;
}
qp->first = *qba1;
// A copy of the QByteArray was assigned to the pair so we no longer
// need it. It may be a temporary instance that should be
// destroyed, or a wrapped instance that should not be destroyed.
// sipReleaseType() will do the right thing.
sipReleaseType(qba1, sipType_QByteArray, state);
/////////////////////////////////////////////
// Second item
QByteArray *qba2;
// Get the address of the element's C++ instance. Note that, in
// this case, we don't apply any ownership changes to the list
// elements, only to the list itself.
qba2 = reinterpret_cast<QByteArray *>(sipConvertToType(
PyTuple_GET_ITEM(sipPy, 1),
sipType_QByteArray, 0,
SIP_NOT_NONE,
&state, sipIsErr));
// Deal with any errors.
if (*sipIsErr)
{
sipReleaseType(qba1, sipType_QByteArray, state);
sipReleaseType(qba2, sipType_QByteArray, state);
// Tidy up.
delete qp;
// There is no temporary instance.
return 0;
}
qp->second = *qba2;
// A copy of the QByteArray was assigned to the pair so we no longer
// need it. It may be a temporary instance that should be
// destroyed, or a wrapped instance that should not be destroyed.
// sipReleaseType() will do the right thing.
sipReleaseType(qba2, sipType_QByteArray, state);
// Return the instance.
*sipCppPtr = qp;
// The instance should be regarded as temporary (and be destroyed as
// soon as it has been used) unless it has been transferred from
// Python. sipGetState() is a convenience function that implements
// this common transfer behavior.
return sipGetState(sipTransferObj);
%End
};
/** \ingroup server
* The QgsServer class provides OGC web services.
*/
@ -185,18 +50,6 @@ class QgsServer
*/
void handleRequest( QgsServerRequest &request, QgsServerResponse &response );
/** Handles the request from query string
* The query string is normally read from environment
* but can be also passed in args and in this case overrides the environment
* variable.
*
* \param urlstr QString containing the request url (simple quely string must be preceded by '?')
* \param requestMethod QgsServerRequest::Method that indicates the method. Only "GET" or "POST" are supported.
* \param data array of bytes containing post data
* \param map of request headers
* \returns the response headers and body QPair of QByteArray
*/
QPair<QByteArray, QByteArray> handleRequest( const QString &urlstr, const QgsServerRequest::Method requestMethod = QgsServerRequest::GetMethod, const QgsServerRequest::Headers &headers = QgsServerRequest::Headers(), const char *data = nullptr );
/** Returns a pointer to the server interface */
QgsServerInterface *serverInterface();

43
src/server/qgsserver.cpp
View File

@ -427,49 +427,6 @@ void QgsServer::handleRequest( QgsServerRequest &request, QgsServerResponse &res
}
}
QPair<QByteArray, QByteArray> QgsServer::handleRequest( const QString &urlstr, const QgsServerRequest::Method requestMethod, const QgsServerRequest::Headers &headers, const char *data )
{
QUrl url( urlstr );
QByteArray ba;
if ( requestMethod == QgsServerRequest::PostMethod )
{
if ( data )
{
ba.append( data );
}
}
else if ( requestMethod != QgsServerRequest::GetMethod )
{
throw QgsServerException( QStringLiteral( "Invalid method in handleRequest(): only GET or POST is supported" ) );
}
QgsBufferServerRequest request( url, requestMethod, headers, &ba );
QgsBufferServerResponse response;
handleRequest( request, response );
/*
* XXX For compatibility only:
* We should return a (moved) QgsBufferServerResponse instead
*/
QByteArray headerBuffer;
QMap<QString, QString>::const_iterator it;
for ( it = response.headers().constBegin(); it != response.headers().constEnd(); ++it )
{
headerBuffer.append( it.key().toUtf8() );
headerBuffer.append( ": " );
headerBuffer.append( it.value().toUtf8() );
headerBuffer.append( "\n" );
}
headerBuffer.append( "\n" );
// TODO: check that this is not an evil bug!
return QPair<QByteArray, QByteArray>( headerBuffer, response.body() );
}
#ifdef HAVE_SERVER_PYTHON_PLUGINS
void QgsServer::initPython()

12
src/server/qgsserver.h
View File

@ -73,18 +73,6 @@ class SERVER_EXPORT QgsServer
*/
void handleRequest( QgsServerRequest &request, QgsServerResponse &response );
/** Handles the request from query string
* The query string is normally read from environment
* but can be also passed in args and in this case overrides the environment
* variable.
*
* \param urlstr QString containing the request url (simple quely string must be preceded by '?')
* \param requestMethod QgsServerRequest::Method that indicates the method. Only "GET" or "POST" are supported.
* \param data array of bytes containing post data
* \param map of request headers
* \returns the response headers and body QPair of QByteArray
*/
QPair<QByteArray, QByteArray> handleRequest( const QString &urlstr, const QgsServerRequest::Method requestMethod = QgsServerRequest::GetMethod, const QgsServerRequest::Headers &headers = QgsServerRequest::Headers(), const char *data = nullptr );
//! Returns a pointer to the server interface
QgsServerInterfaceImpl *serverInterface() { return sServerInterface; }

38
tests/src/python/test_qgsserver.py
View File

@ -180,6 +180,18 @@ class QgsServerTestBase(unittest.TestCase):
self.assertTrue(test, message)
def _execute_request(self, qs, requestMethod=QgsServerRequest.GetMethod, data=None):
request = QgsBufferServerRequest(qs, requestMethod, {}, data)
response = QgsBufferServerResponse()
self.server.handleRequest(request, response)
headers = []
rh = response.headers()
rk = list(rh.keys())
rk.sort()
for k in rk:
headers.append(("%s: %s" % (k, rh[k])).encode('utf-8'))
return b"\n".join(headers) + b"\n\n", bytes(response.body())
class TestQgsServer(QgsServerTestBase):
"""Tests container"""
@ -196,7 +208,9 @@ class TestQgsServer(QgsServerTestBase):
"""Segfaults?"""
for i in range(10):
locals()["s%s" % i] = QgsServer()
locals()["s%s" % i].handleRequest("")
locals()["rq%s" % i] = QgsBufferServerRequest("")
locals()["re%s" % i] = QgsBufferServerResponse()
locals()["s%s" % i].handleRequest(locals()["rq%s" % i], locals()["re%s" % i])
def test_requestHandler(self):
"""Test request handler"""
@ -212,7 +226,7 @@ class TestQgsServer(QgsServerTestBase):
"""Using an empty query string (returns an XML exception)
we are going to test if headers and body are returned correctly"""
# Test as a whole
header, body = [_v for _v in self.server.handleRequest("")]
header, body = self._execute_request("")
response = self.strip_version_xmlns(header + body)
expected = self.strip_version_xmlns(b'Content-Length: 54\nContent-Type: text/xml; charset=utf-8\n\n<ServerException>Project file error</ServerException>\n')
self.assertEqual(response, expected)
@ -222,7 +236,7 @@ class TestQgsServer(QgsServerTestBase):
# Test response when project is specified but without service
project = self.testdata_path + "test_project_wfs.qgs"
qs = '?MAP=%s' % (urllib.parse.quote(project))
header, body = [_v for _v in self.server.handleRequest(qs)]
header, body = self._execute_request(qs)
response = self.strip_version_xmlns(header + body)
expected = self.strip_version_xmlns(b'Content-Length: 206\nContent-Type: text/xml; charset=utf-8\n\n<ServiceExceptionReport version="1.3.0" xmlns="http://www.opengis.net/ogc">\n <ServiceException code="Service configuration error">Service unknown or unsupported</ServiceException>\n</ServiceExceptionReport>\n')
self.assertEqual(response, expected)
@ -239,7 +253,7 @@ class TestQgsServer(QgsServerTestBase):
assert os.path.exists(project), "Project file not found: " + project
query_string = '?MAP=%s&SERVICE=WFS&VERSION=1.0.0&REQUEST=%s' % (urllib.parse.quote(project), request)
header, body = self.server.handleRequest(query_string)
header, body = self._execute_request(query_string)
self.assert_headers(header, body)
response = header + body
reference_path = self.testdata_path + 'wfs_' + request.lower() + '.txt'
@ -262,7 +276,7 @@ class TestQgsServer(QgsServerTestBase):
assert os.path.exists(project), "Project file not found: " + project
query_string = '?MAP=%s&SERVICE=WFS&VERSION=1.0.0&REQUEST=%s' % (urllib.parse.quote(project), request)
header, body = self.server.handleRequest(query_string)
header, body = self._execute_request(query_string)
self.result_compare(
'wfs_getfeature_' + requestid + '.txt',
"request %s failed.\n Query: %s" % (
@ -294,7 +308,7 @@ class TestQgsServer(QgsServerTestBase):
"STYLES": ""
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
for item in str(r).split("\\n"):
if "onlineResource" in item:
@ -310,7 +324,7 @@ class TestQgsServer(QgsServerTestBase):
"STYLES": ""
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
for item in str(r).split("\\n"):
if "onlineResource" in item:
@ -326,7 +340,7 @@ class TestQgsServer(QgsServerTestBase):
"STYLES": ""
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
for item in str(r).split("\\n"):
if "onlineResource" in item:
@ -349,7 +363,7 @@ class TestQgsServer(QgsServerTestBase):
assert os.path.exists(project), "Project file not found: " + project
query_string = '?MAP={}'.format(urllib.parse.quote(project))
header, body = self.server.handleRequest(query_string, requestMethod=QgsServerRequest.PostMethod, data=request)
header, body = self._execute_request(query_string, requestMethod=QgsServerRequest.PostMethod, data=request.encode('utf-8'))
self.result_compare(
'wfs_getfeature_{}.txt'.format(requestid),
@ -389,7 +403,7 @@ class TestQgsServer(QgsServerTestBase):
assert os.path.exists(project), "Project file not found: " + project
query_string = '?MAP=%s&SERVICE=WCS&VERSION=1.0.0&REQUEST=%s' % (urllib.parse.quote(project), request)
header, body = self.server.handleRequest(query_string)
header, body = self._execute_request(query_string)
self.assert_headers(header, body)
response = header + body
reference_path = self.testdata_path + 'wcs_' + request.lower() + '.txt'
@ -418,7 +432,7 @@ class TestQgsServer(QgsServerTestBase):
"STYLES": ""
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
item_found = False
for item in str(r).split("\\n"):
@ -437,7 +451,7 @@ class TestQgsServer(QgsServerTestBase):
"STYLES": ""
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
item_found = False
for item in str(r).split("\\n"):

35
tests/src/python/test_qgsserver_accesscontrol.py
View File

@ -23,7 +23,7 @@ from qgis.testing import unittest
from utilities import unitTestDataPath
from osgeo import gdal
from osgeo.gdalconst import GA_ReadOnly
from qgis.server import QgsServer, QgsAccessControlFilter, QgsServerRequest
from qgis.server import QgsServer, QgsAccessControlFilter, QgsServerRequest, QgsBufferServerRequest, QgsBufferServerResponse
from qgis.core import QgsRenderChecker, QgsApplication
from qgis.PyQt.QtCore import QSize
import tempfile
@ -160,12 +160,27 @@ class RestrictedAccessControl(QgsAccessControlFilter):
class TestQgsServerAccessControl(unittest.TestCase):
@classmethod
def _execute_request(cls, qs, requestMethod=QgsServerRequest.GetMethod, data=None):
if data is not None:
data = data.encode('utf-8')
request = QgsBufferServerRequest(qs, requestMethod, {}, data)
response = QgsBufferServerResponse()
cls._server.handleRequest(request, response)
headers = []
rh = response.headers()
rk = list(rh.keys())
rk.sort()
for k in rk:
headers.append(("%s: %s" % (k, rh[k])).encode('utf-8'))
return b"\n".join(headers) + b"\n\n", bytes(response.body())
@classmethod
def setUpClass(cls):
"""Run before all tests"""
cls._app = QgsApplication([], False)
cls._server = QgsServer()
cls._server.handleRequest("")
cls._execute_request("")
cls._server_iface = cls._server.serverInterface()
cls._accesscontrol = RestrictedAccessControl(cls._server_iface)
cls._server_iface.registerAccessControl(cls._accesscontrol, 100)
@ -1375,7 +1390,7 @@ class TestQgsServerAccessControl(unittest.TestCase):
def _handle_request(self, restricted, query_string, **kwargs):
self._accesscontrol._active = restricted
qs = "?" + query_string if query_string is not None else ''
result = self._result(self._server.handleRequest(qs, **kwargs))
result = self._result(self._execute_request(qs, **kwargs))
return result
def _result(self, data):
@ -1432,14 +1447,16 @@ class TestQgsServerAccessControl(unittest.TestCase):
with open(os.path.join(tempfile.gettempdir(), image + "_result.png"), "rb") as rendered_file:
encoded_rendered_file = base64.b64encode(rendered_file.read())
message = "Image is wrong\n%s\nImage:\necho '%s' | base64 -d >%s/%s_result.png" % (
report, encoded_rendered_file.strip(), tempfile.gettempdir(), image
report, encoded_rendered_file.strip().decode('utf8'), tempfile.gettempdir(), image
)
with open(os.path.join(tempfile.gettempdir(), image + "_result_diff.png"), "rb") as diff_file:
encoded_diff_file = base64.b64encode(diff_file.read())
message += "\nDiff:\necho '%s' | base64 -d > %s/%s_result_diff.png" % (
encoded_diff_file.strip(), tempfile.gettempdir(), image
)
# If the failure is in image sizes the diff file will not exists.
if os.path.exists(os.path.join(tempfile.gettempdir(), image + "_result_diff.png")):
with open(os.path.join(tempfile.gettempdir(), image + "_result_diff.png"), "rb") as diff_file:
encoded_diff_file = base64.b64encode(diff_file.read())
message += "\nDiff:\necho '%s' | base64 -d > %s/%s_result_diff.png" % (
encoded_diff_file.strip().decode('utf8'), tempfile.gettempdir(), image
)
self.assertTrue(test, message)

62
tests/src/python/test_qgsserver_plugins.py
View File

@ -17,17 +17,13 @@ __copyright__ = 'Copyright 2017, The QGIS Project'
__revision__ = '$Format:%H$'
import os
import re
import urllib.request
import urllib.parse
import urllib.error
import email
from io import StringIO
from qgis.server import QgsServer
from qgis.core import QgsMessageLog, QgsApplication
from qgis.core import QgsMessageLog
from qgis.testing import unittest
from utilities import unitTestDataPath
from test_qgsserver import QgsServerTestBase
import osgeo.gdal # NOQA
@ -37,37 +33,7 @@ RE_STRIP_UNCHECKABLE = b'MAP=[^"]+|Content-Length: \d+'
RE_ATTRIBUTES = b'[^>\s]+=[^>\s]+'
class TestQgsServerPlugins(unittest.TestCase):
def assertXMLEqual(self, response, expected, msg=''):
"""Compare XML line by line and sorted attributes"""
response_lines = response.splitlines()
expected_lines = expected.splitlines()
line_no = 1
for expected_line in expected_lines:
expected_line = expected_line.strip()
response_line = response_lines[line_no - 1].strip()
# Compare tag
try:
self.assertEqual(re.findall(b'<([^>\s]+)[ >]', expected_line)[0],
re.findall(b'<([^>\s]+)[ >]', response_line)[0], msg=msg + "\nTag mismatch on line %s: %s != %s" % (line_no, expected_line, response_line))
except IndexError:
self.assertEqual(expected_line, response_line, msg=msg + "\nTag line mismatch %s: %s != %s" % (line_no, expected_line, response_line))
# print("---->%s\t%s == %s" % (line_no, expected_line, response_line))
# Compare attributes
if re.match(RE_ATTRIBUTES, expected_line): # has attrs
expected_attrs = sorted(re.findall(RE_ATTRIBUTES, expected_line))
response_attrs = sorted(re.findall(RE_ATTRIBUTES, response_line))
self.assertEqual(expected_attrs, response_attrs, msg=msg + "\nXML attributes differ at line {0}: {1} != {2}".format(line_no, expected_attrs, response_attrs))
line_no += 1
@classmethod
def setUpClass(cls):
cls.app = QgsApplication([], False)
@classmethod
def tearDownClass(cls):
cls.app.exitQgis()
class TestQgsServerPlugins(QgsServerTestBase):
def setUp(self):
"""Create the server instance"""
@ -85,20 +51,6 @@ class TestQgsServerPlugins(unittest.TestCase):
pass
self.server = QgsServer()
def strip_version_xmlns(self, text):
"""Order of attributes is random, strip version and xmlns"""
return text.replace(b'version="1.3.0"', b'').replace(b'xmlns="http://www.opengis.net/ogc"', b'')
def assert_headers(self, header, body):
stream = StringIO()
header_string = header.decode('utf-8')
stream.write(header_string)
headers = email.message_from_string(header_string)
if 'content-length' in headers:
content_length = int(headers['content-length'])
body_length = len(body)
self.assertEqual(content_length, body_length, msg="Header reported content-length: %d Actual body length was: %d" % (content_length, body_length))
def test_pluginfilters(self):
"""Test python plugins filters"""
try:
@ -193,7 +145,7 @@ class TestQgsServerPlugins(unittest.TestCase):
self.assertTrue(filter2 in serverIface.filters()[100])
self.assertEqual(filter1, serverIface.filters()[101][0])
self.assertEqual(filter2, serverIface.filters()[200][0])
header, body = [_v for _v in self.server.handleRequest('?service=simple')]
header, body = [_v for _v in self._execute_request('?service=simple')]
response = header + body
expected = b'Content-Length: 62\nContent-type: text/plain\n\nHello from SimpleServer!Hello from Filter1!Hello from Filter2!'
self.assertEqual(response, expected)
@ -211,7 +163,7 @@ class TestQgsServerPlugins(unittest.TestCase):
self.assertTrue(filter2 in serverIface.filters()[100])
self.assertEqual(filter1, serverIface.filters()[101][0])
self.assertEqual(filter2, serverIface.filters()[200][0])
header, body = [_v for _v in self.server.handleRequest('?service=simple')]
header, body = [_v for _v in self._execute_request('?service=simple')]
response = header + body
expected = b'Content-Length: 62\nContent-type: text/plain\n\nHello from SimpleServer!Hello from Filter1!Hello from Filter2!'
self.assertEqual(response, expected)
@ -219,7 +171,7 @@ class TestQgsServerPlugins(unittest.TestCase):
# Now, re-run with body setter
filter5 = Filter5(serverIface)
serverIface.registerFilter(filter5, 500)
header, body = [_v for _v in self.server.handleRequest('?service=simple')]
header, body = [_v for _v in self._execute_request('?service=simple')]
response = header + body
expected = b'Content-Length: 19\nContent-type: text/plain\n\nnew body, new life!'
self.assertEqual(response, expected)

9
tests/src/python/test_qgsserver_security.py
View File

@ -26,9 +26,10 @@ from qgis.core import QgsApplication
from qgis.server import QgsServer
from qgis.testing import unittest
from utilities import unitTestDataPath
from test_qgsserver import QgsServerTestBase
class TestQgsServerSecurity(unittest.TestCase):
class TestQgsServerSecurity(QgsServerTestBase):
@classmethod
def setUpClass(cls):
@ -320,7 +321,7 @@ class TestQgsServerSecurity(unittest.TestCase):
"CRS": "EPSG:32613",
"FILTER": filter_xml}.items())])
return self.server.handleRequest(qs)
return self._execute_request(qs)
def handle_request_wms_getfeatureinfo(self, filter_sql):
qs = "?" + "&".join(["%s=%s" % i for i in list({
@ -338,7 +339,7 @@ class TestQgsServerSecurity(unittest.TestCase):
"CRS": "EPSG:32613",
"FILTER": filter_sql}.items())])
return self._result(self.server.handleRequest(qs))
return self._result(self._execute_request(qs))
def handle_request_wms_getmap(self, sld):
qs = "?" + "&".join(["%s=%s" % i for i in list({
@ -356,7 +357,7 @@ class TestQgsServerSecurity(unittest.TestCase):
"CRS": "EPSG:32613",
"SLD": sld}.items())])
return self._result(self.server.handleRequest(qs))
return self._result(self._execute_request(qs))
def is_point_table_still_exist(self):
conn = spatialite_connect(self.db_clone)

74
tests/src/python/test_qgsserver_wms.py
View File

@ -51,7 +51,7 @@ class TestQgsServerWMS(QgsServerTestBase):
query_string = 'https://www.qgis.org/?MAP=%s&SERVICE=WMS&VERSION=1.3&REQUEST=%s' % (urllib.parse.quote(project), request)
if extra is not None:
query_string += extra
header, body = self.server.handleRequest(query_string)
header, body = self._execute_request(query_string)
response = header + body
reference_path = self.testdata_path + (request.lower() if not reference_file else reference_file) + '.txt'
self.store_reference(reference_path, response)
@ -102,7 +102,7 @@ class TestQgsServerWMS(QgsServerTestBase):
assert os.path.exists(project), "Project file not found: " + project
query_string = '?MAP=%s&SERVICE=WMS&VERSION=1.3.0&REQUEST=%s' % (urllib.parse.quote(project), request)
header, body = self.server.handleRequest(query_string)
header, body = self._execute_request(query_string)
response = header + body
reference_path = self.testdata_path + request.lower() + '_inspire.txt'
self.store_reference(reference_path, response)
@ -133,7 +133,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:3857"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetMap_Basic")
def test_wms_getmap_transparent(self):
@ -152,7 +152,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"TRANSPARENT": "TRUE"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetMap_Transparent")
def test_wms_getmap_background(self):
@ -171,7 +171,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"BGCOLOR": "green"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetMap_Background")
qs = "?" + "&".join(["%s=%s" % i for i in list({
@ -189,7 +189,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"BGCOLOR": "0x008000"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetMap_Background_Hex")
def test_wms_getcapabilities_url(self):
@ -203,7 +203,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"STYLES": ""
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
item_found = False
for item in str(r).split("\\n"):
@ -222,7 +222,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"STYLES": ""
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
item_found = False
for item in str(r).split("\\n"):
@ -241,7 +241,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"STYLES": ""
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
item_found = False
for item in str(r).split("\\n"):
@ -265,7 +265,7 @@ class TestQgsServerWMS(QgsServerTestBase):
}.items())])
expected = self.strip_version_xmlns(b'<ServiceExceptionReport version="1.3.0" xmlns="http://www.opengis.net/ogc">\n <ServiceException code="Size error">The requested map size is too large</ServiceException>\n</ServiceExceptionReport>\n')
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self.assertEqual(self.strip_version_xmlns(r), expected)
@ -284,7 +284,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:3857"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetMap_LayerOrder")
def test_wms_getmap_srs(self):
@ -302,7 +302,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:4326"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetMap_SRS")
def test_wms_getmap_style(self):
@ -321,7 +321,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:3857"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetMap_StyleDefault")
# custom style
@ -339,7 +339,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:3857"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetMap_StyleCustom")
def test_wms_getmap_filter(self):
@ -358,7 +358,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"FILTER": "Country:\"name\" = 'eurasia'"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetMap_Filter")
def test_wms_getmap_selection(self):
@ -377,7 +377,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"SELECTION": "Country: 4"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetMap_Selection")
def test_wms_getmap_opacities(self):
@ -396,7 +396,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"OPACITIES": "125, 50"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetMap_Opacities")
def test_wms_getprint_basic(self):
@ -414,7 +414,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:3857"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetPrint_Basic")
@unittest.skip('Randomly failing to draw the map layer')
@ -433,7 +433,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:4326"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetPrint_SRS")
def test_wms_getprint_scale(self):
@ -452,7 +452,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:3857"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetPrint_Scale")
def test_wms_getprint_grid(self):
@ -472,7 +472,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:3857"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetPrint_Grid")
def test_wms_getprint_rotation(self):
@ -491,7 +491,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:3857"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetPrint_Rotation")
def test_wms_getprint_selection(self):
@ -510,7 +510,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"SELECTION": "Country: 4"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetPrint_Selection")
def test_getLegendGraphics(self):
@ -526,7 +526,7 @@ class TestQgsServerWMS(QgsServerTestBase):
'LAYER': 'testlayer%20èé',
}
qs = '?' + '&'.join(["%s=%s" % (k, v) for k, v in parms.items()])
h, r = self.server.handleRequest(qs)
h, r = self._execute_request(qs)
self.assertEqual(-1, h.find(b'Content-Type: text/xml; charset=utf-8'), "Header: %s\nResponse:\n%s" % (h, r))
self.assertNotEqual(-1, h.find(b'Content-Type: image/png'), "Header: %s\nResponse:\n%s" % (h, r))
@ -544,7 +544,7 @@ class TestQgsServerWMS(QgsServerTestBase):
'LAYERTITLE': 'TRUE',
}
qs = '?' + '&'.join([u"%s=%s" % (k, v) for k, v in parms.items()])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetLegendGraphic_test", 250, QSize(15, 15))
parms = {
@ -559,7 +559,7 @@ class TestQgsServerWMS(QgsServerTestBase):
'LAYERTITLE': 'FALSE',
}
qs = '?' + '&'.join([u"%s=%s" % (k, v) for k, v in parms.items()])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetLegendGraphic_test_layertitle_false", 250, QSize(15, 15))
def test_wms_GetLegendGraphic_Basic(self):
@ -576,7 +576,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:3857"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetLegendGraphic_Basic")
def test_wms_GetLegendGraphic_Transparent(self):
@ -594,7 +594,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"TRANSPARENT": "TRUE"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetLegendGraphic_Transparent")
def test_wms_GetLegendGraphic_Background(self):
@ -612,7 +612,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"BGCOLOR": "green"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetLegendGraphic_Background")
qs = "?" + "&".join(["%s=%s" % i for i in list({
@ -629,7 +629,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"BGCOLOR": "0x008000"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetLegendGraphic_Background_Hex")
def test_wms_GetLegendGraphic_BoxSpace(self):
@ -647,7 +647,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:3857"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetLegendGraphic_BoxSpace")
def test_wms_GetLegendGraphic_SymbolSpace(self):
@ -665,7 +665,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:3857"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetLegendGraphic_SymbolSpace")
def test_wms_GetLegendGraphic_IconLabelSpace(self):
@ -683,7 +683,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:3857"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetLegendGraphic_IconLabelSpace")
def test_wms_GetLegendGraphic_SymbolSize(self):
@ -702,7 +702,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:3857"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetLegendGraphic_SymbolSize")
def test_wms_GetLegendGraphic_BBox(self):
@ -720,7 +720,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"CRS": "EPSG:4326"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetLegendGraphic_BBox")
def test_wms_GetLegendGraphic_BBox2(self):
@ -738,7 +738,7 @@ class TestQgsServerWMS(QgsServerTestBase):
"SRS": "EPSG:4326"
}.items())])
r, h = self._result(self.server.handleRequest(qs))
r, h = self._result(self._execute_request(qs))
self._img_diff_error(r, h, "WMS_GetLegendGraphic_BBox2")
# WCS tests
@ -747,7 +747,7 @@ class TestQgsServerWMS(QgsServerTestBase):
assert os.path.exists(project), "Project file not found: " + project
query_string = '?MAP=%s&SERVICE=WCS&VERSION=1.0.0&REQUEST=%s' % (urllib.parse.quote(project), request)
header, body = self.server.handleRequest(query_string)
header, body = self._execute_request(query_string)
self.assert_headers(header, body)
response = header + body
reference_path = self.testdata_path + 'wcs_' + request.lower() + '.txt'