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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
import threading
from twisted.internet import reactor
from twisted.web import resource, server, guard
from twisted.cred.portal import IRealm, Portal
from twisted.cred.checkers import InMemoryUsernamePasswordDatabaseDontUse
from zope.interface import implementer
server_thread = None
# Start the local HTTP server before the tests start.
def pytest_configure(config):
global server_thread
server_thread = _runHttpServer()
# Stop the local HTTP server after the tests are done.
def pytest_unconfigure(config):
global server_thread
reactor.callLater(0, reactor.stop)
server_thread.join()
#
# Run a local HTTP server for testing the HttpRequestManager.
# This server is accessible via http://localhost:8080/, and it has the following URL resources:
# - /success - This request will respond to a GET request normally with an HTML page.
# - /timeout - This request will respond to a GET request first with a few bytes and then do nothing.
# It is intended to emulate a timeout.
# - /auth - Respond to GET requests and requires an authentication. You need to use Basic Auth with user:user as
# the username and password to access this resource.
#
def _runHttpServer() -> "threading.Thread":
class RootResource(resource.Resource):
isLeaf = False
def __init__(self):
super().__init__()
self._success = SuccessfulResource()
self._timeout = TimeoutResource()
checkers = [InMemoryUsernamePasswordDatabaseDontUse(user = b"user")]
portal = Portal(SimpleRealm(), checkers)
self._auth_resource = guard.HTTPAuthSessionWrapper(portal, [guard.BasicCredentialFactory("auth")])
def getChild(self, name, request):
if name == b"success":
return self._success
elif name == b"timeout":
return self._timeout
elif name == b"auth":
return self._auth_resource
return resource.Resource.getChild(self, name, request)
# This resource can be accessed via /success. It will reply with an HTML page.
class SuccessfulResource(resource.Resource):
isLeaf = True
def getChild(self, name, request):
if name == b"":
return self
return resource.Resource.getChild(self, name, request)
def render_GET(self, request):
return "<html>Hello, world!</html>".encode("utf-8")
# This resource can be accessed via /timeout. It will first send 3 replies, with 1 second between each. After that
# It will not do anything to emulate a problem on the server side. This way we can test if the client side can
# handle this case correctly, such as with a timeout.
class TimeoutResource(resource.Resource):
isLeaf = True
def __init__(self):
super().__init__()
self._count = 0
def getChild(self, name, request):
if name == b"":
return self
return resource.Resource.getChild(self, name, request)
def render_GET(self, request):
self._process(request)
return server.NOT_DONE_YET
def _process(self, request):
if self._count > 10:
request.finish()
if self._count == 3:
# emulate a timeout by not replying anything
return
data = ("%s" % self._count) * 10
request.write(data.encode("utf-8"))
self._count += 1
reactor.callLater(0.5, self._process, request)
# A realm which gives out L{SuccessfulResource} instances for authenticated users.
@implementer(IRealm)
class SimpleRealm(object):
def requestAvatar(self, avatar_id, mind, *interfaces):
if resource.IResource in interfaces:
return resource.IResource, SuccessfulResource(), lambda: None
raise NotImplementedError()
def _runServer():
site = server.Site(RootResource())
reactor.listenTCP(8080, site)
reactor.run(installSignalHandlers= 0)
t = threading.Thread(target = _runServer)
t.start()
return t
|
