def do_GET(self):
parsed_path = urlparse.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.split('\n') if h]:
self.send_header(k, v)
self.end_headers()
self.wfile.write(body)
return
def setUp(self):
"""Create the server instance"""
self.testdata_path = unitTestDataPath('qgis_server') + '/'
# 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
self.server = QgsServer()
def setUp(self):
"""Create the server instance"""
self.testdata_path = unitTestDataPath('qgis_server') + '/'
d = unitTestDataPath('qgis_server_accesscontrol') + '/'
self.projectPath = os.path.join(d, "project.qgs")
# 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
self.server = QgsServer()
# Check if OAuth2 is enabled
QGIS_SERVER_OAUTH2_AUTH = (
QGIS_SERVER_OAUTH2_CERTIFICATE is not None and
os.path.isfile(QGIS_SERVER_OAUTH2_CERTIFICATE) and
QGIS_SERVER_OAUTH2_KEY is not None and
os.path.isfile(QGIS_SERVER_OAUTH2_KEY) and
QGIS_SERVER_OAUTH2_AUTHORITY is not None and
os.path.isfile(QGIS_SERVER_OAUTH2_AUTHORITY) and
QGIS_SERVER_OAUTH2_USERNAME and QGIS_SERVER_OAUTH2_PASSWORD)
HTTPS_ENABLED = QGIS_SERVER_PKI_AUTH or QGIS_SERVER_OAUTH2_AUTH
qgs_app = QgsApplication([], False)
qgs_server = QgsServer()
if QGIS_SERVER_HTTP_BASIC_AUTH:
from qgis.server import QgsServerFilter
import base64
class HTTPBasicFilter(QgsServerFilter):
def requestReady(self):
handler = self.serverInterface().requestHandler()
auth = self.serverInterface().requestHandler().requestHeader('HTTP_AUTHORIZATION')
if auth:
username, password = base64.b64decode(auth[6:]).split(b':')
if (username.decode('utf-8') == os.environ.get('QGIS_SERVER_USERNAME', 'username') and
password.decode('utf-8') == os.environ.get('QGIS_SERVER_PASSWORD', 'password')):
class TestQgsServer(unittest.TestCase):
def setUp(self):
"""Create the server instance"""
self.testdata_path = unitTestDataPath('qgis_server') + '/'
# 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
self.server = QgsServer()
def test_destructor_segfaults(self):
"""Segfault on destructor?"""
server = QgsServer()
del server
def test_multiple_servers(self):
"""Segfaults?"""
for i in range(10):
locals()["s%s" % i] = QgsServer()
locals()["s%s" % i].handleRequest()
def test_api(self):
"""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 = [str(_v) for _v in self.server.handleRequest()]
response = header + body
expected = '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)
expected = 'Content-Length: 206\nContent-Type: text/xml; charset=utf-8\n\n'
self.assertEqual(header, expected)
# Test body
expected = '<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(body, expected)
def test_pluginfilters(self):
"""Test python plugins filters"""
try:
from qgis.server import QgsServerFilter
except ImportError:
print "QGIS Server plugins are not compiled. Skipping test"
return
class SimpleHelloFilter(QgsServerFilter):
def requestReady(self):
QgsMessageLog.logMessage("SimpleHelloFilter.requestReady")
def sendResponse(self):
QgsMessageLog.logMessage("SimpleHelloFilter.sendResponse")
def responseComplete(self):
request = self.serverInterface().requestHandler()
params = request.parameterMap()
QgsMessageLog.logMessage("SimpleHelloFilter.responseComplete")
if params.get('SERVICE', '').upper() == 'SIMPLE':
request.clearHeaders()
request.setHeader('Content-type', 'text/plain')
request.clearBody()
request.appendBody('Hello from SimpleServer!')
serverIface = self.server.serverInterface()
filter = SimpleHelloFilter(serverIface)
serverIface.registerFilter(filter, 100)
# Get registered filters
self.assertEqual(filter, serverIface.filters()[100][0])
# Register some more filters
class Filter1(QgsServerFilter):
def responseComplete(self):
request = self.serverInterface().requestHandler()
params = request.parameterMap()
if params.get('SERVICE', '').upper() == 'SIMPLE':
request.appendBody('Hello from Filter1!')
class Filter2(QgsServerFilter):
def responseComplete(self):
request = self.serverInterface().requestHandler()
params = request.parameterMap()
if params.get('SERVICE', '').upper() == 'SIMPLE':
request.appendBody('Hello from Filter2!')
filter1 = Filter1(serverIface)
filter2 = Filter2(serverIface)
serverIface.registerFilter(filter1, 101)
serverIface.registerFilter(filter2, 200)
serverIface.registerFilter(filter2, 100)
self.assertTrue(filter2 in serverIface.filters()[100])
self.assertEqual(filter1, serverIface.filters()[101][0])
self.assertEqual(filter2, serverIface.filters()[200][0])
header, body = [str(_v) for _v in self.server.handleRequest('service=simple')]
response = header + body
expected = 'Content-type: text/plain\n\nHello from SimpleServer!Hello from Filter1!Hello from Filter2!'
self.assertEqual(response, expected)
#.........这里部分代码省略.........
class TestQgsServer(unittest.TestCase):
def setUp(self):
"""Create the server instance"""
self.testdata_path = unitTestDataPath('qgis_server') + '/'
# 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
self.server = QgsServer()
def assert_headers(self, header, body):
headers = Message(StringIO(header))
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_destructor_segfaults(self):
"""Segfault on destructor?"""
server = QgsServer()
del server
def test_multiple_servers(self):
"""Segfaults?"""
for i in range(10):
locals()["s%s" % i] = QgsServer()
locals()["s%s" % i].handleRequest()
def test_api(self):
"""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 = [str(_v) for _v in self.server.handleRequest()]
response = header + body
expected = '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)
expected = 'Content-Length: 206\nContent-Type: text/xml; charset=utf-8\n\n'
self.assertEqual(header, expected)
# Test body
expected = '<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(body, expected)
def test_pluginfilters(self):
"""Test python plugins filters"""
try:
from qgis.server import QgsServerFilter
except ImportError:
print("QGIS Server plugins are not compiled. Skipping test")
return
class SimpleHelloFilter(QgsServerFilter):
def requestReady(self):
QgsMessageLog.logMessage("SimpleHelloFilter.requestReady")
def sendResponse(self):
QgsMessageLog.logMessage("SimpleHelloFilter.sendResponse")
def responseComplete(self):
request = self.serverInterface().requestHandler()
params = request.parameterMap()
QgsMessageLog.logMessage("SimpleHelloFilter.responseComplete")
if params.get('SERVICE', '').upper() == 'SIMPLE':
request.clearHeaders()
request.setHeader('Content-type', 'text/plain')
request.clearBody()
request.appendBody('Hello from SimpleServer!')
serverIface = self.server.serverInterface()
filter = SimpleHelloFilter(serverIface)
serverIface.registerFilter(filter, 100)
# Get registered filters
self.assertEqual(filter, serverIface.filters()[100][0])
# Register some more filters
class Filter1(QgsServerFilter):
def responseComplete(self):
request = self.serverInterface().requestHandler()
params = request.parameterMap()
if params.get('SERVICE', '').upper() == 'SIMPLE':
request.appendBody('Hello from Filter1!')
class Filter2(QgsServerFilter):
def responseComplete(self):
request = self.serverInterface().requestHandler()
params = request.parameterMap()
if params.get('SERVICE', '').upper() == 'SIMPLE':
request.appendBody('Hello from Filter2!')
filter1 = Filter1(serverIface)
filter2 = Filter2(serverIface)
serverIface.registerFilter(filter1, 101)
serverIface.registerFilter(filter2, 200)
serverIface.registerFilter(filter2, 100)
#.........这里部分代码省略.........
客服电话
APP下载
官方微信
请发表评论