summaryrefslogtreecommitdiffstats
path: root/testing/web-platform/tests/annotation-protocol/tools/protocol-server.py
diff options
context:
space:
mode:
Diffstat (limited to 'testing/web-platform/tests/annotation-protocol/tools/protocol-server.py')
-rw-r--r--testing/web-platform/tests/annotation-protocol/tools/protocol-server.py427
1 files changed, 427 insertions, 0 deletions
diff --git a/testing/web-platform/tests/annotation-protocol/tools/protocol-server.py b/testing/web-platform/tests/annotation-protocol/tools/protocol-server.py
new file mode 100644
index 0000000000..fc8fa74997
--- /dev/null
+++ b/testing/web-platform/tests/annotation-protocol/tools/protocol-server.py
@@ -0,0 +1,427 @@
+# 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)
+
+import os
+import sys
+
+here = os.path.abspath(os.path.dirname(__file__))
+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)