testing/web-platform/tests/network-error-logging/support/report.py
author Tsuyoshi Horo <horo@chromium.org>
Wed, 06 Mar 2019 12:33:36 +0000
changeset 464642 5f4503098f271c69d34c5a028f5a5eb909859e79
parent 464361 66b1aba3fb4372ec47986e728b1da9b32dd3d368
permissions -rw-r--r--
Bug 1527961 [wpt PR 15372] - Implement Signed Exchange reporting for distributors, a=testonly Automatic update from web-platform-tests Implement Signed Exchange reporting for distributors This CL implements Signed Exchange reporting for distributors. Spec: https://github.com/WICG/webpackage/pull/374 This CL has two success test cases only. I will add more failure test cases in another CL. Bug: 910516 Change-Id: I31a6e629073970e03a93d2379be1ae63ef6adf6b Reviewed-on: https://chromium-review.googlesource.com/c/1468855 Commit-Queue: Tsuyoshi Horo <horo@chromium.org> Reviewed-by: Christian Dullweber <dullweber@chromium.org> Reviewed-by: Kinuko Yasuda <kinuko@chromium.org> Reviewed-by: Kouhei Ueno <kouhei@chromium.org> Reviewed-by: Kunihiko Sakamoto <ksakamoto@chromium.org> Reviewed-by: Eric Orth <ericorth@chromium.org> Cr-Commit-Position: refs/heads/master@{#634161} -- wpt-commits: 66dbfd0ae579135db73772c534879c51f6dbc674 wpt-pr: 15372

import time
import json
import re

def retrieve_from_stash(request, key, timeout, min_count, default_value):
  t0 = time.time()
  while time.time() - t0 < timeout:
    time.sleep(0.5)
    value = request.server.stash.take(key=key)
    if value is not None and len(value) >= min_count:
      request.server.stash.put(key=key, value=value)
      return json.dumps(value)

  return default_value

def main(request, response):
  # Handle CORS preflight requests
  if request.method == 'OPTIONS':
    # Always reject preflights for one subdomain
    if "www2" in request.headers["Origin"]:
      return (400, [], "CORS preflight rejected for www2")
    return [
      ("Content-Type", "text/plain"),
      ("Access-Control-Allow-Origin", "*"),
      ("Access-Control-Allow-Methods", "post"),
      ("Access-Control-Allow-Headers", "Content-Type"),
    ], "CORS allowed"

  op = request.GET.first("op");
  key = request.GET.first("reportID")

  if op == "retrieve_report":
    try:
      timeout = float(request.GET.first("timeout"))
    except:
      timeout = 0.5
    try:
      min_count = int(request.GET.first("min_count"))
    except:
      min_count = 1
    return [("Content-Type", "application/json")], retrieve_from_stash(request, key, timeout, min_count, '[]')

  # append new reports
  new_reports = json.loads(request.body)
  for report in new_reports:
    report["metadata"] = {
      "content_type": request.headers["Content-Type"],
    }
  with request.server.stash.lock:
    reports = request.server.stash.take(key=key)
    if reports is None:
      reports = []
    reports.extend(new_reports)
    request.server.stash.put(key=key, value=reports)

  # return acknowledgement report
  return [("Content-Type", "text/plain")], "Recorded report"