mirror of
https://github.com/servo/servo.git
synced 2025-08-16 19:05:33 +01:00
Update web-platform-tests to revision 3b3585e368841b77caea8576fa56cef91c3fbdf0
This commit is contained in:
Ms2ger
parent
d00639c55f
commit
3b4f0ec0bb
541 changed files with 14609 additions and 3288 deletions
435
tests/wpt/web-platform-tests/annotation-protocol/tools/protocol-server.py
Executable file
435
tests/wpt/web-platform-tests/annotation-protocol/tools/protocol-server.py
Executable file
|
|
@ -0,0 +1,435 @@
|
|||
# protocol-server
|
||||
#
|
||||
# a reference implementation of the Web Annotation Protocol
|
||||
#
|
||||
# Developed by Benjamin Young (@bigbulehat) and Shane McCarron (@halindrome).
|
||||
# Sponsored by Spec-Ops (https://spec-ops.io)
|
||||
#
|
||||
# Copyright (c) 2016 Spec-Ops
|
||||
#
|
||||
# for license information, see http://www.w3.org/Consortium/Legal/2008/04-testsuite-copyright.html
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
here = os.path.abspath(os.path.split(__file__)[0])
|
||||
repo_root = os.path.abspath(os.path.join(here, os.pardir, os.pardir))
|
||||
|
||||
sys.path.insert(0, os.path.join(repo_root, "tools"))
|
||||
sys.path.insert(0, os.path.join(repo_root, "tools", "six"))
|
||||
sys.path.insert(0, os.path.join(repo_root, "tools", "html5lib"))
|
||||
sys.path.insert(0, os.path.join(repo_root, "tools", "wptserve"))
|
||||
sys.path.insert(0, os.path.join(repo_root, "tools", "pywebsocket", "src"))
|
||||
sys.path.insert(0, os.path.join(repo_root, "tools", "py"))
|
||||
sys.path.insert(0, os.path.join(repo_root, "tools", "pytest"))
|
||||
sys.path.insert(0, os.path.join(repo_root, "tools", "webdriver"))
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import urlparse
|
||||
import uuid
|
||||
|
||||
import wptserve
|
||||
|
||||
myprotocol = 'http'
|
||||
myhost = 'localhost'
|
||||
port = 8080
|
||||
doc_root = os.path.join(repo_root, "annotation-protocol", "files", "")
|
||||
container_path = doc_root + 'annotations/'
|
||||
|
||||
URIroot = myprotocol + '://' + myhost + ':{0}'.format(port)
|
||||
|
||||
per_page = 10
|
||||
|
||||
MEDIA_TYPE = 'application/ld+json; profile="http://www.w3.org/ns/anno.jsonld"'
|
||||
# Prefer header variants
|
||||
PREFER_MINIMAL_CONTAINER = "http://www.w3.org/ns/ldp#PreferMinimalContainer"
|
||||
PREFER_CONTAINED_IRIS = "http://www.w3.org/ns/oa#PreferContainedIRIs"
|
||||
PREFER_CONTAINED_DESCRIPTIONS = \
|
||||
"http://www.w3.org/ns/oa#PreferContainedDescriptions"
|
||||
|
||||
|
||||
# dictionary for annotations that we create on the fly
|
||||
tempAnnotations = {}
|
||||
|
||||
def extract_preference(prefer):
|
||||
"""Extracts the parameters from a Prefer header's value
|
||||
>>> extract_preferences('return=representation;include="http://www.w3.org/ns/ldp#PreferMinimalContainer http://www.w3.org/ns/oa#PreferContainedIRIs"')
|
||||
{"return": "representation", "include": ["http://www.w3.org/ns/ldp#PreferMinimalContainer", "http://www.w3.org/ns/oa#PreferContainedIRIs"]}
|
||||
"""
|
||||
obj = {}
|
||||
if prefer:
|
||||
params = prefer.split(';')
|
||||
for p in params:
|
||||
key, value = p.split('=')
|
||||
obj[key] = value.strip('"').split(' ')
|
||||
return obj
|
||||
|
||||
|
||||
def dump_json(obj):
|
||||
return json.dumps(obj, indent=4, sort_keys=True)
|
||||
|
||||
def add_cors_headers(resp):
|
||||
headers_file = doc_root + 'annotations/cors.headers'
|
||||
resp.headers.update(load_headers_from_file(headers_file))
|
||||
|
||||
def load_headers_from_file(path):
|
||||
headers = []
|
||||
with open(path, 'r') as header_file:
|
||||
data = header_file.read()
|
||||
headers = [tuple(item.strip() for item in line.split(":", 1))
|
||||
for line in data.splitlines() if line]
|
||||
return headers
|
||||
|
||||
def annotation_files():
|
||||
files = []
|
||||
for file in os.listdir(container_path):
|
||||
if file.endswith('.jsonld') or file.endswith('.json'):
|
||||
files.append(file)
|
||||
for item in list(tempAnnotations.keys()):
|
||||
files.append(item)
|
||||
return files
|
||||
|
||||
|
||||
def annotation_iris(skip=0):
|
||||
iris = []
|
||||
for filename in annotation_files():
|
||||
iris.append(URIroot + '/annotations/' + filename)
|
||||
return iris[skip:][:per_page]
|
||||
|
||||
|
||||
def annotations(skip=0):
|
||||
annotations = []
|
||||
files = annotation_files()
|
||||
for file in files:
|
||||
if file.startswith("temp-"):
|
||||
annotations.append(json.loads(tempAnnotations[file]))
|
||||
else:
|
||||
with open(container_path + file, 'r') as annotation:
|
||||
annotations.append(json.load(annotation))
|
||||
return annotations
|
||||
|
||||
|
||||
def total_annotations():
|
||||
return len(annotation_files())
|
||||
|
||||
|
||||
@wptserve.handlers.handler
|
||||
def collection_get(request, response):
|
||||
"""Annotation Collection handler. NOTE: This also routes paging requests"""
|
||||
|
||||
# Paginate if requested
|
||||
qs = urlparse.parse_qs(request.url_parts.query)
|
||||
if 'page' in qs:
|
||||
return page(request, response)
|
||||
|
||||
# stub collection
|
||||
collection_json = {
|
||||
"@context": [
|
||||
"http://www.w3.org/ns/anno.jsonld",
|
||||
"http://www.w3.org/ns/ldp.jsonld"
|
||||
],
|
||||
"id": URIroot + "/annotations/",
|
||||
"type": ["BasicContainer", "AnnotationCollection"],
|
||||
"total": 0,
|
||||
"label": "A Container for Web Annotations",
|
||||
"first": URIroot + "/annotations/?page=0"
|
||||
}
|
||||
|
||||
last_page = (total_annotations() / per_page) - 1
|
||||
collection_json['last'] = URIroot + "/annotations/?page={0}".format(last_page)
|
||||
|
||||
# Default Container format SHOULD be PreferContainedDescriptions
|
||||
preference = extract_preference(request.headers.get('Prefer'))
|
||||
if 'include' in preference:
|
||||
preference = preference['include']
|
||||
else:
|
||||
preference = None
|
||||
|
||||
collection_json['total'] = total_annotations()
|
||||
# TODO: calculate last page and add it's page number
|
||||
|
||||
if (qs.get('iris') and qs.get('iris')[0] is '1') \
|
||||
or (preference and PREFER_CONTAINED_IRIS in preference):
|
||||
return_iris = True
|
||||
else:
|
||||
return_iris = False
|
||||
|
||||
# only PreferContainedIRIs has unqiue content
|
||||
if return_iris:
|
||||
collection_json['id'] += '?iris=1'
|
||||
collection_json['first'] += '&iris=1'
|
||||
collection_json['last'] += '&iris=1'
|
||||
|
||||
if preference and PREFER_MINIMAL_CONTAINER not in preference:
|
||||
if return_iris:
|
||||
collection_json['first'] = annotation_iris()
|
||||
else:
|
||||
collection_json['first'] = annotations()
|
||||
|
||||
collection_headers_file = doc_root + 'annotations/collection.headers'
|
||||
add_cors_headers(response)
|
||||
response.headers.update(load_headers_from_file(collection_headers_file))
|
||||
# this one's unique per request
|
||||
response.headers.set('Content-Location', collection_json['id'])
|
||||
return dump_json(collection_json)
|
||||
|
||||
|
||||
@wptserve.handlers.handler
|
||||
def collection_head(request, response):
|
||||
container_path = doc_root + request.request_path
|
||||
if os.path.isdir(container_path):
|
||||
response.status = 200
|
||||
else:
|
||||
response.status = 404
|
||||
|
||||
add_cors_headers(response)
|
||||
headers_file = doc_root + 'annotations/collection.headers'
|
||||
for header, value in load_headers_from_file(headers_file):
|
||||
response.headers.append(header, value)
|
||||
|
||||
response.content = None
|
||||
|
||||
|
||||
@wptserve.handlers.handler
|
||||
def collection_options(request, response):
|
||||
container_path = doc_root + request.request_path
|
||||
if os.path.isdir(container_path):
|
||||
response.status = 200
|
||||
else:
|
||||
response.status = 404
|
||||
|
||||
add_cors_headers(response)
|
||||
headers_file = doc_root + 'annotations/collection.options.headers'
|
||||
for header, value in load_headers_from_file(headers_file):
|
||||
response.headers.append(header, value)
|
||||
|
||||
response.content = "Collection Options\n";
|
||||
|
||||
|
||||
def page(request, response):
|
||||
page_json = {
|
||||
"@context": "http://www.w3.org/ns/anno.jsonld",
|
||||
"id": URIroot + "/annotations/",
|
||||
"type": "AnnotationPage",
|
||||
"partOf": {
|
||||
"id": URIroot + "/annotations/",
|
||||
"total": 42023
|
||||
},
|
||||
"next": URIroot + "/annotations/",
|
||||
"items": [
|
||||
]
|
||||
}
|
||||
|
||||
add_cors_headers(response)
|
||||
headers_file = doc_root + 'annotations/collection.headers'
|
||||
response.headers.update(load_headers_from_file(headers_file))
|
||||
|
||||
qs = urlparse.parse_qs(request.url_parts.query)
|
||||
page_num = int(qs.get('page')[0])
|
||||
page_json['id'] += '?page={0}'.format(page_num)
|
||||
|
||||
total = total_annotations()
|
||||
so_far = (per_page * (page_num+1))
|
||||
remaining = total - so_far
|
||||
|
||||
if page_num != 0:
|
||||
page_json['prev'] = URIroot + '/annotations/?page={0}'.format(page_num-1)
|
||||
|
||||
page_json['partOf']['total'] = total
|
||||
|
||||
if remaining > per_page:
|
||||
page_json['next'] += '?page={0}'.format(page_num+1)
|
||||
else:
|
||||
del page_json['next']
|
||||
|
||||
if qs.get('iris') and qs.get('iris')[0] is '1':
|
||||
page_json['items'] = annotation_iris(so_far)
|
||||
page_json['id'] += '&iris=1'
|
||||
if 'prev' in page_json:
|
||||
page_json['prev'] += '&iris=1'
|
||||
if 'next' in page_json:
|
||||
page_json['next'] += '&iris=1'
|
||||
else:
|
||||
page_json['items'] = annotations(so_far)
|
||||
|
||||
return dump_json(page_json)
|
||||
|
||||
|
||||
@wptserve.handlers.handler
|
||||
def annotation_get(request, response):
|
||||
"""Individual Annotations"""
|
||||
requested_file = doc_root + request.request_path[1:]
|
||||
base = os.path.basename( requested_file )
|
||||
|
||||
headers_file = doc_root + 'annotations/annotation.headers'
|
||||
|
||||
if base.startswith("temp-") and tempAnnotations[base]:
|
||||
response.headers.update(load_headers_from_file(headers_file))
|
||||
response.headers.set('Etag', hashlib.sha1(base).hexdigest())
|
||||
data = dump_json(tempAnnotations[base])
|
||||
if data != "" :
|
||||
response.content = data
|
||||
response.status = 200
|
||||
else:
|
||||
response.content = ""
|
||||
response.status = 404
|
||||
elif os.path.isfile(requested_file):
|
||||
response.headers.update(load_headers_from_file(headers_file))
|
||||
# Calculate ETag using Apache httpd's default method (more or less)
|
||||
# http://www.askapache.info//2.3/mod/core.html#fileetag
|
||||
statinfo = os.stat(requested_file)
|
||||
etag = "{0}{1}{2}".format(statinfo.st_ino, statinfo.st_mtime,
|
||||
statinfo.st_size)
|
||||
# obfuscate so we don't leak info; hexdigest for string compatibility
|
||||
response.headers.set('Etag', hashlib.sha1(etag).hexdigest())
|
||||
|
||||
with open(requested_file, 'r') as data_file:
|
||||
data = data_file.read()
|
||||
response.content = data
|
||||
response.status = 200
|
||||
else:
|
||||
response.content = 'Not Found'
|
||||
response.status = 404
|
||||
|
||||
add_cors_headers(response)
|
||||
|
||||
|
||||
@wptserve.handlers.handler
|
||||
def annotation_head(request, response):
|
||||
requested_file = doc_root + request.request_path[1:]
|
||||
base = os.path.basename(requested_file)
|
||||
|
||||
headers_file = doc_root + 'annotations/annotation.options.headers'
|
||||
|
||||
if base.startswith("temp-") and tempAnnotations[base]:
|
||||
response.status = 200
|
||||
response.headers.update(load_headers_from_file(headers_file))
|
||||
elif os.path.isfile(requested_file):
|
||||
response.status = 200
|
||||
response.headers.update(load_headers_from_file(headers_file))
|
||||
else:
|
||||
response.status = 404
|
||||
|
||||
add_cors_headers(response)
|
||||
response.content = "Annotation Options\n"
|
||||
|
||||
@wptserve.handlers.handler
|
||||
def annotation_options(request, response):
|
||||
requested_file = doc_root + request.request_path[1:]
|
||||
base = os.path.basename(requested_file)
|
||||
|
||||
headers_file = doc_root + 'annotations/annotation.options.headers'
|
||||
|
||||
if base.startswith("temp-") and tempAnnotations[base]:
|
||||
response.status = 200
|
||||
response.headers.update(load_headers_from_file(headers_file))
|
||||
elif os.path.isfile(requested_file):
|
||||
response.status = 200
|
||||
response.headers.update(load_headers_from_file(headers_file))
|
||||
else:
|
||||
response.status = 404
|
||||
|
||||
add_cors_headers(response)
|
||||
response.content = "Annotation Options\n"
|
||||
|
||||
|
||||
def create_annotation(body):
|
||||
# TODO: verify media type is JSON of some kind (at least)
|
||||
incoming = json.loads(body)
|
||||
id = "temp-"+str(uuid.uuid4())
|
||||
if 'id' in incoming:
|
||||
incoming['canonical'] = incoming['id']
|
||||
incoming['id'] = URIroot + '/annotations/' + id
|
||||
|
||||
return incoming
|
||||
|
||||
|
||||
@wptserve.handlers.handler
|
||||
def annotation_post(request, response):
|
||||
incoming = create_annotation(request.body)
|
||||
newID = incoming['id']
|
||||
key = os.path.basename(newID)
|
||||
|
||||
print "post:" + newID
|
||||
print "post:" + key
|
||||
|
||||
tempAnnotations[key] = dump_json(incoming)
|
||||
|
||||
headers_file = doc_root + 'annotations/annotation.headers'
|
||||
response.headers.update(load_headers_from_file(headers_file))
|
||||
response.headers.append('Location', newID)
|
||||
add_cors_headers(response)
|
||||
response.content = dump_json(incoming)
|
||||
response.status = 201
|
||||
|
||||
@wptserve.handlers.handler
|
||||
def annotation_put(request, response):
|
||||
incoming = create_annotation(request.body)
|
||||
|
||||
# remember it in our local cache too
|
||||
# tempAnnotations[request.request_path[1:]] = dump_jason(incoming)
|
||||
newID = incoming['id']
|
||||
key = os.path.basename(newID)
|
||||
|
||||
print "put:" + newID
|
||||
print "put:" + key
|
||||
|
||||
tempAnnotations[key] = dump_json(incoming)
|
||||
|
||||
headers_file = doc_root + 'annotations/annotation.headers'
|
||||
response.headers.update(load_headers_from_file(headers_file))
|
||||
response.headers.append('Location', incoming['id'])
|
||||
add_cors_headers(response)
|
||||
response.content = dump_json(incoming)
|
||||
response.status = 200
|
||||
|
||||
|
||||
@wptserve.handlers.handler
|
||||
def annotation_delete(request, response):
|
||||
base = os.path.basename(request.request_path[1:])
|
||||
requested_file = doc_root + request.request_path[1:]
|
||||
|
||||
add_cors_headers(response)
|
||||
|
||||
headers_file = doc_root + 'annotations/annotation.headers'
|
||||
|
||||
try:
|
||||
if base.startswith("temp-"):
|
||||
if tempAnnotations[base]:
|
||||
del tempAnnotations[base]
|
||||
else:
|
||||
os.remove(requested_file)
|
||||
response.headers.update(load_headers_from_file(headers_file))
|
||||
response.status = 204
|
||||
response.content = ''
|
||||
except OSError:
|
||||
response.status = 404
|
||||
response.content = 'Not Found'
|
||||
|
||||
if __name__ == '__main__':
|
||||
print 'http://' + myhost + ':{0}/'.format(port)
|
||||
|
||||
routes = [
|
||||
("GET", "", wptserve.handlers.file_handler),
|
||||
("GET", "index.html", wptserve.handlers.file_handler),
|
||||
|
||||
# container/collection responses
|
||||
("HEAD", "annotations/", collection_head),
|
||||
("OPTIONS", "annotations/", collection_options),
|
||||
("GET", "annotations/", collection_get),
|
||||
|
||||
# create annotations in the collection
|
||||
("POST", "annotations/", annotation_post),
|
||||
|
||||
# single annotation responses
|
||||
("HEAD", "annotations/*", annotation_head),
|
||||
("OPTIONS", "annotations/*", annotation_options),
|
||||
("GET", "annotations/*", annotation_get),
|
||||
("PUT", "annotations/*", annotation_put),
|
||||
("DELETE", "annotations/*", annotation_delete)
|
||||
]
|
||||
|
||||
httpd = wptserve.server.WebTestHttpd(host=myhost, bind_hostname=myhost, port=port, doc_root=doc_root,
|
||||
routes=routes)
|
||||
httpd.start(block=True)
|
||||
Loading…
Add table
Add a link
Reference in a new issue