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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
|
#!/usr/bin/env python
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
"""Specialisation of wptserver.server.WebTestHttpd for testing
Marionette.
"""
import argparse
import os
import select
import sys
import time
from six.moves.urllib import parse as urlparse
from wptserve import handlers, request, server
from wptserve import routes as default_routes
root = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
default_doc_root = os.path.join(root, "www")
default_ssl_cert = os.path.join(root, "certificates", "test.cert")
default_ssl_key = os.path.join(root, "certificates", "test.key")
@handlers.handler
def http_auth_handler(req, response):
# Allow the test to specify the username and password
params = dict(urlparse.parse_qsl(req.url_parts.query))
username = params.get("username", "guest")
password = params.get("password", "guest")
auth = request.Authentication(req.headers)
content = """<!doctype html>
<title>HTTP Authentication</title>
<p id="status">{}</p>"""
if auth.username == username and auth.password == password:
response.status = 200
response.content = content.format("success")
else:
response.status = 401
response.headers.set("WWW-Authenticate", 'Basic realm="secret"')
response.content = content.format("restricted")
@handlers.handler
def upload_handler(request, response):
return 200, [], [request.headers.get("Content-Type")] or []
@handlers.handler
def slow_loading_handler(request, response):
# Allow the test specify the delay for delivering the content
params = dict(urlparse.parse_qsl(request.url_parts.query))
delay = int(params.get("delay", 5))
time.sleep(delay)
# Do not allow the page to be cached to circumvent the bfcache of the browser
response.headers.set("Cache-Control", "no-cache, no-store")
response.content = """<!doctype html>
<meta charset="UTF-8">
<title>Slow page loading</title>
<p>Delay: <span id="delay">{}</span></p>
""".format(
delay
)
@handlers.handler
def slow_coop_handler(request, response):
# Allow the test specify the delay for delivering the content
params = dict(urlparse.parse_qsl(request.url_parts.query))
delay = int(params.get("delay", 5))
time.sleep(delay)
# Isolate the browsing context exclusively to same-origin documents
response.headers.set("Cross-Origin-Opener-Policy", "same-origin")
response.headers.set("Cache-Control", "no-cache, no-store")
response.content = """<!doctype html>
<meta charset="UTF-8">
<title>Slow cross-origin page loading</title>
<p>Delay: <span id="delay">{}</span></p>
""".format(
delay
)
@handlers.handler
def update_xml_handler(request, response):
response.headers.set("Content-Type", "text/xml")
mar_digest = (
"75cd68e6c98c84c435cd27e353f5b4f6a3f2c50f6802aa9bf62b47e47138757306769fd9befa08793635ee649"
"2319253480860b4aa8ed9ee1caaa4c83ebc90b9"
)
response.content = """
<updates>
<update type="minor" displayVersion="9999.0" appVersion="9999.0" platformVersion="9999.0"
buildID="20220627075547">
<patch type="complete" URL="{}://{}/update/complete.mar" size="86612"
hashFunction="sha512" hashValue="{}"/>
</update>
</updates>
""".format(
request.url_parts.scheme, request.url_parts.netloc, mar_digest
)
class NotAliveError(Exception):
"""Occurs when attempting to run a function that requires the HTTPD
to have been started, and it has not.
"""
pass
class FixtureServer(object):
def __init__(
self,
doc_root,
url="http://127.0.0.1:0",
use_ssl=False,
ssl_cert=None,
ssl_key=None,
):
if not os.path.isdir(doc_root):
raise ValueError("Server root is not a directory: %s" % doc_root)
url = urlparse.urlparse(url)
if url.scheme is None:
raise ValueError("Server scheme not provided")
scheme, host, port = url.scheme, url.hostname, url.port
if host is None:
host = "127.0.0.1"
if port is None:
port = 0
routes = [
("POST", "/file_upload", upload_handler),
("GET", "/http_auth", http_auth_handler),
("GET", "/slow", slow_loading_handler),
("GET", "/slow-coop", slow_coop_handler),
("GET", "/update.xml", update_xml_handler),
]
routes.extend(default_routes.routes)
self._httpd = server.WebTestHttpd(
host=host,
port=port,
bind_address=True,
doc_root=doc_root,
routes=routes,
use_ssl=True if scheme == "https" else False,
certificate=ssl_cert,
key_file=ssl_key,
)
def start(self):
if self.is_alive:
return
self._httpd.start()
def wait(self):
if not self.is_alive:
return
try:
select.select([], [], [])
except KeyboardInterrupt:
self.stop()
def stop(self):
if not self.is_alive:
return
self._httpd.stop()
def get_url(self, path):
if not self.is_alive:
raise NotAliveError()
return self._httpd.get_url(path)
@property
def doc_root(self):
return self._httpd.router.doc_root
@property
def router(self):
return self._httpd.router
@property
def routes(self):
return self._httpd.router.routes
@property
def is_alive(self):
return self._httpd.started
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Specialised HTTP server for testing Marionette."
)
parser.add_argument(
"url",
help="""
service address including scheme, hostname, port, and prefix for document root,
e.g. \"https://0.0.0.0:0/base/\"""",
)
parser.add_argument(
"-r",
dest="doc_root",
default=default_doc_root,
help="path to document root (default %(default)s)",
)
parser.add_argument(
"-c",
dest="ssl_cert",
default=default_ssl_cert,
help="path to SSL certificate (default %(default)s)",
)
parser.add_argument(
"-k",
dest="ssl_key",
default=default_ssl_key,
help="path to SSL certificate key (default %(default)s)",
)
args = parser.parse_args()
httpd = FixtureServer(
args.doc_root, args.url, ssl_cert=args.ssl_cert, ssl_key=args.ssl_key
)
httpd.start()
print(
"{0}: started fixture server on {1}".format(sys.argv[0], httpd.get_url("/")),
file=sys.stderr,
)
httpd.wait()
|
