Bug 1650049 [wpt PR 24422] - Python 3: port tests of content-type, http-cache, metadata, origin in …, a=testonly

Automatic update from web-platform-tests
Python 3: port tests of content-type, http-cache, metadata, origin in fetch (#24422)

--

wpt-commits: ecd73dd12d21e70208f8917079be6f1baada0168
wpt-pr: 24422
This commit is contained in:
ziransun 2020-07-17 11:12:46 +00:00 коммит произвёл moz-wptsync-bot
Родитель 2d49731661
Коммит 484f04acff
9 изменённых файлов: 187 добавлений и 174 удалений

Просмотреть файл

@ -1,15 +1,17 @@
from wptserve.utils import isomorphic_encode
def main(request, response):
values = request.GET.get_list("value")
content = request.GET.first("content", "<b>hi</b>\n")
output = "HTTP/1.1 200 OK\r\n"
output += "X-Content-Type-Options: nosniff\r\n"
if "single_header" in request.GET:
output += "Content-Type: " + ",".join(values) + "\r\n"
values = request.GET.get_list(b"value")
content = request.GET.first(b"content", b"<b>hi</b>\n")
output = b"HTTP/1.1 200 OK\r\n"
output += b"X-Content-Type-Options: nosniff\r\n"
if b"single_header" in request.GET:
output += b"Content-Type: " + b",".join(values) + b"\r\n"
else:
for value in values:
output += "Content-Type: " + value + "\r\n"
output += "Content-Length: " + str(len(content)) + "\r\n"
output += "\r\n"
output += b"Content-Type: " + value + b"\r\n"
output += b"Content-Length: " + isomorphic_encode(str(len(content))) + b"\r\n"
output += b"\r\n"
output += content
response.writer.write(output)
response.close_connection = True

Просмотреть файл

@ -5,98 +5,100 @@ import json
import time
from base64 import b64decode
NOTEHDRS = set(['content-type', 'access-control-allow-origin', 'last-modified', 'etag'])
from wptserve.utils import isomorphic_decode, isomorphic_encode
NOTEHDRS = set([u'content-type', u'access-control-allow-origin', u'last-modified', u'etag'])
NOBODYSTATUS = set([204, 304])
LOCATIONHDRS = set(['location', 'content-location'])
DATEHDRS = set(['date', 'expires', 'last-modified'])
LOCATIONHDRS = set([u'location', u'content-location'])
DATEHDRS = set([u'date', u'expires', u'last-modified'])
def main(request, response):
dispatch = request.GET.first("dispatch", None)
uuid = request.GET.first("uuid", None)
dispatch = request.GET.first(b"dispatch", None)
uuid = request.GET.first(b"uuid", None)
if request.method == "OPTIONS":
if request.method == u"OPTIONS":
return handle_preflight(uuid, request, response)
if not uuid:
response.status = (404, "Not Found")
response.headers.set("Content-Type", "text/plain")
return "UUID not found"
if dispatch == 'test':
response.status = (404, b"Not Found")
response.headers.set(b"Content-Type", b"text/plain")
return b"UUID not found"
if dispatch == b'test':
return handle_test(uuid, request, response)
elif dispatch == 'state':
elif dispatch == b'state':
return handle_state(uuid, request, response)
response.status = (404, "Not Found")
response.headers.set("Content-Type", "text/plain")
return "Fallthrough"
response.status = (404, b"Not Found")
response.headers.set(b"Content-Type", b"text/plain")
return b"Fallthrough"
def handle_preflight(uuid, request, response):
response.status = (200, "OK")
response.headers.set("Access-Control-Allow-Origin", "*")
response.headers.set("Access-Control-Allow-Methods", "GET")
response.headers.set("Access-Control-Allow-Headers", "*")
response.headers.set("Access-Control-Max-Age", "86400")
return "Preflight request"
response.status = (200, b"OK")
response.headers.set(b"Access-Control-Allow-Origin", b"*")
response.headers.set(b"Access-Control-Allow-Methods", b"GET")
response.headers.set(b"Access-Control-Allow-Headers", b"*")
response.headers.set(b"Access-Control-Max-Age", b"86400")
return b"Preflight request"
def handle_state(uuid, request, response):
response.headers.set("Content-Type", "text/plain")
response.headers.set(b"Content-Type", b"text/plain")
return json.dumps(request.server.stash.take(uuid))
def handle_test(uuid, request, response):
server_state = request.server.stash.take(uuid) or []
try:
requests = json.loads(b64decode(request.headers.get('Test-Requests', "")))
requests = json.loads(b64decode(request.headers.get(b'Test-Requests', b"")))
except:
response.status = (400, "Bad Request")
response.headers.set("Content-Type", "text/plain")
return "No or bad Test-Requests request header"
response.status = (400, b"Bad Request")
response.headers.set(b"Content-Type", b"text/plain")
return b"No or bad Test-Requests request header"
config = requests[len(server_state)]
if not config:
response.status = (404, "Not Found")
response.headers.set("Content-Type", "text/plain")
return "Config not found"
response.status = (404, b"Not Found")
response.headers.set(b"Content-Type", b"text/plain")
return b"Config not found"
noted_headers = {}
now = time.time()
for header in config.get('response_headers', []):
for header in config.get(u'response_headers', []):
if header[0].lower() in LOCATIONHDRS: # magic locations
if (len(header[1]) > 0):
header[1] = "%s&target=%s" % (request.url, header[1])
header[1] = u"%s&target=%s" % (request.url, header[1])
else:
header[1] = request.url
if header[0].lower() in DATEHDRS and isinstance(header[1], int): # magic dates
header[1] = http_date(now, header[1])
response.headers.set(header[0], header[1])
response.headers.set(isomorphic_encode(header[0]), isomorphic_encode(header[1]))
if header[0].lower() in NOTEHDRS:
noted_headers[header[0].lower()] = header[1]
state = {
'now': now,
'request_method': request.method,
'request_headers': dict([[h.lower(), request.headers[h]] for h in request.headers]),
'response_headers': noted_headers
u'now': now,
u'request_method': request.method,
u'request_headers': dict([[isomorphic_decode(h.lower()), isomorphic_decode(request.headers[h])] for h in request.headers]),
u'response_headers': noted_headers
}
server_state.append(state)
request.server.stash.put(uuid, server_state)
if "access-control-allow-origin" not in noted_headers:
response.headers.set("Access-Control-Allow-Origin", "*")
if "content-type" not in noted_headers:
response.headers.set("Content-Type", "text/plain")
response.headers.set("Server-Request-Count", len(server_state))
if u"access-control-allow-origin" not in noted_headers:
response.headers.set(b"Access-Control-Allow-Origin", b"*")
if u"content-type" not in noted_headers:
response.headers.set(b"Content-Type", b"text/plain")
response.headers.set(b"Server-Request-Count", len(server_state))
code, phrase = config.get("response_status", [200, "OK"])
if config.get("expected_type", "").endswith('validated'):
ref_hdrs = server_state[0]['response_headers']
previous_lm = ref_hdrs.get('last-modified', False)
if previous_lm and request.headers.get("If-Modified-Since", False) == previous_lm:
code, phrase = [304, "Not Modified"]
previous_etag = ref_hdrs.get('etag', False)
if previous_etag and request.headers.get("If-None-Match", False) == previous_etag:
code, phrase = [304, "Not Modified"]
code, phrase = config.get(u"response_status", [200, b"OK"])
if config.get(u"expected_type", u"").endswith(u'validated'):
ref_hdrs = server_state[0][u'response_headers']
previous_lm = ref_hdrs.get(u'last-modified', False)
if previous_lm and request.headers.get(b"If-Modified-Since", False) == isomorphic_encode(previous_lm):
code, phrase = [304, b"Not Modified"]
previous_etag = ref_hdrs.get(u'etag', False)
if previous_etag and request.headers.get(b"If-None-Match", False) == isomorphic_encode(previous_etag):
code, phrase = [304, b"Not Modified"]
if code != 304:
code, phrase = [999, '304 Not Generated']
code, phrase = [999, b'304 Not Generated']
response.status = (code, phrase)
content = config.get("response_body", uuid)
content = config.get(u"response_body", uuid)
if code in NOBODYSTATUS:
return ""
return b""
return content
@ -107,13 +109,13 @@ def get_header(headers, header_name):
result = header[1]
return result
WEEKDAYS = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']
MONTHS = [None, 'Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul',
'Aug', 'Sep', 'Oct', 'Nov', 'Dec']
WEEKDAYS = [u'Mon', u'Tue', u'Wed', u'Thu', u'Fri', u'Sat', u'Sun']
MONTHS = [None, u'Jan', u'Feb', u'Mar', u'Apr', u'May', u'Jun', u'Jul',
u'Aug', u'Sep', u'Oct', u'Nov', u'Dec']
def http_date(now, delta_secs=0):
date = datetime.datetime.utcfromtimestamp(now + delta_secs)
return "%s, %.2d %s %.4d %.2d:%.2d:%.2d GMT" % (
return u"%s, %.2d %s %.4d %.2d:%.2d:%.2d GMT" % (
WEEKDAYS[date.weekday()],
date.day,
MONTHS[date.month],

Просмотреть файл

@ -1,17 +1,19 @@
# -*- coding: utf-8 -
def main(request, response):
image_url = str.replace(request.url, "fetch/http-cache/resources/securedimage.py", "images/green.png")
from wptserve.utils import isomorphic_decode, isomorphic_encode
if "authorization" not in request.headers:
def main(request, response):
image_url = str.replace(request.url, u"fetch/http-cache/resources/securedimage.py", u"images/green.png")
if b"authorization" not in request.headers:
response.status = 401
response.headers.set("WWW-Authenticate", "Basic")
response.headers.set(b"WWW-Authenticate", b"Basic")
return
else:
auth = request.headers.get("Authorization")
if auth != "Basic dGVzdHVzZXI6dGVzdHBhc3M=":
response.set_error(403, "Invalid username or password - " + auth)
auth = request.headers.get(b"Authorization")
if auth != b"Basic dGVzdHVzZXI6dGVzdHBhc3M=":
response.set_error(403, u"Invalid username or password - " + isomorphic_decode(auth))
return
response.status = 301
response.headers.set("Location", image_url)
response.headers.set(b"Location", isomorphic_encode(image_url))

Просмотреть файл

@ -1,27 +1,29 @@
import json
from wptserve.utils import isomorphic_decode
def main(request, response):
headers = [("Content-Type", "application/json"),
("Access-Control-Allow-Credentials", "true")]
headers = [(b"Content-Type", b"application/json"),
(b"Access-Control-Allow-Credentials", b"true")]
if "origin" in request.headers:
headers.append(("Access-Control-Allow-Origin", request.headers["origin"]))
if b"origin" in request.headers:
headers.append((b"Access-Control-Allow-Origin", request.headers[b"origin"]))
body = ""
body = u""
# If we're in a preflight, verify that `Sec-Fetch-Mode` is `cors`.
if request.method == 'OPTIONS':
if request.headers.get("sec-fetch-mode") != "cors":
return (403, "Failed"), [], body
if request.method == u'OPTIONS':
if request.headers.get(b"sec-fetch-mode") != b"cors":
return (403, b"Failed"), [], body
headers.append(("Access-Control-Allow-Methods", "*"))
headers.append(("Access-Control-Allow-Headers", "*"))
headers.append((b"Access-Control-Allow-Methods", b"*"))
headers.append((b"Access-Control-Allow-Headers", b"*"))
else:
body = json.dumps({
"dest": request.headers.get("sec-fetch-dest", ""),
"mode": request.headers.get("sec-fetch-mode", ""),
"site": request.headers.get("sec-fetch-site", ""),
"user": request.headers.get("sec-fetch-user", ""),
u"dest": isomorphic_decode(request.headers.get(b"sec-fetch-dest", b"")),
u"mode": isomorphic_decode(request.headers.get(b"sec-fetch-mode", b"")),
u"site": isomorphic_decode(request.headers.get(b"sec-fetch-site", b"")),
u"user": isomorphic_decode(request.headers.get(b"sec-fetch-user", b"")),
})
return headers, body

Просмотреть файл

@ -1,13 +1,14 @@
import json
def main(request, response):
headers = [("Content-Type", "text/javascript")]
from wptserve.utils import isomorphic_decode
body = "var header = %s;" % json.dumps({
"dest": request.headers.get("sec-fetch-dest", ""),
"mode": request.headers.get("sec-fetch-mode", ""),
"site": request.headers.get("sec-fetch-site", ""),
"user": request.headers.get("sec-fetch-user", ""),
});
def main(request, response):
headers = [(b"Content-Type", b"text/javascript")]
body = u"var header = %s;" % json.dumps({
u"dest": isomorphic_decode(request.headers.get(b"sec-fetch-dest", b"")),
u"mode": isomorphic_decode(request.headers.get(b"sec-fetch-mode", b"")),
u"site": isomorphic_decode(request.headers.get(b"sec-fetch-site", b"")),
u"user": isomorphic_decode(request.headers.get(b"sec-fetch-user", b"")),
})
return headers, body

Просмотреть файл

@ -1,12 +1,14 @@
import json
from wptserve.utils import isomorphic_decode
def main(request, response):
headers = [
("Content-Type", "text/html"),
("Cache-Control", "no-cache, no-store, must-revalidate")
(b"Content-Type", b"text/html"),
(b"Cache-Control", b"no-cache, no-store, must-revalidate")
]
body = """
body = u"""
<!DOCTYPE html>
<script>
var data = %s;
@ -18,9 +20,9 @@ def main(request, response):
window.portalHost.postMessage(data, "*");
</script>
""" % json.dumps({
"dest": request.headers.get("sec-fetch-dest", ""),
"mode": request.headers.get("sec-fetch-mode", ""),
"site": request.headers.get("sec-fetch-site", ""),
"user": request.headers.get("sec-fetch-user", ""),
u"dest": isomorphic_decode(request.headers.get(b"sec-fetch-dest", b"")),
u"mode": isomorphic_decode(request.headers.get(b"sec-fetch-mode", b"")),
u"site": isomorphic_decode(request.headers.get(b"sec-fetch-site", b"")),
u"user": isomorphic_decode(request.headers.get(b"sec-fetch-user", b"")),
})
return headers, body

Просмотреть файл

@ -1,29 +1,29 @@
import os
import uuid
import hashlib
import time
import json
from wptserve.utils import isomorphic_decode
def main(request, response):
## Get the query parameter (key) from URL ##
## Tests will record POST requests (CSP Report) and GET (rest) ##
if request.GET:
key = request.GET['file']
key = request.GET[b'file']
elif request.POST:
key = request.POST['file']
key = request.POST[b'file']
## Convert the key from String to UUID valid String ##
testId = hashlib.md5(key).hexdigest()
## Handle the header retrieval request ##
if 'retrieve' in request.GET:
if b'retrieve' in request.GET:
response.writer.write_status(200)
response.writer.end_headers()
try:
header_value = request.server.stash.take(testId)
response.writer.write(header_value)
except (KeyError, ValueError) as e:
response.writer.write("No header has been recorded")
response.writer.write(u"No header has been recorded")
pass
response.close_connection = True
@ -34,10 +34,10 @@ def main(request, response):
## Return a serialized JSON object with one member per header. If the ##
## header isn't present, the member will contain an empty string. ##
header = json.dumps({
"dest": request.headers.get("sec-fetch-dest", ""),
"mode": request.headers.get("sec-fetch-mode", ""),
"site": request.headers.get("sec-fetch-site", ""),
"user": request.headers.get("sec-fetch-user", ""),
u"dest": isomorphic_decode(request.headers.get(b"sec-fetch-dest", b"")),
u"mode": isomorphic_decode(request.headers.get(b"sec-fetch-mode", b"")),
u"site": isomorphic_decode(request.headers.get(b"sec-fetch-site", b"")),
u"user": isomorphic_decode(request.headers.get(b"sec-fetch-user", b"")),
})
request.server.stash.put(testId, header)
except KeyError:
@ -45,101 +45,101 @@ def main(request, response):
pass
## Prevent the browser from caching returned responses and allow CORS ##
response.headers.set("Access-Control-Allow-Origin", "*")
response.headers.set("Cache-Control", "no-cache, no-store, must-revalidate")
response.headers.set("Pragma", "no-cache")
response.headers.set("Expires", "0")
response.headers.set(b"Access-Control-Allow-Origin", b"*")
response.headers.set(b"Cache-Control", b"no-cache, no-store, must-revalidate")
response.headers.set(b"Pragma", b"no-cache")
response.headers.set(b"Expires", b"0")
## Add a valid ServiceWorker Content-Type ##
if key.startswith("serviceworker"):
response.headers.set("Content-Type", "application/javascript")
if key.startswith(b"serviceworker"):
response.headers.set(b"Content-Type", b"application/javascript")
## Add a valid image Content-Type ##
if key.startswith("image"):
response.headers.set("Content-Type", "image/png")
file = open(os.path.join(request.doc_root, "media", "1x1-green.png"), "r")
if key.startswith(b"image"):
response.headers.set(b"Content-Type", b"image/png")
file = open(os.path.join(request.doc_root, u"media", u"1x1-green.png"), u"rb")
image = file.read()
file.close()
return image
## Return a valid .vtt content for the <track> tag ##
if key.startswith("track"):
return "WEBVTT"
if key.startswith(b"track"):
return b"WEBVTT"
## Return a valid SharedWorker ##
if key.startswith("sharedworker"):
response.headers.set("Content-Type", "application/javascript")
file = open(os.path.join(request.doc_root, "fetch", "metadata",
"resources", "sharedWorker.js"), "r")
if key.startswith(b"sharedworker"):
response.headers.set(b"Content-Type", b"application/javascript")
file = open(os.path.join(request.doc_root, u"fetch", u"metadata",
u"resources", u"sharedWorker.js"), u"rb")
shared_worker = file.read()
file.close()
return shared_worker
## Return a valid font content and Content-Type ##
if key.startswith("font"):
response.headers.set("Content-Type", "application/x-font-ttf")
file = open(os.path.join(request.doc_root, "fonts", "Ahem.ttf"), "r")
if key.startswith(b"font"):
response.headers.set(b"Content-Type", b"application/x-font-ttf")
file = open(os.path.join(request.doc_root, u"fonts", u"Ahem.ttf"), u"rb")
font = file.read()
file.close()
return font
## Return a valid audio content and Content-Type ##
if key.startswith("audio"):
response.headers.set("Content-Type", "audio/mpeg")
file = open(os.path.join(request.doc_root, "media", "sound_5.mp3"), "r")
if key.startswith(b"audio"):
response.headers.set(b"Content-Type", b"audio/mpeg")
file = open(os.path.join(request.doc_root, u"media", u"sound_5.mp3"), u"rb")
audio = file.read()
file.close()
return audio
## Return a valid video content and Content-Type ##
if key.startswith("video"):
response.headers.set("Content-Type", "video/mp4")
file = open(os.path.join(request.doc_root, "media", "A4.mp4"), "r")
if key.startswith(b"video"):
response.headers.set(b"Content-Type", b"video/mp4")
file = open(os.path.join(request.doc_root, u"media", u"A4.mp4"), u"rb")
video = file.read()
file.close()
return video
## Return valid style content and Content-Type ##
if key.startswith("style"):
response.headers.set("Content-Type", "text/css")
return "div { }"
if key.startswith(b"style"):
response.headers.set(b"Content-Type", b"text/css")
return b"div { }"
## Return a valid embed/object content and Content-Type ##
if key.startswith("embed") or key.startswith("object"):
response.headers.set("Content-Type", "text/html")
return "<html>EMBED!</html>"
if key.startswith(b"embed") or key.startswith(b"object"):
response.headers.set(b"Content-Type", b"text/html")
return b"<html>EMBED!</html>"
## Return a valid image content and Content-Type for redirect requests ##
if key.startswith("redirect"):
response.headers.set("Content-Type", "image/jpeg")
file = open(os.path.join(request.doc_root, "media", "1x1-green.png"), "r")
if key.startswith(b"redirect"):
response.headers.set(b"Content-Type", b"image/jpeg")
file = open(os.path.join(request.doc_root, u"media", u"1x1-green.png"), u"rb")
image = file.read()
file.close()
return image
## Return a valid dedicated worker
if key.startswith("worker"):
response.headers.set("Content-Type", "application/javascript")
return "self.postMessage('loaded');"
if key.startswith(b"worker"):
response.headers.set(b"Content-Type", b"application/javascript")
return b"self.postMessage('loaded');"
## Return an appcache manifest
if key.startswith("appcache-manifest"):
response.headers.set("Content-Type", "text/cache-manifest")
return """CACHE MANIFEST
if key.startswith(b"appcache-manifest"):
response.headers.set(b"Content-Type", b"text/cache-manifest")
return b"""CACHE MANIFEST
/fetch/metadata/resources/record-header.py?file=appcache-resource%s
NETWORK:
*""" % key[17:]
## Return an appcache resource
if key.startswith("appcache-resource"):
response.headers.set("Content-Type", "text/html")
return "<html>Appcache!</html>"
if key.startswith(b"appcache-resource"):
response.headers.set(b"Content-Type", b"text/html")
return b"<html>Appcache!</html>"
## Return a valid XSLT
if key.startswith("xslt"):
response.headers.set("Content-Type", "text/xsl")
return """<?xml version="1.0" encoding="UTF-8"?>
if key.startswith(b"xslt"):
response.headers.set(b"Content-Type", b"text/xsl")
return b"""<?xml version="1.0" encoding="UTF-8"?>
<xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform" version="1.0">
<xsl:template match="@*|node()">
<xsl:copy>

Просмотреть файл

@ -1,30 +1,32 @@
import json
from wptserve.utils import isomorphic_decode
def main(request, response):
key = request.GET.first("stash")
origin = request.headers.get("origin")
key = request.GET.first(b"stash")
origin = request.headers.get(b"origin")
if origin is None:
origin = "no Origin header"
origin = b"no Origin header"
origin_list = request.server.stash.take(key)
if "dump" in request.GET:
response.headers.set("Content-Type", "application/json")
if b"dump" in request.GET:
response.headers.set(b"Content-Type", b"application/json")
response.content = json.dumps(origin_list)
return
if origin_list is None:
origin_list = [origin]
origin_list = [isomorphic_decode(origin)]
else:
origin_list.append(origin)
origin_list.append(isomorphic_decode(origin))
request.server.stash.put(key, origin_list)
if "location" in request.GET:
if b"location" in request.GET:
response.status = 308
response.headers.set("Location", request.GET.first("location"))
response.headers.set(b"Location", request.GET.first(b"location"))
return
response.headers.set("Content-Type", "text/html")
response.headers.set("Access-Control-Allow-Origin", "*")
response.content = "<meta charset=utf-8>\n<body><script>parent.postMessage('loaded','*')</script></body>"
response.headers.set(b"Content-Type", b"text/html")
response.headers.set(b"Access-Control-Allow-Origin", b"*")
response.content = b"<meta charset=utf-8>\n<body><script>parent.postMessage('loaded','*')</script></body>"

Просмотреть файл

@ -1,7 +1,7 @@
def main(request, response):
if "referrerPolicy" in request.GET:
response.headers.set("Referrer-Policy",
request.GET.first("referrerPolicy"))
if b"referrerPolicy" in request.GET:
response.headers.set(b"Referrer-Policy",
request.GET.first(b"referrerPolicy"))
response.status = 200
response.headers.set("Content-Type", "text/html")
response.content = "<meta charset=utf-8>\n<body><script>parent.postMessage('action','*')</script></body>"
response.headers.set(b"Content-Type", b"text/html")
response.content = b"<meta charset=utf-8>\n<body><script>parent.postMessage('action','*')</script></body>"