blob: 1c7ad234461831c6c68a7da488894df030c23075 [file] [log] [blame]
#!/usr/bin/env python3
# Copyright 2022 The Fuchsia Authors
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
"""This module is where all the record definitions and record containers live.
import collections
import copy
import io
import json
from antlion import logger
from antlion.libs import yaml_writer
from mobly.records import ExceptionRecord
from mobly.records import OUTPUT_FILE_SUMMARY
from mobly.records import TestResultEnums as MoblyTestResultEnums
from mobly.records import TestResultRecord as MoblyTestResultRecord
from mobly.records import TestResult as MoblyTestResult
from mobly.records import TestSummaryEntryType
from mobly.records import TestSummaryWriter as MoblyTestSummaryWriter
class TestSummaryWriter(MoblyTestSummaryWriter):
"""Writes test results to a summary file in real time. Inherits from Mobly's
def dump(self, content, entry_type):
"""Update Mobly's implementation of dump to work on OrderedDict.
See MoblyTestSummaryWriter.dump for documentation.
new_content = collections.OrderedDict(copy.deepcopy(content))
new_content["Type"] = entry_type.value
new_content.move_to_end("Type", last=False)
# Both user code and Mobly code can trigger this dump, hence the lock.
with self._lock:
# For Python3, setting the encoding on yaml.safe_dump does not work
# because Python3 file descriptors set an encoding by default, which
# PyYAML uses instead of the encoding on yaml.safe_dump. So, the
# encoding has to be set on the open call instead.
with, "a", encoding="utf-8") as f:
# Use safe_dump here to avoid language-specific tags in final
# output.
yaml_writer.safe_dump(new_content, f)
class TestResultEnums(MoblyTestResultEnums):
"""Enums used for TestResultRecord class. Inherits from Mobly's
Includes the tokens to mark test result with, and the string names for each
field in TestResultRecord.
class TestResultRecord(MoblyTestResultRecord):
"""A record that holds the information of a test case execution. This class
inherits from Mobly's TestResultRecord class.
test_name: A string representing the name of the test case.
begin_time: Epoch timestamp of when the test case started.
end_time: Epoch timestamp of when the test case ended.
self.uid: Unique identifier of a test case.
self.result: Test result, PASS/FAIL/SKIP.
self.extras: User defined extra information of the test result.
self.details: A string explaining the details of the test case.
def __init__(self, t_name, t_class=None):
super().__init__(t_name, t_class)
self.log_begin_time = None
self.log_end_time = None
def test_begin(self):
"""Call this when the test case it records begins execution.
Sets the begin_time of this record.
self.log_begin_time = logger.epoch_to_log_line_timestamp(self.begin_time)
def _test_end(self, result, e):
"""Class internal function to signal the end of a test case execution.
result: One of the TEST_RESULT enums in TestResultEnums.
e: A test termination signal (usually an exception object). It can
be any exception instance or of any subclass of
super()._test_end(result, e)
if self.end_time:
self.log_end_time = logger.epoch_to_log_line_timestamp(self.end_time)
def to_dict(self):
"""Gets a dictionary representing the content of this class.
A dictionary representing the content of this class.
d = collections.OrderedDict()
d[TestResultEnums.RECORD_NAME] = self.test_name
d[TestResultEnums.RECORD_CLASS] = self.test_class
d[TestResultEnums.RECORD_BEGIN_TIME] = self.begin_time
d[TestResultEnums.RECORD_END_TIME] = self.end_time
d[TestResultEnums.RECORD_LOG_BEGIN_TIME] = self.log_begin_time
d[TestResultEnums.RECORD_LOG_END_TIME] = self.log_end_time
d[TestResultEnums.RECORD_RESULT] = self.result
d[TestResultEnums.RECORD_UID] = self.uid
d[TestResultEnums.RECORD_EXTRAS] = self.extras
d[TestResultEnums.RECORD_DETAILS] = self.details
d[TestResultEnums.RECORD_EXTRA_ERRORS] = {
key: value.to_dict() for (key, value) in self.extra_errors.items()
d[TestResultEnums.RECORD_STACKTRACE] = self.stacktrace
return d
def json_str(self):
"""Converts this test record to a string in json format.
Format of the json string is:
'Test Name': <test name>,
'Begin Time': <epoch timestamp>,
'Details': <details>,
A json-format string representing the test record.
return json.dumps(self.to_dict())
class TestResult(MoblyTestResult):
"""A class that contains metrics of a test run. This class inherits from
Mobly's TestResult class.
This class is essentially a container of TestResultRecord objects.
self.requested: A list of strings, each is the name of a test requested
by user.
self.failed: A list of records for tests failed.
self.executed: A list of records for tests that were actually executed.
self.passed: A list of records for tests passed.
self.skipped: A list of records for tests skipped.
def __add__(self, r):
"""Overrides '+' operator for TestResult class.
The add operator merges two TestResult objects by concatenating all of
their lists together.
r: another instance of TestResult to be added
A TestResult instance that's the sum of two TestResult instances.
if not isinstance(r, MoblyTestResult):
raise TypeError("Operand %s of type %s is not a TestResult." % (r, type(r)))
sum_result = TestResult()
for name in sum_result.__dict__:
r_value = getattr(r, name)
l_value = getattr(self, name)
if isinstance(r_value, list):
setattr(sum_result, name, l_value + r_value)
return sum_result
def json_str(self):
"""Converts this test result to a string in json format.
Format of the json string is:
"Results": [
{<executed test record 1>},
{<executed test record 2>},
"Summary": <summary dict>
A json-format string representing the test results.
d = collections.OrderedDict()
d["ControllerInfo"] = {
record.controller_name: record.controller_info
for record in self.controller_info
d["Results"] = [record.to_dict() for record in self.executed]
d["Summary"] = self.summary_dict()
d["Error"] = self.errors_list()
json_str = json.dumps(d, indent=4)
return json_str
def summary_str(self):
"""Gets a string that summarizes the stats of this test result.
The summary provides the counts of how many test cases fall into each
category, like "Passed", "Failed" etc.
Format of the string is:
Requested <int>, Executed <int>, ...
A summary string of this test result.
l = ["%s %s" % (k, v) for k, v in self.summary_dict().items()]
msg = ", ".join(l)
return msg
def errors_list(self):
l = list()
for record in self.error:
if isinstance(record, TestResultRecord):
keys = [
elif isinstance(record, ExceptionRecord):
keys = [TestResultEnums.RECORD_DETAILS, TestResultEnums.RECORD_POSITION]
return []
l.append({k: record.to_dict()[k] for k in keys})
return l