236 lines
9.5 KiB
Python
236 lines
9.5 KiB
Python
import re
|
||
from datetime import datetime
|
||
from typing import Dict, List
|
||
|
||
"""
|
||
cmocka.json
|
||
"""
|
||
|
||
Passed = "Passed"
|
||
Failed = "Failed"
|
||
Unexecuted = "Unexecuted"
|
||
|
||
|
||
class CaseInfo:
|
||
def __init__(self, test_suite_name, test_case_name, status, log=None):
|
||
self.test_suite_name = test_suite_name
|
||
# case name，e.g. "TestNuttxMm01"
|
||
self.test_case_name = test_case_name
|
||
# result: Passed or Failed
|
||
self.status = status
|
||
# log
|
||
self.log: List = [""] if log is None else log
|
||
|
||
|
||
class SuiteInfo:
|
||
def __init__(self, test_suite_name):
|
||
# suite name, e.g. "NuttxMmTestSuites"
|
||
self.test_suite_name = test_suite_name
|
||
# all test cases in the current test suite
|
||
self.test_cases: Dict[str, CaseInfo] = dict()
|
||
# number of cases passed in the current test suites
|
||
self.passed_count = 0
|
||
# number of cases failed in the current test suites
|
||
self.failed_count = 0
|
||
# case run count
|
||
self.run_count = 0
|
||
# unexecuted count
|
||
self.unexecuted_count = 0
|
||
# number of cases in the current test suites
|
||
self.cases_count = 0
|
||
# suite run flag
|
||
self.is_suite_run = False
|
||
|
||
|
||
class CmockaSummary:
|
||
def __init__(self, duration=0):
|
||
# number of all test suites
|
||
self.total_suites_count = 0
|
||
# all test cases number
|
||
self.total_cases_count = 0
|
||
# number of all passed cases
|
||
self.total_passed_count = 0
|
||
# number of all failed cases
|
||
self.total_failed_count = 0
|
||
# number of all unknown cases
|
||
self.total_unexecuted_count = 0
|
||
# duration
|
||
self.duration = duration
|
||
|
||
|
||
class CmockaSingleCoreRecord:
|
||
def __init__(self, lines, core="", board="", log="", duration=0):
|
||
# create time
|
||
self.create_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
# core
|
||
self.core = "" if core is None else core
|
||
# board
|
||
self.board = "" if board is None else board
|
||
# cmocka info
|
||
self.test_suites: Dict[str, SuiteInfo] = dict()
|
||
# summary
|
||
self.summary = CmockaSummary(duration)
|
||
# log path
|
||
self.log = "" if log is None else log
|
||
# bad_case
|
||
self.bad_case_tip = ""
|
||
|
||
suite_pattern = r"\] (?P<test_suite_name>[a-zA-Z]*TestSuites)"
|
||
case_pattern = r"\]\s+(?P<test_case_name>TestNuttx\w+)"
|
||
lines_iter = iter(lines)
|
||
|
||
current_suite = None
|
||
while True:
|
||
try:
|
||
line = next(lines_iter)
|
||
if (suite_match := re.search(suite_pattern, line)) is not None:
|
||
current_suite = suite_match.group("test_suite_name")
|
||
|
||
elif (
|
||
current_suite is not None
|
||
and (case_match := re.search(case_pattern, line)) is not None
|
||
):
|
||
current_case = case_match.group("test_case_name")
|
||
self.append(CaseInfo(current_suite, current_case, Unexecuted))
|
||
|
||
except StopIteration:
|
||
break
|
||
|
||
def append(self, object: CaseInfo):
|
||
suite: SuiteInfo = self.test_suites.get(object.test_suite_name)
|
||
if suite is None:
|
||
suite = SuiteInfo(object.test_suite_name)
|
||
self.test_suites.update({object.test_suite_name: suite})
|
||
suite.test_cases.update({object.test_case_name: object})
|
||
|
||
passed_count = 0
|
||
failed_count = 0
|
||
unexecuted_count = 0
|
||
|
||
test_case: CaseInfo
|
||
for test_case in suite.test_cases.values():
|
||
if test_case.status == Passed:
|
||
passed_count += 1
|
||
elif test_case.status == Failed:
|
||
failed_count += 1
|
||
else:
|
||
unexecuted_count += 1
|
||
suite.passed_count = passed_count
|
||
suite.failed_count = failed_count
|
||
suite.unexecuted_count = unexecuted_count
|
||
suite.run_count = passed_count + failed_count
|
||
suite.cases_count = passed_count + failed_count + unexecuted_count
|
||
if passed_count + failed_count != 0:
|
||
suite.is_suite_run = True
|
||
|
||
total_passed_count = 0
|
||
total_failed_count = 0
|
||
total_unexecuted_count = 0
|
||
total_cases_count = 0
|
||
|
||
suite: SuiteInfo
|
||
for suite in self.test_suites.values():
|
||
total_passed_count += suite.passed_count
|
||
total_failed_count += suite.failed_count
|
||
total_unexecuted_count += suite.unexecuted_count
|
||
total_cases_count += suite.cases_count
|
||
|
||
self.summary.total_passed_count = total_passed_count
|
||
self.summary.total_failed_count = total_failed_count
|
||
self.summary.total_unexecuted_count = total_unexecuted_count
|
||
self.summary.total_cases_count = total_cases_count
|
||
self.summary.total_suites_count = len(self.test_suites)
|
||
|
||
def process(self, lines, err_code):
|
||
# regular expression
|
||
suite_start_pattern = r"\] (?P<test_suite_name>[a-zA-Z]*TestSuites): Running (?P<cases_count>\d+) test\(s\)"
|
||
case_run_pattern = r"\[\s+RUN\s+\] (?P<test_case_name>TestNuttx\w+)"
|
||
case_pass_pattern = r"\[\s+OK\s+\] (?P<test_case_name>TestNuttx\w+)"
|
||
case_fail_pattern = r"\[\s+FAILED\s+\] (?P<test_case_name>TestNuttx\w+)"
|
||
|
||
lines_iter = iter(lines)
|
||
line = next(lines_iter)
|
||
while True:
|
||
try:
|
||
interrupt_flag = False
|
||
# matching new test suites
|
||
if (
|
||
suite_start_match := re.search(suite_start_pattern, line)
|
||
) is not None:
|
||
test_suite_name = suite_start_match.group("test_suite_name")
|
||
cases_count = int(suite_start_match.group("cases_count"))
|
||
suite_end_pattern = r"{}: {} test(s) run.".format(
|
||
test_suite_name, cases_count
|
||
)
|
||
|
||
line = next(lines_iter)
|
||
while True:
|
||
if (
|
||
case_run_match := re.search(case_run_pattern, line)
|
||
) is not None:
|
||
test_case_name = case_run_match.group("test_case_name")
|
||
log = [line]
|
||
|
||
while True:
|
||
try:
|
||
line = next(lines_iter)
|
||
log.append(line)
|
||
except StopIteration:
|
||
self.append(
|
||
CaseInfo(
|
||
test_suite_name, test_case_name, Failed, log
|
||
)
|
||
)
|
||
if err_code == -3:
|
||
self.bad_case_tip = f"This case was not executed, \
|
||
because crash occurred after running '{test_case_name}'."
|
||
elif err_code == -4:
|
||
self.bad_case_tip = f"This case was not executed, \
|
||
because no response for a long time after running '{test_case_name}'."
|
||
elif err_code == -1:
|
||
self.bad_case_tip = f"This case was not executed, \
|
||
because the maximum waiting time has been exceeded \
|
||
while running '{test_case_name}'."
|
||
else:
|
||
self.bad_case_tip = "This case was not executed due to unknown reasons."
|
||
raise StopIteration
|
||
|
||
if re.search(case_pass_pattern, line) is not None:
|
||
self.append(
|
||
CaseInfo(
|
||
test_suite_name, test_case_name, Passed, log
|
||
)
|
||
)
|
||
break
|
||
|
||
elif re.search(case_fail_pattern, line) is not None:
|
||
self.append(
|
||
CaseInfo(
|
||
test_suite_name, test_case_name, Failed, log
|
||
)
|
||
)
|
||
break
|
||
|
||
elif re.search(suite_start_pattern, line) is not None:
|
||
self.append(
|
||
CaseInfo(
|
||
test_suite_name, test_case_name, Failed, log
|
||
)
|
||
)
|
||
interrupt_flag = True
|
||
break
|
||
|
||
elif suite_end_pattern in line:
|
||
break
|
||
|
||
if interrupt_flag:
|
||
break
|
||
|
||
line = next(lines_iter)
|
||
|
||
if interrupt_flag:
|
||
continue
|
||
line = next(lines_iter)
|
||
except StopIteration:
|
||
break
|