forked from mozilla/wpt-api
-
Notifications
You must be signed in to change notification settings - Fork 0
/
send_to_telemetry.py
executable file
·99 lines (84 loc) · 3.13 KB
/
send_to_telemetry.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
from dataclasses import asdict, dataclass
import json
import os
import sys
import uuid
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(path) as f:
data = json.load(f)
with open("metrics.json") as f:
metrics = json.load(f)
for test in data:
# create an empty dictionary
values = {}
# grab each metric from metrics.json
for metric in metrics:
name = metric["name"]
# now, grab all metrics' values
for measure in ["median", "standardDeviation"]:
sample = None
try:
sample = test["data"][measure]["firstView"].get(name)
except AttributeError:
pass
# sefdefault()s here will return each metric name & value,
# or return and set an empty dict
if sample is not None:
m = values.setdefault(name, {})
# write each metric having a "firstView" entry + value,
# or return and set an empty dict
first_view = m.setdefault("firstView", {})
first_view[measure] = sample
sample = test["data"]["median"]["firstView"]
# get browser name and channel, then lower()-case them
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 the generated JSON output
with open(f"wpt-telemetry-{test['data']['id']}.json", "w") as f:
json.dump(asdict(result), f)
# validate our generated JSON output against Telemetry's Pipeline schema
# https://github.com/mozilla-services/mozilla-pipeline-schemas/blob/dev/schemas/webpagetest/webpagetest-run/webpagetest-run.1.schema.json
with open("wpt-schema.json") as f:
schema = json.load(f)
validate(asdict(result), schema)
# send to telemetry
wpt_run_uuid = str(uuid.uuid4())
telemetry_url = f"https://incoming.telemetry.mozilla.org/submit/webpagetest/webpagetest-run/1/{wpt_run_uuid}"
results_json = json.dumps(asdict(result))
r = requests.post(
url=telemetry_url,
data=results_json,
headers={"Content-Type": "application/json"},
)
r.raise_for_status()
if __name__ == "__main__":
if not len(sys.argv) == 2:
print("Usage: python send_to_telemetry.py path_to_wpt.json")
exit()
main(*sys.argv[1:])