测试报告管理模块
T8840 2023/1/15
# 测试报告原理
# 方式一：利用树（数据结构）来设计树形测试结果输出
步骤：
- 首先使用ResultReporter类进行实例化
- 添加一个测试列表节点，调用add_list方法
- 添加一个测试用例节点，调用add_test方法
- 添加一个SETUP步骤节点，调用add_step_group方法
- 添加步骤，调用add方法
- 结束第一个测试用例，调用end_test方法
- 循环上面方法
import os
from enum import IntEnum
from threading import Event, Lock
from functools import wraps
import time
def get_local_time():
return time.strftime("%B %d, %y %H:%M:%S", time.localtime())
class StepResult(IntEnum):
"""
表示节点的状态
"""
INFO = 1
PASS = 2
FAIL = 4
EXCEPTION = 8
WARNING = 16
ERROR = 32
class NodeType(IntEnum):
"""
表示节点的类型
"""
Step = 1
Case = 2
TestList = 4
Other = 256
class ResultNode:
"""
测试结果节点
"""
def __init__(self, header, status=None, message="", parent=None,
node_type=NodeType.Other):
self.status = StepResult.INFO if status is None else status
self.header = header
self.message = message
self.children = list()
self.parent = parent
self.type = node_type
self.timestamp = get_local_time()
self.log = None
def add_child(self, header, status=StepResult.INFO, message="",
node_type=NodeType.Other):
"""
添加新的子节点并返回该节点
"""
new_node = ResultNode(header, message=message, parent=self,
node_type=node_type)
# 在case或者step类型的节点中，只允许类型是Step
if self.type in [NodeType.Step, NodeType.Case]:
new_node.type = NodeType.Step
self.children.append(new_node)
new_node.set_status(status)
return new_node
def set_status(self, status):
"""
设置当前节点的状态，并且同时更新父节点的状态
"""
if self.type == NodeType.Other:
# 对于类型是非case或者是step的节点，不作状态设置
return
if status == StepResult.INFO:
return
if self.status in [StepResult.INFO, StepResult.PASS]:
self.status = status
self.parent.set_status(status)
def add(self, status, header, message=""):
"""
简化的add方法，提供给事件驱动
"""
self.add_child(header, status, message, NodeType.Step)
if self.log:
self.log.info(header)
def get_test_point_stats(self):
stats_pass = 0
stats_fail = 0
stats_error = 0
stats_warning = 0
stats_exception = 0
for child in self.children:
child_stats = child.get_test_point_stats()
stats_pass += child_stats[0]
stats_fail += child_stats[1]
stats_error += child_stats[2]
stats_warning += child_stats[3]
stats_exception += child_stats[4]
if not any(self.children):
if self.status == StepResult.PASS:
stats_pass = 1
elif self.status == StepResult.FAIL:
stats_fail = 1
elif self.status == StepResult.ERROR:
stats_error = 1
elif self.status == StepResult.WARNING:
stats_warning = 1
elif self.status == StepResult.EXCEPTION:
stats_exception = 1
return stats_pass, stats_fail, stats_error, stats_warning, stats_exception
def get_test_case_stats(self):
stats_pass = 0
stats_fail = 0
stats_error = 0
stats_warning = 0
stats_exception = 0
if self.type == NodeType.Case:
if self.status == StepResult.PASS:
stats_pass = 1
elif self.status == StepResult.FAIL:
stats_fail = 1
elif self.status == StepResult.ERROR:
stats_error = 1
elif self.status == StepResult.WARNING:
stats_warning = 1
elif self.status == StepResult.EXCEPTION:
stats_exception = 1
else:
for child in self.children:
child_stats = child.get_test_case_stats()
stats_pass += child_stats[0]
stats_fail += child_stats[1]
stats_error += child_stats[2]
stats_warning += child_stats[3]
stats_exception += child_stats[4]
return stats_pass, stats_fail, stats_error, stats_warning, stats_exception
@property
def is_leaf(self):
return any(self.children)
def to_dict(self):
"""
将结果节点输出成字典结构
"""
rv = dict()
rv["header"] = self.header
rv["status"] = self.status.value
rv["message"] = self.message
rv["type"] = self.type.value
rv["children"] = list()
rv["timestamp"] = self.timestamp
for child in self.children:
rv["children"].append(child.to_dict())
return rv
def to_text(self, indent=0):
"""
将结果生成文本类型的结构
"""
rv = f"{self._get_intent(indent)}[{self.timestamp}]"
if self.type == NodeType.Case:
rv += "[TestCase] "
if self.type == NodeType.TestList:
rv += "[Test List] "
rv += self.header
if self.type in [NodeType.Case, NodeType.Step]:
rv += self._get_dot_line(rv, 80)
rv += self.status.name
rv += os.linesep
if self.message:
rv += self._get_intent(indent)
rv += f"Description:{self.message}{os.linesep}"
for child in self.children:
rv += child.to_text(indent+1)
return rv
def _get_intent(self, indent):
return "+" * (indent * 2)
def _get_dot_line(self, line, line_max):
if len(line) >= line_max:
return line
else:
return "-" * (line_max - len(line))
def locker(lock):
def outer(func):
@wraps(func)
def inner(*args, **kwargs):
try:
lock.acquire()
return func(*args, **kwargs)
finally:
lock.release()
return inner
return outer
class ResultReporter:
my_lock = Lock()
def __init__(self, logger):
self.halt_on_failure = False
self.halt_on_exception = False
self.halt_event = Event()
self.root = ResultNode("Root")
self.recent_case = None
self.recent_node = self.root
self.recent_list = None
self.logger = logger
self.case_logger = None
def search_result(self, case_name):
"""
搜索给定的测试用例名称的测试结果
"""
self._search_result(self.root, case_name)
def _search_result(self, node, case_name):
if node.type == NodeType.Step:
return None
for child in node.children:
if child.header == case_name:
return child.status
else:
rv = self._search_result(child, case_name)
if rv:
return rv
else:
return None
@locker(my_lock)
def add_node(self, header, message="", status=StepResult.INFO, node_type=NodeType.Other):
self.recent_node = self.recent_node.add_child(header=header,
status=status,
message=message,
node_type=node_type)
self._log_info(f"Result Node {header}, Message: {message}")
return self.recent_node
@locker(my_lock)
def pop(self):
if self.recent_node.parent:
self.recent_node = self.recent_node.parent
@locker(my_lock)
def add_test(self, case_name):
self.recent_node = self.recent_node.add_child(header=case_name,
node_type=NodeType.Case)
self.recent_case = self.recent_node
self._log_info(f"[Test Case] {case_name}")
@locker(my_lock)
def end_test(self):
if self.recent_case is None:
return
self.recent_node = self.recent_case.parent
self.recent_case = None
@locker(my_lock)
def add_list(self, list_name):
self.recent_node = self.recent_node.add_child(header=list_name,
node_type=NodeType.TestList)
self.recent_list = self.recent_node
self._log_info(f"[Test list] {list_name}")
@locker(my_lock)
def end_list(self):
if self.recent_list is None:
return
self.recent_node = self.recent_list.parent
self.recent_list = None
@locker(my_lock)
def add_step_group(self, group_name):
self.recent_node = self.recent_node.add_child(header=group_name,
node_type=NodeType.Step)
self._log_info(f"[Test Step Group] {group_name}")
@locker(my_lock)
def add_event_group(self, group_name):
rv = self.recent_node.add_child(header=group_name,
node_type=NodeType.Step)
rv.log = self.case_logger if self.case_logger is not None else self.logger
self._log_info(f"[Event] {group_name}")
return rv
@locker(my_lock)
def end_step_group(self):
if self.recent_node.parent:
self.recent_node = self.recent_node.parent
def add_precheck_result(self, result, headline):
pass
def is_high_priority_passed(self, priority):
pass
@locker(my_lock)
def add(self, status: StepResult, headline, message=""):
self.recent_node.add_child(header=headline,
message=message,
node_type=NodeType.Step,
status=status)
self._log_info("Step: " + headline)
self._log_info("Message" + message)
self.halt_event.clear()
if status == StepResult.FAIL and self.halt_on_failure:
self.halt_event.wait()
elif status == StepResult.EXCEPTION and self.halt_on_exception:
self.halt_event.wait()
def _log_info(self, message):
if self.case_logger:
self.case_logger.info(message)
else:
self.logger.info(message)
if __name__ == "__main__":
import logging
rr = ResultReporter(logging)
# 添加一个测试列表节点
rr.add_list("Test List 1")
# 添加一个测试用例节点
rr.add_test("Test Case 1")
# 添加一个SETUP步骤节点
rr.add_step_group("SETUP")
# 添加一些步骤
rr.add(StepResult.PASS, "Do Something Setup", "I'm doing something")
rr.end_step_group()
# 添加一个测试步骤节点
rr.add_step_group("TEST")
rr.add_step_group("Login to website")
rr.add(StepResult.PASS, "Input Username", "Username is admin")
rr.add(StepResult.PASS, "Input Password", "Password is admin")
rr.add(StepResult.FAIL, "Login", "Login is failed")
rr.end_step_group()
# 这里我们少了一个end_group, 但是end_test会把我们带回正确的位置。
rr.end_test()
# 第二个测试用例
rr.add_test("Test Case 2")
rr.add_step_group("SETUP")
rr.add(StepResult.PASS, "Do Something Setup", "I'm doing something")
rr.end_step_group()
rr.add_step_group("TEST")
rr.add_step_group("Login to website")
rr.add(StepResult.PASS, "Input Username", "Username is admin")
rr.add(StepResult.PASS, "Input Password", "Password is admin")
rr.end_step_group()
rr.end_test()
# 第三个测试用例
rr.add_list("Sub Test List")
rr.add_test("Test Case 3")
rr.add_step_group("SETUP")
rr.add(StepResult.PASS, "Do Something Setup", "I'm doing something")
rr.end_step_group()
rr.add_step_group("TEST")
rr.add_step_group("Login to website")
rr.add(StepResult.PASS, "Input Username", "Username is admin")
rr.add(StepResult.PASS, "Input Password", "Password is admin")
rr.end_step_group()
rr.end_test()
rr.end_list()
# 第四个测试用例
rr.add_test("Test Case 4")
rr.add_step_group("SETUP")
rr.add(StepResult.PASS, "Do Something Setup", "I'm doing something")
rr.end_step_group()
rr.add_step_group("TEST")
rr.add_step_group("Login to website")
rr.add(StepResult.PASS, "Input Username", "Username is admin")
rr.add(StepResult.PASS, "Input Password", "Password is admin")
rr.end_step_group()
rr.end_test()
rr.end_list()
print(rr.root.to_text())
tp_stats = rr.root.get_test_point_stats()
print(f"PASS: {tp_stats[0]}, FAIL: {tp_stats[1]}")
print(f"ERROR: {tp_stats[2]}, WARNING: {tp_stats[3]}, EXCEPTION: {tp_stats[4]}")
tc_stats = rr.root.get_test_case_stats()
print(f"PASS: {tc_stats[0]}, FAIL: {tc_stats[1]}")
print(f"ERROR: {tc_stats[2]}, WARNING: {tc_stats[3]}, EXCEPTION: {tc_stats[4]}")
# import json
# print(json.dumps(rr.root.to_dict(), indent=4))
"""
# 输出
[February 05, 23 09:00:24]Root
++[February 05, 23 09:00:24][Test List] Test List 1
++++[February 05, 23 09:00:24][TestCase] Test Case 1----------------------------FAIL
++++++[February 05, 23 09:00:24]SETUP-------------------------------------------PASS
++++++++[February 05, 23 09:00:24]Do Something Setup----------------------------PASS
++++++++Description:I'm doing something
++++++[February 05, 23 09:00:24]TEST--------------------------------------------FAIL
++++++++[February 05, 23 09:00:24]Login to website------------------------------FAIL
++++++++++[February 05, 23 09:00:24]Input Username------------------------------PASS
++++++++++Description:Username is admin
++++++++++[February 05, 23 09:00:24]Input Password------------------------------PASS
++++++++++Description:Password is admin
++++++++++[February 05, 23 09:00:24]Login---------------------------------------FAIL
++++++++++Description:Login is failed
++++[February 05, 23 09:00:24][TestCase] Test Case 2----------------------------PASS
++++++[February 05, 23 09:00:24]SETUP-------------------------------------------PASS
++++++++[February 05, 23 09:00:24]Do Something Setup----------------------------PASS
++++++++Description:I'm doing something
++++++[February 05, 23 09:00:24]TEST--------------------------------------------PASS
++++++++[February 05, 23 09:00:24]Login to website------------------------------PASS
++++++++++[February 05, 23 09:00:24]Input Username------------------------------PASS
++++++++++Description:Username is admin
++++++++++[February 05, 23 09:00:24]Input Password------------------------------PASS
++++++++++Description:Password is admin
++++[February 05, 23 09:00:24][Test List] Sub Test List
++++++[February 05, 23 09:00:24][TestCase] Test Case 3--------------------------PASS
++++++++[February 05, 23 09:00:24]SETUP-----------------------------------------PASS
++++++++++[February 05, 23 09:00:24]Do Something Setup--------------------------PASS
++++++++++Description:I'm doing something
++++++++[February 05, 23 09:00:24]TEST------------------------------------------PASS
++++++++++[February 05, 23 09:00:24]Login to website----------------------------PASS
++++++++++++[February 05, 23 09:00:24]Input Username----------------------------PASS
++++++++++++Description:Username is admin
++++++++++++[February 05, 23 09:00:24]Input Password----------------------------PASS
++++++++++++Description:Password is admin
++++[February 05, 23 09:00:24][TestCase] Test Case 4----------------------------PASS
++++++[February 05, 23 09:00:24]SETUP-------------------------------------------PASS
++++++++[February 05, 23 09:00:24]Do Something Setup----------------------------PASS
++++++++Description:I'm doing something
++++++[February 05, 23 09:00:24]TEST--------------------------------------------PASS
++++++++[February 05, 23 09:00:24]Login to website------------------------------PASS
++++++++++[February 05, 23 09:00:24]Input Username------------------------------PASS
++++++++++Description:Username is admin
++++++++++[February 05, 23 09:00:24]Input Password------------------------------PASS
++++++++++Description:Password is admin
PASS: 12, FAIL: 1
ERROR: 0, WARNING: 0, EXCEPTION: 0
PASS: 3, FAIL: 1
ERROR: 0, WARNING: 0, EXCEPTION: 0
"""
# 方式二：使用python with上下文进行设计
import json
import time
import os
import logging
from enum import Enum
TIME_FORMAT = "%Y-%m-%d %H:%M:%S"
class ResultType(Enum):
PASS = "Passed"
FAIL = "Failed"
ERROR = "Errored"
BLOCK = "Blocked"
SKIP = "Skipped"
INFO = "Information"
class StepEnd(Exception):
def __init__(self, result):
self.result = result
class NodeEntry:
"""
代表一般节点，比如测试列表
"""
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_tb:
self.message += str(exc_tb)
def __init__(self, headline, parent=None, message="", update_action=None):
self.headline = headline
self.message = message
self.parent = parent
self.children = list()
self.timestamp = time.localtime()
self.update_action = update_action
def start_node(self, headline, message=""):
ret = NodeEntry(headline, parent=self, message=message, update_action=self.update_action)
self.children.append(ret)
if self.update_action is not None:
self.update_action()
return ret
def start_case(self, headline):
ret = CaseEntry(headline, parent=self)
ret.update_action = self.update_action
self.children.append(ret)
if self.update_action is not None:
self.update_action()
return ret
def get_json(self):
json_obj = dict()
json_obj["headline"] = self.headline
json_obj["message"] = self.message
json_obj['timestamp'] = time.strftime(TIME_FORMAT, self.timestamp)
json_obj["children"] = list()
for child in self.children:
json_obj["children"].append(child.get_json())
return json_obj
def get_friend_print(self, indent=0):
ret = "" * indent
ret += self.__str__()
for child in self.children:
ret += child.get_friend_print(indent+4)
return ret
def __str__(self):
return self.headline + "[" + time.strftime(TIME_FORMAT, self.timestamp) + "]" + os.linesep
class CaseEntry(NodeEntry):
"""
代表测试用例的节点
"""
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
self.update_result()
if self.result is None:
self.result = ResultType.PASS
if self.update_action:
self.update_action()
return
if exc_type is StepEnd:
self.result = exc_val.result
if self.result != ResultType.PASS:
self.message += exc_tb
else:
self.result = ResultType.ERROR
self.message = str(exc_tb)
if self.update_action:
self.update_action()
return True
def __str__(self):
return self.headline + "[" + time.strftime(TIME_FORMAT, self.timestamp) + "]"
def _get_result_headline(self, width, indent, headline_max=59):
self.update_result()
head_str = self.__str__()
if len(head_str) > headline_max:
head_str = head_str[0: headline_max] + "..."
space_count = width - indent - len(head_str)
ret = (" " * indent) + head_str + ("-" * space_count) + self.result.value.upper()
return ret
def start(self, headline, message, prefix=None):
entry = CaseStepEntry(headline=headline, parent=self, message=message)
entry.update_action = self.update_action
if prefix is not None:
entry.step_prefix = prefix
self.children.append(entry)
if self.update_action is not None:
self.update_action()
return entry
def passed(self, message):
self.message += message + os.linesep
if self.update_action is not None:
self.update_action()
raise StepEnd(ResultType.PASS)
def failed(self, message):
self.message += message + os.linesep
if self.update_action is not None:
self.update_action()
raise StepEnd(ResultType.FAIL)
def blocked(self, message):
self.message += message + os.linesep
if self.update_action is not None:
self.update_action()
raise StepEnd(ResultType.BLOCK)
def skipped(self, message):
self.message += message + os.linesep
if self.update_action is not None:
self.update_action()
raise StepEnd(ResultType.SKIP)
def errored(self, message):
self.message += message + os.linesep
if self.update_action is not None:
self.update_action()
raise StepEnd(ResultType.ERROR)
def info(self, message):
if self.update_action is not None:
self.update_action()
self.message += "INFO: " + message + os.linesep
def __init__(self, headline, parent=None, message=""):
super().__init__(headline, parent, message)
self.result = None
def get_json(self):
self.update_result()
json_obj = super().get_json()
json_obj['result'] = self.result.value
return json_obj
def get_friend_print(self, indent=0):
ret = self._get_result_headline(100, indent, ) + os.linesep
for child in self.children:
ret += child.get_friend_print(indent + 4)
return ret
def update_result(self):
if not any(self.children):
return
has_error = False
has_failed = False
has_block = False
has_pass = False
for child in self.children:
child.update_result()
if child.result == ResultType.ERROR:
has_error = True
elif child.result == ResultType.FAIL:
has_failed = True
elif child.result == ResultType.BLOCK:
has_block = True
elif child.result == ResultType.PASS:
has_pass = True
if has_block:
self.result = ResultType.BLOCK
elif has_error:
self.result = ResultType.ERROR
elif has_failed:
self.result = ResultType.FAIL
elif has_pass:
self.result = ResultType.PASS
else:
self.result = ResultType.INFO
class CaseStepEntry(CaseEntry):
"""
测试步骤节点
"""
def __init__(self, headline, parent=None, message="", step_prefix="", step_no=1, _continue=False):
super().__init__(headline, parent, message)
self.result = None
self.step_prefix = step_prefix
self.step_no = step_no
self._continue = _continue
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
self.update_result()
if self.result is None:
self.result = ResultType.PASS
if self.update_action:
self.update_action()
return True
if exc_type is StepEnd:
self.result = exc_val.result
if self.result == ResultType.PASS:
return True
else:
return self._continue
else:
self.result = ResultType.ERROR
self.message += str(exc_tb)
return self._continue
def __str__(self):
if self.step_prefix == "COLLECT_RESOURCE" or self.step_prefix == "SETUP" or self.step_prefix == "TEST" or self.step_prefix == "CLEANUP":
headline = self.step_prefix + "[" + time.strftime(TIME_FORMAT, self.timestamp) + "]"
else:
headline = "STEP-" + self.step_prefix + str(self.step_no) + ": " + self.headline + \
"[" + time.strftime(TIME_FORMAT, self.timestamp) + "]"
return headline
def start(self, headline, message, _continue=False, prefix=None):
entry = CaseStepEntry(headline=headline, parent=self, message=message, _continue=_continue)
if prefix is not None:
entry.step_prefix = prefix
elif self.step_prefix == "SETUP" or self.step_prefix == "TEST" or self.step_prefix == "CLEANUP" or self.step_prefix == "COLLECT_RESOURCE":
entry.step_prefix = ""
else:
entry.step_prefix = self.step_prefix + str(self.step_no) + "-"
if any(self.children):
entry.step_no = self.children[-1].step_no + 1
entry.update_action = self.update_action
if self.update_action is not None:
self.update_action()
self.children.append(entry)
return entry
def get_json(self):
json_obj = super().get_json()
if self.step_prefix == "SETUP" or self.step_prefix == "TEST" or self.step_prefix == "CLEANUP":
json_obj['headline'] = self.step_prefix
else:
json_obj['headline'] = "STEP-" + self.step_prefix + str(self.step_no) + self.headline
return json_obj
class StepReporter:
"""
测试结果，用单例实现
"""
instance = None
path = ""
json_path = ""
txt_path = ""
@classmethod
def get_instance(cls, logger):
# cls.path = ("%s/testresult/%s/%s/%s" % (FwConfig.result_path,
# RuntimeData.uuid,
# FwConfig.branch,
# FwConfig.app_num))
cls.path = "/home/emc/"
cls.json_path = os.path.join(cls.path, "step_result.json")
cls.txt_path = os.path.join(cls.path, "step_result.txt")
if cls.instance is None:
return StepReporter(logger)
else:
return cls.instance
def __init__(self, logger):
self.logger = logger
self.case_node = None
self.root = NodeEntry(headline="Test Result")
self.recent_node = self.root
def update_file(self):
with open(self.json_path, "w") as jsonfile:
json.dump(self.root.get_json(), jsonfile, indent=4)
with open(self.txt_path, "w") as txtfile:
txtfile.write(self.root.get_friend_print())
def start_node(self, headline, message):
ret = NodeEntry(headline, parent=self.root, message=message)
self.root.children.append(ret)
return ret
def print(self):
print(self.root.get_friend_print())
if __name__ == "__main__":
logger = logging.getLogger("单元测试")
rr = StepReporter.get_instance(logger)
with rr.root.start_node("测试列表") as testlist:
with testlist.start_case("test_feature_001") as case:
with case.start(headline="", message="", prefix="SETUP") as step:
with step.start("设置浏览器", "启动浏览器，选择chrome") as substep:
substep.passed("成功设置浏览器")
with step.start("登录系统", "输入用户名密码") as substep:
substep.passed("登录成功")
with case.start(headline="", message="", prefix="TEST") as step:
with step.start("测试步骤1", "第一个测试步骤") as substep:
with substep.start("一个pass的子步骤", "") as ssubstep:
pass
with step.start("一个异常的步骤", "", _continue=True) as substep:
1/0 # 造成异常
with step.start("继续测试步骤", "") as substep:
substep.passed("成功")
with case.start(headline="", message="", prefix="CLEANUP") as step:
step.passed("执行清理")
print(rr.root.get_friend_print())
"""
# 输出
Test Result[2023-02-05 09:41:09]
测试列表[2023-02-05 09:41:09]
test_feature_001[2023-02-05 09:41:09]-------------------------------------------------------ERRORED
SETUP[2023-02-05 09:41:09]--------------------------------------------------------------PASSED
STEP-1: 设置浏览器[2023-02-05 09:41:09]--------------------------------------------------PASSED
STEP-2: 登录系统[2023-02-05 09:41:09]---------------------------------------------------PASSED
TEST[2023-02-05 09:41:09]---------------------------------------------------------------ERRORED
STEP-1: 测试步骤1[2023-02-05 09:41:09]--------------------------------------------------PASSED
STEP-1-1: 一个pass的子步骤[2023-02-05 09:41:09]---------------------------------------PASSED
STEP-2: 一个异常的步骤[2023-02-05 09:41:09]------------------------------------------------ERRORED
STEP-3: 继续测试步骤[2023-02-05 09:41:09]-------------------------------------------------PASSED
CLEANUP[2023-02-05 09:41:09]------------------------------------------------------------PASSED
"""