testing/web-platform/tests/attribution-reporting/resources/reports.py

   1 """Methods for the report-event-attribution and report-aggregate-attribution endpoints"""
   2 import json
   3 from typing import List, Optional, Tuple, TypedDict
   4
   5 from wptserve.request import Request
   6 from wptserve.stash import Stash
   7 from wptserve.utils import isomorphic_decode, isomorphic_encode
   8
   9 # Key used to access the reports in the stash.
  10 REPORTS = "4691a2d7fca5430fb0f33b1bd8a9d388"
  11
  12 Header = Tuple[str, str]
  13 Status = Tuple[int, str]
  14 Response = Tuple[Status, List[Header], str]
  15
  16
  17 def decode_headers(headers: dict) -> dict:
  18   """Decodes the headers from wptserve.
  19
  20   wptserve headers are encoded like
  21   {
  22     encoded(key): [encoded(value1), encoded(value2),...]
  23   }
  24   This method decodes the above using the wptserve.utils.isomorphic_decode
  25   method
  26   """
  27   return {
  28       isomorphic_decode(key): [isomorphic_decode(el) for el in value
  29                               ] for key, value in headers.items()
  30   }
  31
  32 def get_request_origin(request: Request) -> str:
  33   return "%s://%s" % (request.url_parts.scheme,
  34                       request.url_parts.netloc)
  35
  36
  37 def handle_post_report(request: Request, headers: List[Header]) -> Response:
  38   """Handles POST request for reports.
  39
  40   Retrieves the report from the request body and stores the report in the
  41   stash.
  42   """
  43   store_report(
  44       request.server.stash, get_request_origin(request), {
  45           "body": request.body.decode("utf-8"),
  46           "headers": decode_headers(request.headers)
  47       })
  48   return (201, "OK"), headers, json.dumps({
  49       "code": 201,
  50       "message": "Report successfully stored."
  51   })
  52
  53
  54 def handle_get_reports(request: Request, headers: List[Header]) -> Response:
  55   """Handles GET request for reports.
  56
  57   Retrieves and returns all reports from the stash.
  58   """
  59   reports = take_reports(request.server.stash, get_request_origin(request))
  60   headers.append(("Access-Control-Allow-Origin", "*"))
  61   return (200, "OK"), headers, json.dumps({
  62       "code": 200,
  63       "reports": reports,
  64   })
  65
  66
  67 def store_report(stash: Stash, origin: str, report: str) -> None:
  68   """Stores the report in the stash. Report here is a JSON."""
  69   with stash.lock:
  70     reports_dict = stash.take(REPORTS)
  71     if not reports_dict:
  72       reports_dict = {}
  73     reports = reports_dict.get(origin, [])
  74     reports.append(report)
  75     reports_dict[origin] = reports
  76     stash.put(REPORTS, reports_dict)
  77   return None
  78
  79
  80 def take_reports(stash: Stash, origin: str) -> List[str]:
  81   """Takes all the reports from the stash and returns them."""
  82   with stash.lock:
  83     reports_dict = stash.take(REPORTS)
  84     if not reports_dict:
  85       reports_dict = {}
  86
  87     reports = reports_dict.pop(origin, [])
  88     stash.put(REPORTS, reports_dict)
  89   return reports
  90
  91
  92 def handle_reports(request: Request) -> Response:
  93   """Handles request to get or store reports."""
  94   headers = [("Content-Type", "application/json")]
  95   if request.method == "POST":
  96     return handle_post_report(request, headers)
  97   if request.method == "GET":
  98     return handle_get_reports(request, headers)
  99   return (405, "Method Not Allowed"), headers, json.dumps({
 100       "code": 405,
 101       "message": "Only GET or POST methods are supported."
 102   })