Source code for testbot.case.base

#!/usr/bin/env python
# -*- coding: utf-8 -*-


[docs]__author__ = "Nuanguang Gu(Sunny)"
[docs]__email__ = "nuanguang.gu@aliyun.com"
import os import json import traceback from enum import IntEnum from abc import abstractmethod, ABCMeta from testbot.resource.pool import ResourcePool from testbot.result.logger import logger_manager from testbot.result.testreporter import StepReporter from testbot.result.testreporter import CaseEntry from testbot.config import RUNNER_LOGS_PATH from testbot.resource.error import ResourceNotMeetConstraintError
[docs]class TestType(IntEnum): """ 测试类型分类 * 类型编码规则 每个类型对应一个8位的二进制编码,前4位二进制表示主类,后4位二进制表示次类。如单元测试类型为0b00010000,0001为主类编码,0000为次类编码 * 主类类型及编码 测试类型的主类有:单元测试(0000)、沙盒测试(0001)、集成测试(0010)、冒烟测试(0011)、系统测试(0100)、稳定性测试(0101)、性能测试(0110)、点检测试(0111)、接口测试(1000)、专项测试(1001)、通用测试(1111)等 * 次类类型及编码 测试类型的次类,是主类类型的进一步分类,如系统冒烟测试,属于大类冒烟测试(0011),是其次类的一种类型,其测试类型编码为00110001 * 测试类型列表 ================================================ ================================================ ================================================ ================================================ 测试类型名称(主类型) 测试类型名称(次类型) 测试类型代码 测试类型编码 ================================================ ================================================ ================================================ ================================================ 单元测试 UNIT_TEST 0b00000000 沙盒测试 SANITY_TEST 0b00010000 集成测试 INTEGRATION_TEST 0b00100000 冒烟测试 SMOKE_TEST 0b00110000 - 系统冒烟测试 SMOKE_TEST_SYSTEM 0b00110001 - 中间件冒烟测试 SMOKE_TEST_MIDDLEWARE 0b00110010 系统测试 SYSTEM_TEST 0b01000000 稳定性测试 STABILITY_TEST 0b01000000 性能测试 PERFORMANCE_TEST 0b01100000 点检测试 CHECK_TEST 0b01110000 接口测试 INTERFACE_TEST 0b10000000 专项测试 SPECIAL_TEST 0b10000000 - 媒资专项测试 SPECIAL_TEST_MEDIA 0b10000001 通用测试 COMMON_TEST 0b11111111 ================================================ ================================================ ================================================ ================================================ """ # 单元测试
[docs] UNIT_TEST = 0b00000000
# 沙盒测试
[docs] SANITY_TEST = 0b00010000
# 集成测试
[docs] INTEGRATION_TEST = 0b00100000
# 冒烟测试
[docs] SMOKE_TEST = 0b00110000
# 系统冒烟测试
[docs] SMOKE_TEST_SYSTEM = 0b00110001
# 中间件冒烟测试
[docs] SMOKE_TEST_MIDDLEWARE = 0b00110010
# 系统测试
[docs] SYSTEM_TEST = 0b01000000
# 稳定性测试
[docs] STABILITY_TEST = 0b01010000
# 性能测试
[docs] PERFORMANCE_TEST = 0b01100000
# 点检测试
[docs] CHECK_TEST = 0b01110000
# 接口测试
[docs] INTERFACE_TEST = 0b10000000
# 专项测试
[docs] SPECIAL_TEST = 0b10000000
# 媒资专项测试
[docs] SPECIAL_TEST_MEDIA = 0b10000001
# 通用测试
[docs] COMMON_TEST = 0b11111111
[docs]class TestCaseBase(metaclass=ABCMeta): """ 测试用例基类 用户应该实现以下3个方法: * collect_resource: 初始化资源对象 * setup: 测试执行之前的初始化工作 * test: 测试执行体 * cleanup: 测试执行之后的清理工作 """ def __init__(self, **kwargs): self.reporter = kwargs.get("reporter", StepReporter.get_instance(logger=logger_manager.register("CaseRunner", filename=os.path.join(RUNNER_LOGS_PATH, "CaseRunner.log"), default_level="INFO", for_test=True))) self.logger = self.reporter.logger self._output_var = dict() self.setting = None self.test_data_var = dict() self.result = None @abstractmethod
[docs] def collect_resource(self, node: CaseEntry, pool: ResourcePool): """ 初始化资源对象 :param pool: 资源池 :type pool: ResourcePool :return: :rtype: """ pass
[docs] def setup_class(self, node: CaseEntry, **kwargs): """ 执行测试之前的初始化工作 :param args: :return: """ pass
@abstractmethod
[docs] def setup(self, node: CaseEntry, **kwargs): """ 执行测试的每个循环之前的初始化工作 :param args: :return: """ pass
@abstractmethod
[docs] def test(self, node: CaseEntry, **kwargs): """ 测试执行体 :param args: :return: """ pass
@abstractmethod
[docs] def cleanup(self, node: CaseEntry, **kwargs): """ 执行每个循环之后的清理工作 :param args: :return: """ pass
[docs] def cleanup_class(self, node: CaseEntry, **kwargs): """ 执行测试之后的清理工作 :param args: :return: """ pass
@property
[docs] def output_var(self): """ The test case output variable Can be collected by Test Engine :return: """ return self._output_var
[docs] def get_setting(self, setting_path, filename): """ 获取测试用例配置文件实例 """ for k,v in self.__class__.__dict__.items(): if hasattr(v, "__base__") and v.__base__.__name__ == "TestSettingBase": self.setting = v(setting_path=setting_path, filename=filename) self.setting.load()
[docs] def _run_case(self, pool: ResourcePool, node: CaseEntry): """ 测试用例执行线程 """ node.info(message=self.__class__.__name__) _continue = True with node.start(headline="收集测试资源", message="", prefix="COLLECT_RESOURCE") as node2: pass try: self.logger.info(f"self.resource_pool={pool.topology}") self.collect_resource(node=node2, pool=pool) except ResourceNotMeetConstraintError as ex: traceinfo = "".join(traceback.format_exception(type(ex), ex, ex.__traceback__)) node2.info(message=f"测试资源不满足条件: {traceinfo}") _continue = False except Exception as ex: traceinfo = "".join(traceback.format_exception(type(ex), ex, ex.__traceback__)) self.logger.error(traceinfo) node2.info(message=f"捕获异常: {traceinfo}") _continue = False finally: pass if not _continue: return with node.start(headline="初始化前置条件", message="", prefix="SETUP") as node2: try: self.setup(node=node2) except Exception as ex: traceinfo = "".join(traceback.format_exception(type(ex), ex, ex.__traceback__)) self.logger.error(traceinfo) node2.info(message=f"捕获异常: {traceinfo}") self.__call_cleanup(node=node) return with node.start(headline="执行测试主体", message="", prefix="TEST") as node2: try: iterations = getattr(self.setting, "iterations", 1) self.logger.info(f"执行次数:{iterations}") for iteration in range(iterations): with node2.start(headline=f"执行测试主体第{iteration}次", message="", prefix=f"TEST-{iteration}") as node3: self.test(node=node3) except Exception as ex: traceinfo = "".join(traceback.format_exception(type(ex), ex, ex.__traceback__)) self.logger.error(traceinfo) node2.info(message=f"捕获异常: {traceinfo}") self.__call_cleanup(node=node) return self.__call_cleanup(node=node)
[docs] def __call_cleanup(self, node: CaseEntry): """ 执行清除操作 """ with node.start(headline="清理后置条件", message="", prefix="CLEANUP") as node2: try: self.cleanup(node=node2) except Exception as ex: traceinfo = "".join(traceback.format_exception(type(ex), ex, ex.__traceback__)) self.logger.error(traceinfo) node2.info(message=f"捕获异常: {traceinfo}") finally: pass
[docs] def start(self): pool = ResourcePool() reporter = StepReporter.get_instance(logger=self.logger) with reporter.root.start_node(headline="执行测试节点", message="").start_case(headline="执行测试用例") as node: self._run_case(pool=pool, node=node) self.logger.info(reporter.root.get_friend_print()) self.logger.info(json.dumps(reporter.root.to_dict()))