https://github.com/web-platform-tests/wpt
Tip revision: 7cb0d28a36927eb21b7984a80c6629969634a1e2 authored by moz-wptsync-bot on 13 March 2018, 19:13:50 UTC
Performance.measure(name) should not throw if name is one of the readonly attribute of the Performance interface,
Performance.measure(name) should not throw if name is one of the readonly attribute of the Performance interface,
Tip revision: 7cb0d28
protocol-server.py
# 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
from __future__ import print_function
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)
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)
@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)
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))
print('container URI is http://' + myhost + ':{0}/'.format(port) + "/annotations/")
print('example annotation URI is http://' + myhost + ':{0}/'.format(port) + "/annotations/anno1.json")
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)