1# -*- coding: utf-8 -*- 2"""QGIS Unit tests for QgsNetworkContentFetcherTask 3 4.. note:: This program is free software; you can redistribute it and/or modify 5it under the terms of the GNU General Public License as published by 6the Free Software Foundation; either version 2 of the License, or 7(at your option) any later version. 8""" 9 10from builtins import chr 11from builtins import str 12 13__author__ = 'Nyall Dawson' 14__date__ = '29/03/2018' 15__copyright__ = 'Copyright 2018, The QGIS Project' 16 17import qgis # NOQA 18 19import os 20from qgis.testing import unittest, start_app 21from qgis.core import QgsNetworkContentFetcher, QgsNetworkContentFetcherTask, QgsApplication 22from utilities import unitTestDataPath 23from qgis.PyQt.QtCore import QUrl 24from qgis.PyQt.QtNetwork import QNetworkReply, QNetworkRequest 25import socketserver 26import threading 27import http.server 28 29app = start_app() 30 31 32class TestQgsNetworkContentFetcherTask(unittest.TestCase): 33 34 @classmethod 35 def setUpClass(cls): 36 # Bring up a simple HTTP server 37 os.chdir(unitTestDataPath() + '') 38 handler = http.server.SimpleHTTPRequestHandler 39 40 cls.httpd = socketserver.TCPServer(('localhost', 0), handler) 41 cls.port = cls.httpd.server_address[1] 42 43 cls.httpd_thread = threading.Thread(target=cls.httpd.serve_forever) 44 cls.httpd_thread.setDaemon(True) 45 cls.httpd_thread.start() 46 47 def __init__(self, methodName): 48 """Run once on class initialization.""" 49 unittest.TestCase.__init__(self, methodName) 50 51 self.loaded = False 52 53 def contentLoaded(self): 54 self.loaded = True 55 56 def testFetchBadUrl(self): 57 fetcher = QgsNetworkContentFetcherTask(QUrl('http://x')) 58 self.loaded = False 59 60 def check_reply(): 61 r = fetcher.reply() 62 assert r.error() != QNetworkReply.NoError 63 self.loaded = True 64 65 fetcher.fetched.connect(check_reply) 66 QgsApplication.taskManager().addTask(fetcher) 67 while not self.loaded: 68 app.processEvents() 69 70 def testFetchUrlContent(self): 71 fetcher = QgsNetworkContentFetcherTask( 72 QUrl('http://localhost:' + str(self.port) + '/qgis_local_server/index.html')) 73 self.loaded = False 74 75 def check_reply(): 76 r = fetcher.reply() 77 assert r.error() == QNetworkReply.NoError, r.error() 78 79 assert b'QGIS' in r.readAll() 80 self.loaded = True 81 82 fetcher.fetched.connect(check_reply) 83 QgsApplication.taskManager().addTask(fetcher) 84 while not self.loaded: 85 app.processEvents() 86 87 88if __name__ == "__main__": 89 unittest.main() 90