多进程使用unittest并将测试类的self对象传入装饰器美化日志输出

目录结构

多进程入口 run.py

# -*- coding: utf-8 -* from test.test_demo import TestDemo import multiprocessing import unittest from multiprocessing import Process from utils.unittest_utils import ParametrizedTestCase class MyProcess(Process): def __init__(self, data): Process.__init__(self) self.data = data
        self.queue = data["queue"] def run(self): suite = unittest.TestSuite() ParametrizedTestCase.data = self.data # ParametrizedTestCase 该类主要功能是在调用addTest方法时添加额外参数使用 suite.addTest(ParametrizedTestCase.parametrize(TestDemo, param=self.data)) runner = unittest.TextTestRunner() # 运行测试类的测试方法 runner.run(suite) # 获取子进程通过队列产生的数据 result = self.queue.get() print(result) def create_process(data): process_instances = [] for d in data: instance = MyProcess(d) process_instances.append(instance) for instance in process_instances: # 默认调用instance 对象的run方法 instance.start() if __name__ == "__main__": # 使用队列获取子进程产生的数据 q = multiprocessing.Queue() process01 = { 'device_id': "192.168.224.150:5555", "queue": q } process02 = { 'device_id': "192.168.224.130:5555", "queue": q } create_process([process01, process02]) 

调用addTest方法额外传参的工具类ParametrizedTestCase （unittest_utils.py）

import unittest class ParametrizedTestCase(unittest.TestCase): """ TestCase classes that want to be parametrized should
        inherit from this class.
    """ def __init__(self, method_name='runTest', param=None): super(ParametrizedTestCase, self).__init__(method_name) self.param = param

    @staticmethod def parametrize(test_case_class, param=None): """ Create a suite containing all tests taken from the given
            subclass, passing them the parameter 'param'.
        """ test_loader = unittest.TestLoader() test_names = test_loader.getTestCaseNames(test_case_class) suite = unittest.TestSuite() for name in test_names: suite.addTest(test_case_class(name, param=param)) return suite 

美化日志输出功能（log.py）

# -*- coding: utf-8 -* import logging # colorlog需安装   pip install colorlog import colorlog import logging.handlers class Logger: # 不同级别的日志输出颜色不同 log_colors_config = { 'DEBUG': 'black', 'INFO': 'cyan', 'WARNING': 'yellow', 'ERROR': 'red', 'CRITICAL': 'red', } def __init__(self, name="root"): self._logger = logging.getLogger(name) formatter = colorlog.ColoredFormatter( '%(log_color)s[%(asctime)s] [%(levelname)s] - %(filename)s(line:%(lineno)d) - %(message)s', log_colors=self.log_colors_config) # 避免多个句柄重复输出log if not self._logger.handlers: handler_info = logging.handlers.TimedRotatingFileHandler('logs/info.log', when='D', interval=1) handler_warn = logging.handlers.TimedRotatingFileHandler('logs/warn.log', when='D', interval=1) handler_info.setFormatter(formatter) handler_warn.setFormatter(formatter) sh = logging.StreamHandler() # 往屏幕上输出 sh.setFormatter(formatter) # 设置屏幕上显示的格式 # 当handler的log level高于logger本身的log level时，此设置才会生效 handler_info.setLevel(logging.INFO) handler_warn.setLevel(logging.WARN) self._logger.addHandler(handler_info) self._logger.addHandler(handler_warn) self._logger.addHandler(sh) # 默认情况下，logger本身的log level是warn，为了让info handler的level等级生效， # 所以调低logger本身的level self._logger.setLevel(logging.INFO) def get_logger(self): return self._logger class LoggerDecorator(object): logger = Logger().get_logger() def __init__(self, func): # self.func即为类装饰器装饰的函数名 self.func = func def __call__(self, *args, **kwargs): # self.instance 是通过对象调用该类的属性， 而该类的属性是通过测试类TestDemo中动态添加的， # 即Decorator.instance = self self.logger.info(self.instance.device_id + " " + self.func.__name__ + " : start ...\n") # self.func(self.instance)相当于调用测试方法，默认有参数self，所以需传入参数self.instance self.func(self.instance) self.logger.info(self.instance.device_id + " " + self.func.__name__ + " : end ...") logger = Logger().get_logger() 

测试类（test_demo.py）

# -*- coding: utf-8 -* from utils.log import LoggerDecorator as Decorator, logger from utils.unittest_utils import ParametrizedTestCase class TestDemo(ParametrizedTestCase): def __init__(self, *args, **kwargs): super(TestDemo, self).__init__(*args, **kwargs) self.device_id = self.param["device_id"] self.queue = self.param["queue"] # **重要：给类装饰器动态添加属性instance ,该属性的值为self对象** Decorator.instance = self

    @Decorator def test_01_open_device(self): logger.info("open device ： %s" % self.device_id) @Decorator def test_02_deal_something(self): logger.info("[%s]deal something ... " % self.device_id) @Decorator def test_03_close_device(self): logger.info("close device ： [%s]" % self.device_id) # 子进程产生数据传输给主进程 self.queue.put({"name": "hello " + self.device_id}) 

结果输出

本文地址：https://blog.csdn.net/qq_27525737/article/details/109059302