1############################################################################# 2## 3## Copyright (C) 2016 The Qt Company Ltd. 4## Contact: https://www.qt.io/licensing/ 5## 6## This file is part of the test suite of Qt for Python. 7## 8## $QT_BEGIN_LICENSE:GPL-EXCEPT$ 9## Commercial License Usage 10## Licensees holding valid commercial Qt licenses may use this file in 11## accordance with the commercial license agreement provided with the 12## Software or, alternatively, in accordance with the terms contained in 13## a written agreement between you and The Qt Company. For licensing terms 14## and conditions see https://www.qt.io/terms-conditions. For further 15## information use the contact form at https://www.qt.io/contact-us. 16## 17## GNU General Public License Usage 18## Alternatively, this file may be used under the terms of the GNU 19## General Public License version 3 as published by the Free Software 20## Foundation with exceptions as appearing in the file LICENSE.GPL3-EXCEPT 21## included in the packaging of this file. Please review the following 22## information to ensure the GNU General Public License requirements will 23## be met: https://www.gnu.org/licenses/gpl-3.0.html. 24## 25## $QT_END_LICENSE$ 26## 27############################################################################# 28 29import os 30import random 31import select 32import sys 33import tempfile 34import threading 35 36sys.path.append(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "util")) 37 38import py3kcompat as py3k 39 40if py3k.IS_PY3K: 41 import socketserver as SocketServer 42 import http.server as BaseHTTPServer 43else: 44 import SocketServer 45 import BaseHTTPServer 46 47class TestHandler(BaseHTTPServer.BaseHTTPRequestHandler): 48 DATA = "PySide Server" 49 allow_reuse_address = True 50 51 def do_GET(self): 52 self.send_head() 53 self.wfile.write(TestHandler.DATA) 54 55 def do_HEAD(self): 56 self.send_head() 57 58 def send_head(self): 59 self.send_response(200) 60 self.send_header("Content-type", "text/plain") 61 self.send_header("Content-Length", str(len(TestHandler.DATA))) 62 self.end_headers() 63 64class TestSecureHandler(BaseHTTPServer.BaseHTTPRequestHandler): 65 DATA = "PySide" 66 allow_reuse_address = True 67 68 def do_GET(self): 69 self.send_head() 70 self.wfile.write(py3k.b(TestHandler.DATA)) 71 72 def do_HEAD(self): 73 self.send_head() 74 75 def send_head(self): 76 try: 77 handler = self.marshall_handler() 78 handler.do_request(self) 79 except: 80 self.send_response(401) 81 self.send_header("WWW-Authenticate", "Basic realm='Secure Area'") 82 self.send_header("Content-type", "text/plain") 83 self.send_header("Content-Length", str(len(TestHandler.DATA))) 84 self.end_headers() 85 86# Workaround for the missing shutdown method in python2.5 87class CompatTCPServer(SocketServer.TCPServer): 88 def __init__(self, server_address, RequestHandlerClass): 89 SocketServer.TCPServer.__init__(self, server_address, RequestHandlerClass) 90 91 self.isPy25 = sys.version_info[0] == 2 and sys.version_info[1] == 5 92 if self.isPy25: 93 self.__is_shut_down = threading.Event() 94 self.__serving = False 95 96 def serve_forever(self, poll_interval=0.5): 97 """Handle one request at a time until shutdown. 98 99 Polls for shutdown every poll_interval seconds. Ignores 100 self.timeout. If you need to do periodic tasks, do them in 101 another thread. 102 """ 103 if self.isPy25: 104 self.__serving = True 105 self.__is_shut_down.clear() 106 while self.__serving: 107 # XXX: Consider using another file descriptor or 108 # connecting to the socket to wake this up instead of 109 # polling. Polling reduces our responsiveness to a 110 # shutdown request and wastes cpu at all other times. 111 r, w, e = select.select([self], [], [], poll_interval) 112 if r: 113 self.py25_handle_request_noblock() 114 self.__is_shut_down.set() 115 else: 116 SocketServer.TCPServer.serve_forever(self, poll_interval) 117 118 def py25_handle_request_noblock(self): 119 """Handle one request, without blocking. 120 121 I assume that select.select has returned that the socket is 122 readable before this function was called, so there should be 123 no risk of blocking in get_request(). 124 """ 125 if self.isPy25: 126 try: 127 request, client_address = self.get_request() 128 except socket.error: 129 return 130 if self.verify_request(request, client_address): 131 try: 132 self.process_request(request, client_address) 133 except: 134 self.handle_error(request, client_address) 135 self.close_request(request) 136 137 def shutdown(self): 138 """Stops the serve_forever loop. 139 140 Blocks until the loop has finished. This must be called while 141 serve_forever() is running in another thread, or it will 142 deadlock. 143 """ 144 if self.isPy25: 145 self.__serving = False 146 if not self.__is_shut_down: 147 self.__is_shut_down.wait() 148 else: 149 SocketServer.TCPServer.shutdown(self) 150 151 152class TestServer(threading.Thread): 153 154 def __init__(self, secure=False): 155 threading.Thread.__init__(self) 156 157 self._port = int(os.getenv("PYSIDE_TESTSERVER_PORT") or 12321) 158 self.keep_running = True 159 160 if secure: 161 handle = TestSecureHandler 162 else: 163 handle = TestHandler 164 165 while True: 166 try: 167 self.httpd = CompatTCPServer(('' , self._port), handle) 168 break 169 except: 170 self._port = self._port + random.randint(1, 100) 171 172 def port(self): 173 return self._port 174 175 def run(self): 176 self.httpd.serve_forever() 177 178 def shutdown(self): 179 self.httpd.shutdown() 180 self.join() 181 182