path: root/testing/web-platform/tests/annotation-protocol/tools/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

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)