send_to_telemetry.py

from dataclasses import asdict, dataclass
import json
import os
import sys

from jsonschema import validate
# import requests


@dataclass
class TestResult:
    appName: str
    channel: str
    version: str
    connection: str
    url: str
    platform: str
    runner: str
    runId: str
    sessionState: str
    metrics: dict


def main(path):
    with open("wpt.json") as f:
        data = json.load(f)

    with open("metrics.json") as f:
        metrics = json.load(f)

    for test in data:
        values = {}
        for metric in metrics:
            name = metric["name"]
            for measure in ["median", "standardDeviation"]:
                sample = test["data"][measure]["firstView"].get(name)
                if sample is not None:
                    m = values.setdefault(name, {})
                    first_view = m.setdefault("firstView", {})
                    first_view[measure] = sample

        sample = test["data"]["median"]["firstView"]
        browser, _, channel = sample["browser_name"].lower().partition(" ")

        result = TestResult(
            appName=browser,
            channel=channel or "release",
            version=sample["browser_version"],
            connection=test["data"]["connectivity"].lower(),
            url=test["data"]["testUrl"],
            platform="desktop",
            runner=os.getenv("JENKINS_URL", "unknown"),
            runId=os.getenv("BUILD_TAG", "unknown"),
            sessionState="noAuth",
            metrics=values)
        print(asdict(result))

        # save
        with open(f"wpt-telemetry-{test['data']['id']}.json", "w") as f:
            json.dump(asdict(result), f)

        # validate
        with open("wpt-schema.json") as f:
            schema = json.load(f)
            validate(asdict(result), schema)

        # send to telemetry
        # r = requests.post(url="", data=asdict(result), type="json")
        # r.raise_on_error()


if __name__ == "__main__":
    if not len(sys.argv) == 2:
        print("Usage: python send_to_telemetry.py path")
        exit()
    main(*sys.argv[1:])