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