structuredlog.py (1758B)
1 # produce mozlog-compatible log messages, following the spec at 2 # https://firefox-source-docs.mozilla.org/mozbase/mozlog.html 3 4 import json 5 import os 6 from time import time 7 8 9 class TestLogger: 10 def __init__(self, source, threadname="main"): 11 self.template = { 12 "source": source, 13 "thread": threadname, 14 "pid": os.getpid(), 15 } 16 directory = os.environ.get("MOZ_UPLOAD_DIR", ".") 17 self.fh = open(os.path.join(directory, threadname + "_raw.log"), "a") 18 19 def _record(self, **kwargs): 20 record = self.template.copy() 21 record.update(**kwargs) 22 if "time" not in record: 23 record["time"] = time() 24 return record 25 26 def _log_obj(self, obj): 27 print(json.dumps(obj, sort_keys=True), file=self.fh) 28 29 def _log(self, **kwargs): 30 self._log_obj(self._record(**kwargs)) 31 32 def suite_start(self): 33 self._log(action="suite_start", tests=[]) 34 35 def suite_end(self): 36 self._log(action="suite_end") 37 38 def test_start(self, testname): 39 self._log(action="test_start", test=testname) 40 41 def test_end(self, testname, status): 42 self._log(action="test_end", test=testname, status=status) 43 44 def test(self, testname, status, duration, **details): 45 record = self._record( 46 action="test_start", test=testname, **details.get("extra", {}) 47 ) 48 end_time = record["time"] 49 record["time"] -= duration 50 self._log_obj(record) 51 52 record["action"] = "test_end" 53 record["time"] = end_time 54 record["status"] = status 55 record.update(**details) 56 self._log_obj(record) 57 58 def log_info(self, message): 59 self._log(action="log", level="INFO", message=message)