This repository has been archived by the owner on Dec 7, 2022. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 27
/
http_static_test_server.py
68 lines (51 loc) · 1.87 KB
/
http_static_test_server.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
# -*- coding: utf-8 -*-
"""
HTTP test server for writing tests against an "external" server.
"""
import atexit
import socket
import threading
from BaseHTTPServer import HTTPServer
from SimpleHTTPServer import SimpleHTTPRequestHandler
class HTTPServerIPV6(HTTPServer):
address_family = socket.AF_INET6
class HTTPStaticTestServer(object):
"""
Static test server that server files from the local directory and
sub-directories. Highly recommended that each test suite puts the files it
wants served into a custom sub-directory under 'data/'. Then tests can
reach the files by using the url:
http://localhost:8088/data/<custom-sub-directory>/<file>
This server is run in a thread over the local loopback and should be
started and stopped on each test or test suite run.
It's recommended that start be put in setUpClass and stop be put in
tearDownClass to avoid the overhead of starting and stopping on each test
run.
"""
def __init__(self, port=8088):
self.server = HTTPServerIPV6(('', port), SimpleHTTPRequestHandler)
self.server.timeout = 0.1 # timeout after a tenth of a second
self._is_running = False
self._server_thread = None
_SERVERS.append(self)
def _serve(self):
while self._is_running:
self.server.handle_request()
def start(self):
self._is_running = True
self._server_thread = threading.Thread(target=self._serve)
self._server_thread.setDaemon(True)
self._server_thread.start()
def stop(self):
self._is_running = False
self._server_thread.join()
self._server_thread = None
_SERVERS.remove(self)
_SERVERS = []
def _cleanup_servers():
"""
Cleanup all of the running server in case we were ctrl+c'd
"""
for server in _SERVERS:
server.stop()
atexit.register(_cleanup_servers)