1# -*- coding: utf-8 -*- 2"""QGIS Unit tests for QgsServer. 3 4Set the env var ENCODED_OUTPUT to enable printing the base64 encoded image diff 5 6FIXME: keep here only generic server tests and move specific services 7 tests to test_qgsserver_<service>.py 8 9 Already moved services and functionality: 10 - WMS 11 - plugins 12 - settings 13 - WFS-T integration test 14 15 TODO: 16 - WFS 17 - WCS 18 19.. note:: This program is free software; you can redistribute it and/or modify 20it under the terms of the GNU General Public License as published by 21the Free Software Foundation; either version 2 of the License, or 22(at your option) any later version. 23 24""" 25__author__ = 'Alessandro Pasotti' 26__date__ = '25/05/2015' 27__copyright__ = 'Copyright 2015, The QGIS Project' 28 29import os 30 31# Deterministic XML 32os.environ['QT_HASH_SEED'] = '1' 33 34import re 35import urllib.request 36import urllib.parse 37import urllib.error 38import email 39import difflib 40 41from io import StringIO 42from qgis.server import QgsServer, QgsServerRequest, QgsBufferServerRequest, QgsBufferServerResponse 43from qgis.core import QgsRenderChecker, QgsApplication, QgsFontUtils, QgsMultiRenderChecker 44from qgis.testing import unittest, start_app 45from qgis.PyQt.QtCore import QSize 46from qgis.PyQt.QtGui import QColor 47from utilities import unitTestDataPath 48 49import osgeo.gdal # NOQA 50import tempfile 51import base64 52 53 54start_app() 55 56# Strip path and content length because path may vary 57RE_STRIP_UNCHECKABLE = br'MAP=[^"]+|Content-Length: \d+' 58RE_ELEMENT = br'</*([^>\[\s]+)[ >]' 59RE_ELEMENT_CONTENT = br'<[^>\[]+>(.+)</[^>\[\s]+>' 60RE_ATTRIBUTES = rb'((?:(?!\s|=).)*)\s*?=\s*?["\']?((?:(?<=")(?:(?<=\\)"|[^"])*|(?<=\')(?:(?<=\\)\'|[^\'])*)|(?:(?!"|\')(?:(?!\/>|>|\s).)+))' 61 62 63class QgsServerTestBase(unittest.TestCase): 64 65 """Base class for QGIS server tests""" 66 67 # Set to True in child classes to re-generate reference files for this class 68 regenerate_reference = False 69 70 def assertXMLEqual(self, response, expected, msg='', raw=False): 71 """Compare XML line by line and sorted attributes""" 72 response_lines = response.splitlines() 73 expected_lines = expected.splitlines() 74 line_no = 1 75 76 diffs = [] 77 for diff in difflib.unified_diff([l.decode('utf8') for l in expected_lines], [l.decode('utf8') for l in response_lines]): 78 diffs.append(diff) 79 80 self.assertEqual(len(expected_lines), len(response_lines), "Expected and response have different number of lines!\n{}\n{}".format(msg, '\n'.join(diffs))) 81 for expected_line in expected_lines: 82 expected_line = expected_line.strip() 83 response_line = response_lines[line_no - 1].strip() 84 response_line = response_line.replace(b'e+6', br'e+06') 85 # Compare tag 86 if re.match(RE_ELEMENT, expected_line) and not raw: 87 expected_elements = re.findall(RE_ELEMENT, expected_line) 88 response_elements = re.findall(RE_ELEMENT, response_line) 89 self.assertEqual(expected_elements[0], 90 response_elements[0], msg=msg + "\nTag mismatch on line %s: %s != %s" % (line_no, expected_line, response_line)) 91 # Compare content 92 if len(expected_elements) == 2 and expected_elements[0] == expected_elements[1]: 93 expected_element_content = re.findall(RE_ELEMENT_CONTENT, expected_line) 94 response_element_content = re.findall(RE_ELEMENT_CONTENT, response_line) 95 self.assertEqual(len(expected_element_content), len(response_element_content), 96 msg=msg + "\nContent mismatch on line %s: %s != %s" % (line_no, expected_line, response_line)) 97 if len(expected_element_content): 98 self.assertEqual(expected_element_content[0], 99 response_element_content[0], msg=msg + "\nContent mismatch on line %s: %s != %s" % (line_no, expected_line, response_line)) 100 else: 101 self.assertEqual(expected_line, response_line, msg=msg + "\nTag line mismatch %s: %s != %s\n%s" % (line_no, expected_line, response_line, msg)) 102 # print("---->%s\t%s == %s" % (line_no, expected_line, response_line)) 103 # Compare attributes 104 if re.findall(RE_ATTRIBUTES, expected_line): # has attrs 105 expected_attrs, expected_values = zip(*sorted(re.findall(RE_ATTRIBUTES, expected_line))) 106 self.assertTrue(re.findall(RE_ATTRIBUTES, response_line), msg=msg + "\nXML attributes differ at line {0}: {1} != {2}".format(line_no, expected_line, response_line)) 107 response_attrs, response_values = zip(*sorted(re.findall(RE_ATTRIBUTES, response_line))) 108 self.assertEqual(expected_attrs, response_attrs, msg=msg + "\nXML attributes differ at line {0}: {1} != {2}".format(line_no, expected_attrs, response_attrs)) 109 self.assertEqual(expected_values, response_values, msg=msg + "\nXML attribute values differ at line {0}: {1} != {2}".format(line_no, expected_values, response_values)) 110 line_no += 1 111 112 def setUp(self): 113 """Create the server instance""" 114 self.fontFamily = QgsFontUtils.standardTestFontFamily() 115 QgsFontUtils.loadStandardTestFonts(['All']) 116 117 self.testdata_path = unitTestDataPath('qgis_server') + '/' 118 119 d = unitTestDataPath('qgis_server_accesscontrol') + '/' 120 self.projectPath = os.path.join(d, "project.qgs") 121 self.projectAnnotationPath = os.path.join(d, "project_with_annotations.qgs") 122 self.projectStatePath = os.path.join(d, "project_state.qgs") 123 self.projectUseLayerIdsPath = os.path.join(d, "project_use_layerids.qgs") 124 self.projectGroupsPath = os.path.join(d, "project_groups.qgs") 125 126 # Clean env just to be sure 127 env_vars = ['QUERY_STRING', 'QGIS_PROJECT_FILE'] 128 for ev in env_vars: 129 try: 130 del os.environ[ev] 131 except KeyError: 132 pass 133 134 self.server = QgsServer() 135 136 # Disable landing page API to test standard legacy XML responses in case of errors 137 os.environ["QGIS_SERVER_DISABLED_APIS"] = "Landing Page" 138 139 def tearDown(self): 140 """Cleanup env""" 141 142 super().tearDown() 143 try: 144 del os.environ["QGIS_SERVER_DISABLED_APIS"] 145 except KeyError: 146 pass 147 148 def strip_version_xmlns(self, text): 149 """Order of attributes is random, strip version and xmlns""" 150 return text.replace(b'version="1.3.0"', b'').replace(b'xmlns="http://www.opengis.net/ogc"', b'') 151 152 def assert_headers(self, header, body): 153 stream = StringIO() 154 header_string = header.decode('utf-8') 155 stream.write(header_string) 156 headers = email.message_from_string(header_string) 157 if 'content-length' in headers: 158 content_length = int(headers['content-length']) 159 body_length = len(body) 160 self.assertEqual(content_length, body_length, msg="Header reported content-length: %d Actual body length was: %d" % (content_length, body_length)) 161 162 @classmethod 163 def store_reference(self, reference_path, response): 164 """Utility to store reference files""" 165 166 # Normally this is false 167 if not self.regenerate_reference: 168 return 169 170 # Store the output for debug or to regenerate the reference documents: 171 f = open(reference_path, 'wb+') 172 f.write(response) 173 f.close() 174 175 def _result(self, data): 176 headers = {} 177 for line in data[0].decode('UTF-8').split("\n"): 178 if line != "": 179 header = line.split(":") 180 self.assertEqual(len(header), 2, line) 181 headers[str(header[0])] = str(header[1]).strip() 182 183 return data[1], headers 184 185 def _img_diff(self, image, control_image, max_diff, max_size_diff=QSize(), outputFormat='PNG'): 186 187 if outputFormat == 'PNG': 188 extFile = 'png' 189 elif outputFormat == 'JPG': 190 extFile = 'jpg' 191 elif outputFormat == 'WEBP': 192 extFile = 'webp' 193 else: 194 raise RuntimeError('Yeah, new format implemented') 195 196 temp_image = os.path.join(tempfile.gettempdir(), "%s_result.%s" % (control_image, extFile)) 197 198 with open(temp_image, "wb") as f: 199 f.write(image) 200 201 if outputFormat != 'PNG': 202 return (True, "QgsRenderChecker can only be used for PNG") 203 204 control = QgsMultiRenderChecker() 205 control.setControlPathPrefix("qgis_server") 206 control.setControlName(control_image) 207 control.setRenderedImage(temp_image) 208 if max_size_diff.isValid(): 209 control.setSizeTolerance(max_size_diff.width(), max_size_diff.height()) 210 return control.runTest(control_image, max_diff), control.report() 211 212 def _img_diff_error(self, response, headers, image, max_diff=100, max_size_diff=QSize(), unittest_data_path='control_images', outputFormat='PNG'): 213 """ 214 :param outputFormat: PNG, JPG or WEBP 215 """ 216 217 if outputFormat == 'PNG': 218 extFile = 'png' 219 contentType = 'image/png' 220 elif outputFormat == 'JPG': 221 extFile = 'jpg' 222 contentType = 'image/jpeg' 223 elif outputFormat == 'WEBP': 224 extFile = 'webp' 225 contentType = 'image/webp' 226 else: 227 raise RuntimeError('Yeah, new format implemented') 228 229 reference_path = unitTestDataPath(unittest_data_path) + '/qgis_server/' + image + '/' + image + '.' + extFile 230 self.store_reference(reference_path, response) 231 232 self.assertEqual( 233 headers.get("Content-Type"), contentType, 234 "Content type is wrong: %s instead of %s\n%s" % (headers.get("Content-Type"), contentType, response)) 235 236 test, report = self._img_diff(response, image, max_diff, max_size_diff, outputFormat) 237 238 with open(os.path.join(tempfile.gettempdir(), image + "_result." + extFile), "rb") as rendered_file: 239 encoded_rendered_file = base64.b64encode(rendered_file.read()) 240 if not os.environ.get('ENCODED_OUTPUT'): 241 message = "Image is wrong: rendered file %s/%s_result.%s" % (tempfile.gettempdir(), image, extFile) 242 else: 243 message = "Image is wrong\n%s\nImage:\necho '%s' | base64 -d >%s/%s_result.%s" % ( 244 report, encoded_rendered_file.strip().decode('utf8'), tempfile.gettempdir(), image, extFile 245 ) 246 247 # If the failure is in image sizes the diff file will not exists. 248 if os.path.exists(os.path.join(tempfile.gettempdir(), image + "_result_diff." + extFile)): 249 with open(os.path.join(tempfile.gettempdir(), image + "_result_diff." + extFile), "rb") as diff_file: 250 if not os.environ.get('ENCODED_OUTPUT'): 251 message = "Image is wrong: diff file %s/%s_result_diff.%s" % (tempfile.gettempdir(), image, extFile) 252 else: 253 encoded_diff_file = base64.b64encode(diff_file.read()) 254 message += "\nDiff:\necho '%s' | base64 -d > %s/%s_result_diff.%s" % ( 255 encoded_diff_file.strip().decode('utf8'), tempfile.gettempdir(), image, extFile 256 ) 257 258 self.assertTrue(test, message) 259 260 def _execute_request(self, qs, requestMethod=QgsServerRequest.GetMethod, data=None, request_headers=None): 261 request = QgsBufferServerRequest(qs, requestMethod, request_headers or {}, data) 262 response = QgsBufferServerResponse() 263 self.server.handleRequest(request, response) 264 headers = [] 265 rh = response.headers() 266 rk = sorted(rh.keys()) 267 for k in rk: 268 headers.append(("%s: %s" % (k, rh[k])).encode('utf-8')) 269 return b"\n".join(headers) + b"\n\n", bytes(response.body()) 270 271 def _execute_request_project(self, qs, project, requestMethod=QgsServerRequest.GetMethod, data=None): 272 request = QgsBufferServerRequest(qs, requestMethod, {}, data) 273 response = QgsBufferServerResponse() 274 self.server.handleRequest(request, response, project) 275 headers = [] 276 rh = response.headers() 277 rk = sorted(rh.keys()) 278 for k in rk: 279 headers.append(("%s: %s" % (k, rh[k])).encode('utf-8')) 280 return b"\n".join(headers) + b"\n\n", bytes(response.body()) 281 282 def _assert_status_code(self, status_code, qs, requestMethod=QgsServerRequest.GetMethod, data=None, project=None): 283 request = QgsBufferServerRequest(qs, requestMethod, {}, data) 284 response = QgsBufferServerResponse() 285 self.server.handleRequest(request, response, project) 286 assert response.statusCode() == status_code, "%s != %s" % (response.statusCode(), status_code) 287 288 def _assertRed(self, color: QColor): 289 self.assertEqual(color.red(), 255) 290 self.assertEqual(color.green(), 0) 291 self.assertEqual(color.blue(), 0) 292 293 def _assertGreen(self, color: QColor): 294 self.assertEqual(color.red(), 0) 295 self.assertEqual(color.green(), 255) 296 self.assertEqual(color.blue(), 0) 297 298 def _assertBlue(self, color: QColor): 299 self.assertEqual(color.red(), 0) 300 self.assertEqual(color.green(), 0) 301 self.assertEqual(color.blue(), 255) 302 303 def _assertBlack(self, color: QColor): 304 self.assertEqual(color.red(), 0) 305 self.assertEqual(color.green(), 0) 306 self.assertEqual(color.blue(), 255) 307 308 def _assertWhite(self, color: QColor): 309 self.assertEqual(color.red(), 255) 310 self.assertEqual(color.green(), 255) 311 self.assertEqual(color.blue(), 255) 312 313 314class TestQgsServerTestBase(unittest.TestCase): 315 316 def test_assert_xml_equal(self): 317 engine = QgsServerTestBase() 318 319 # test bad assertion 320 expected = b'</WFSLayers>\n<Layer queryable="1">\n' 321 response = b'<Layer>\n' 322 self.assertRaises(AssertionError, engine.assertXMLEqual, response, expected) 323 324 expected = b'</WFSLayers>\n<Layer queryable="1">\n' 325 response = b'</WFSLayers>\n<Layer>\n' 326 self.assertRaises(AssertionError, engine.assertXMLEqual, response, expected) 327 328 expected = b'</WFSLayers>\n<Layer queryable="1">\n' 329 response = b'</WFSLayers>\n<Layer fake="1">\n' 330 self.assertRaises(AssertionError, engine.assertXMLEqual, response, expected) 331 332 expected = b'</WFSLayers>\n<Layer queryable="1">\n' 333 response = b'</WFSLayers>\n<Layer queryable="2">\n' 334 self.assertRaises(AssertionError, engine.assertXMLEqual, response, expected) 335 336 expected = b'<TreeName>QGIS Test Project</TreeName>\n<Layer geometryType="Point" queryable="1" displayField="name" visible="1">\n' 337 response = b'<TreeName>QGIS Test Project</TreeName>\n<Layer geometryType="Point" queryable="1" displayField="name">\n' 338 self.assertRaises(AssertionError, engine.assertXMLEqual, response, expected) 339 340 expected = b'<TreeName>QGIS Test Project</TreeName>\n<Layer geometryType="Point" queryable="1" displayField="name" visible="1">\n' 341 response = b'<TreeName>QGIS Test Project</TreeName>\n<Layer geometryType="Point" queryable="1" displayField="name" visible="0">\n' 342 self.assertRaises(AssertionError, engine.assertXMLEqual, response, expected) 343 344 # test valid assertion 345 expected = b'</WFSLayers>\n<Layer queryable="1">\n' 346 response = b'</WFSLayers>\n<Layer queryable="1">\n' 347 self.assertFalse(engine.assertXMLEqual(response, expected)) 348 349 expected = b'<TreeName>QGIS Test Project</TreeName>\n<Layer geometryType="Point" queryable="1" displayField="name" visible="1">\n' 350 response = b'<TreeName>QGIS Test Project</TreeName>\n<Layer geometryType="Point" queryable="1" displayField="name" visible="1">\n' 351 self.assertFalse(engine.assertXMLEqual(response, expected)) 352 353 354class TestQgsServer(QgsServerTestBase): 355 356 """Tests container""" 357 358 # Set to True to re-generate reference files for this class 359 regenerate_reference = False 360 361 def test_destructor_segfaults(self): 362 """Segfault on destructor?""" 363 server = QgsServer() 364 del server 365 366 def test_multiple_servers(self): 367 """Segfaults?""" 368 for i in range(10): 369 locals()["s%s" % i] = QgsServer() 370 locals()["rq%s" % i] = QgsBufferServerRequest("") 371 locals()["re%s" % i] = QgsBufferServerResponse() 372 locals()["s%s" % i].handleRequest(locals()["rq%s" % i], locals()["re%s" % i]) 373 374 def test_requestHandler(self): 375 """Test request handler""" 376 headers = {'header-key-1': 'header-value-1', 'header-key-2': 'header-value-2'} 377 request = QgsBufferServerRequest('http://somesite.com/somepath', QgsServerRequest.GetMethod, headers) 378 response = QgsBufferServerResponse() 379 self.server.handleRequest(request, response) 380 self.assertEqual(bytes(response.body()), b'<ServerException>Project file error. For OWS services: please provide a SERVICE and a MAP parameter pointing to a valid QGIS project file</ServerException>\n') 381 self.assertEqual(response.headers(), {'Content-Length': '156', 'Content-Type': 'text/xml; charset=utf-8'}) 382 self.assertEqual(response.statusCode(), 500) 383 384 def test_requestHandlerProject(self): 385 """Test request handler with none project""" 386 headers = {'header-key-1': 'header-value-1', 'header-key-2': 'header-value-2'} 387 request = QgsBufferServerRequest('http://somesite.com/somepath', QgsServerRequest.GetMethod, headers) 388 response = QgsBufferServerResponse() 389 self.server.handleRequest(request, response, None) 390 self.assertEqual(bytes(response.body()), b'<ServerException>Project file error. For OWS services: please provide a SERVICE and a MAP parameter pointing to a valid QGIS project file</ServerException>\n') 391 self.assertEqual(response.headers(), {'Content-Length': '156', 'Content-Type': 'text/xml; charset=utf-8'}) 392 self.assertEqual(response.statusCode(), 500) 393 394 def test_api(self): 395 """Using an empty query string (returns an XML exception) 396 we are going to test if headers and body are returned correctly""" 397 # Test as a whole 398 header, body = self._execute_request("") 399 response = self.strip_version_xmlns(header + body) 400 expected = self.strip_version_xmlns(b'Content-Length: 156\nContent-Type: text/xml; charset=utf-8\n\n<ServerException>Project file error. For OWS services: please provide a SERVICE and a MAP parameter pointing to a valid QGIS project file</ServerException>\n') 401 self.assertEqual(response, expected) 402 expected = b'Content-Length: 156\nContent-Type: text/xml; charset=utf-8\n\n' 403 self.assertEqual(header, expected) 404 405 # Test response when project is specified but without service 406 project = self.testdata_path + "test_project_wfs.qgs" 407 qs = '?MAP=%s' % (urllib.parse.quote(project)) 408 header, body = self._execute_request(qs) 409 response = self.strip_version_xmlns(header + body) 410 expected = self.strip_version_xmlns(b'Content-Length: 326\nContent-Type: text/xml; charset=utf-8\n\n<ServiceExceptionReport >\n <ServiceException code="Service configuration error">Service unknown or unsupported. Current supported services (case-sensitive): WMS WFS WCS WMTS SampleService, or use a WFS3 (OGC API Features) endpoint</ServiceException>\n</ServiceExceptionReport>\n') 411 self.assertEqual(response, expected) 412 expected = b'Content-Length: 326\nContent-Type: text/xml; charset=utf-8\n\n' 413 self.assertEqual(header, expected) 414 415 # Test body 416 expected = self.strip_version_xmlns(b'<ServiceExceptionReport >\n <ServiceException code="Service configuration error">Service unknown or unsupported. Current supported services (case-sensitive): WMS WFS WCS WMTS SampleService, or use a WFS3 (OGC API Features) endpoint</ServiceException>\n</ServiceExceptionReport>\n') 417 self.assertEqual(self.strip_version_xmlns(body), expected) 418 419 # WCS tests 420 def wcs_request_compare(self, request): 421 project = self.projectPath 422 assert os.path.exists(project), "Project file not found: " + project 423 424 query_string = '?MAP=%s&SERVICE=WCS&VERSION=1.0.0&REQUEST=%s' % (urllib.parse.quote(project), request) 425 header, body = self._execute_request(query_string) 426 self.assert_headers(header, body) 427 response = header + body 428 reference_path = self.testdata_path + 'wcs_' + request.lower() + '.txt' 429 f = open(reference_path, 'rb') 430 self.store_reference(reference_path, response) 431 expected = f.read() 432 f.close() 433 response = re.sub(RE_STRIP_UNCHECKABLE, b'', response) 434 expected = re.sub(RE_STRIP_UNCHECKABLE, b'', expected) 435 436 self.assertXMLEqual(response, expected, msg="request %s failed.\n Query: %s\n Expected:\n%s\n\n Response:\n%s" % (query_string, request, expected.decode('utf-8'), response.decode('utf-8'))) 437 438 def test_project_wcs(self): 439 """Test some WCS request""" 440 for request in ('GetCapabilities', 'DescribeCoverage'): 441 self.wcs_request_compare(request) 442 443 def test_wcs_getcapabilities_url(self): 444 # empty url in project 445 project = os.path.join(self.testdata_path, "test_project_without_urls.qgs") 446 qs = "?" + "&".join(["%s=%s" % i for i in list({ 447 "MAP": urllib.parse.quote(project), 448 "SERVICE": "WCS", 449 "VERSION": "1.0.0", 450 "REQUEST": "GetCapabilities", 451 "STYLES": "" 452 }.items())]) 453 454 r, h = self._result(self._execute_request(qs)) 455 456 item_found = False 457 for item in str(r).split("\\n"): 458 if "OnlineResource" in item: 459 self.assertEqual("=\"?" in item, True) 460 item_found = True 461 self.assertTrue(item_found) 462 463 # url well defined in project 464 project = os.path.join(self.testdata_path, "test_project_with_urls.qgs") 465 qs = "?" + "&".join(["%s=%s" % i for i in list({ 466 "MAP": urllib.parse.quote(project), 467 "SERVICE": "WCS", 468 "VERSION": "1.0.0", 469 "REQUEST": "GetCapabilities", 470 "STYLES": "" 471 }.items())]) 472 473 r, h = self._result(self._execute_request(qs)) 474 475 item_found = False 476 for item in str(r).split("\\n"): 477 if "OnlineResource" in item: 478 self.assertEqual("\"my_wcs_advertised_url" in item, True) 479 item_found = True 480 self.assertTrue(item_found) 481 482 # Service URL in header 483 for header_name, header_value in (("X-Qgis-Service-Url", "http://test1"), ("X-Qgis-Wcs-Service-Url", "http://test2")): 484 # empty url in project 485 project = os.path.join(self.testdata_path, "test_project_without_urls.qgs") 486 qs = "?" + "&".join(["%s=%s" % i for i in list({ 487 "MAP": urllib.parse.quote(project), 488 "SERVICE": "WCS", 489 "VERSION": "1.0.0", 490 "REQUEST": "GetCapabilities", 491 "STYLES": "" 492 }.items())]) 493 494 r, h = self._result(self._execute_request(qs, request_headers={header_name: header_value})) 495 496 item_found = False 497 for item in str(r).split("\\n"): 498 if "OnlineResource" in item: 499 print(item) 500 print(header_name) 501 print(header_value) 502 self.assertEqual(header_value in item, True) 503 item_found = True 504 self.assertTrue(item_found) 505 506 # Other headers combinaison 507 for headers, online_resource in ( 508 ({"Forwarded": "host=test3;proto=https"}, "https://test3"), 509 ({"Forwarded": "host=test4;proto=https, host=test5;proto=https"}, "https://test4"), 510 ({"X-Forwarded-Host": "test6", "X-Forwarded-Proto": "https"}, "https://test6"), 511 ({"Host": "test7"}, "test7"), 512 ): 513 # empty url in project 514 project = os.path.join(self.testdata_path, "test_project_without_urls.qgs") 515 qs = "?" + "&".join(["%s=%s" % i for i in list({ 516 "MAP": urllib.parse.quote(project), 517 "SERVICE": "WCS", 518 "VERSION": "1.0.0", 519 "REQUEST": "GetCapabilities", 520 "STYLES": "" 521 }.items())]) 522 523 r, h = self._result(self._execute_request(qs, request_headers=headers)) 524 525 item_found = False 526 for item in str(r).split("\\n"): 527 if "OnlineResource" in item: 528 self.assertEqual(online_resource in item, True) 529 item_found = True 530 self.assertTrue(item_found) 531 532 533if __name__ == '__main__': 534 unittest.main() 535