当前位置：首页 > news >正文

python自动化测试——unittest二次开发之自定义测试用例执行器和测试结果记录器（二）

news 来源：原创 2024/4/26 18:20:36

CSDN话题挑战赛第2期
参赛话题：学习笔记

在这里插入图片描述

一、自定义用例执行器

重写TestRunner实现了
根据具体业务要求，初始化TestRunner对象时传入套件对象、并发线程数和并发执行的最小任务单元实现自动化测试。

class TestRunner:

    def __init__(self,suites,thread_count,task_unit='cls'):
        """
        :param suites: 套件对象
        :param task_unit: 并发执行的最小任务单元（cls、case、other）
        :param thread_count: 并发线程数
        """
        self.suites=suites
        self.task_unit=task_unit
        self.thread_count=thread_count

    def __parser_suite(self):
        """将套件拆分成并发执行的任务，放到一个列表中"""
        case_list=[]
        if self.task_unit=='case':
            for items in self.suites:
                for item in items:
                    for i in item:
                        case_list.append(i)
        elif self.task_unit=='cls':
            for items in self.suites:
                for i in items:
                    case_list.append(i)
        else:
            case_list=[i for i in self.suites]
        return case_list
    def run(self):
        """执行用例的启动方法"""

        tasks=self.__parser_suite()
        #result = unittest.TestResult()
        result = TestResult()
        result.startTestRun()
        with ThreadPoolExecutor(max_workers=self.thread_count) as tp:
            for i in tasks:
                tp.submit(i.run,result)
        result_info=result.stopTestRun()
        #获取用例执行的结果
        return result_info

if __name__ == '__main__':
    suites=unittest.defaultTestLoader.discover(r'D:\project_development\测试开发\day12\testcases')
    runner = TestRunner(suites,thread_count=8)
    result=runner.run()
    print(result)

执行结果：
…
…
…
result的输出结果：<unittest.result.TestResult run=32 errors=4 failures=4>

二、自定义测试结果记录器

测试结果记录器默认为：
<unittest.result.TestResult run=32 errors=4 failures=4>

重写TestResult类

__init__(self):
startTest(self, test: unittest.case.TestCase)：每次执行用例时自动调用的方法；
实现了每条用例开始执行时间
stopTest(self, test: unittest.case.TestCase)：每次用例执行完后自动调用的方法；
实现了每条用例执行完毕的时间
startTestRun(self)：所有用例开始执行之前执行的方法；
实现了用例执行的开始时间
stopTestRun(self)：所有用例执行完毕后执行的方法；
实现了统计所有的测试用例成功的总数、失败的总数、跳过的总数、错误的总数、所有用例执行完毕消耗时间、用例总数、用例的具体信息统计
addError(self, test, err)：用例执行错误自动执行的方法；
对错误的用例进行包装
addFailure(self, test, err)：用例断言错误自动执行的方法；
对失败的用例进行包装
addSuccess(self, test: unittest.case.TestCase)：用例执行成功自动执行的测试用例；
对执行成功的用例进行包装

class TestResult(unittest.TestResult):
    """测试结果记录器"""
    def __init__(self):
        super().__init__()
        self.success=0
        self.cases=[]

    def startTest(self, test: unittest.case.TestCase):
        """每次执行用例时调用的方法"""
        #统计用例的数量
        self.testsRun+=1
        print('开始执行用例:{}'.format(test))
        # 统计用例开始执行的时间
        test.start_time=time.time()


    def stopTest(self, test: unittest.case.TestCase):
        super().stopTest(test)
        """每次用例执行完后调用的方法"""
        #统计用例结束时间
        test.end_time=time.time()
        test.runTime=test.end_time-getattr(test,'start_time')
        print('用例执行时长:',test.runTime)


    def startTestRun(self):
        """所有用例开始执行之前执行的方法"""
        super().startTestRun()
        #用例执行的开始时间
        self.start_time=time.time()

    def stopTestRun(self):
        """所有用例执行完毕后执行的方法"""
        #super().stopTestRun()
        """统计和汇总执行结果"""
        result={
            "success":self.success,
            "errors":len(self.errors),
            "failures":len(self.failures),
            "skip":len(self.skipped),
            "RunTime":time.time()-self.start_time,
            "all_case":self.testsRun,
            "cases":self.cases
        }
        return result

    def addError(self, test, err):
        super().addError(test,err)
        info={
            "name":test._testMethodName,
            "res":"错误",
            "error":err[1]
        }
        self.cases.append(info)

    def addFailure(self, test, err):
        super().addFailure(test,err)
        info={
            "name":test._testMethodName,
            "res":"失败",
            "assertInfo":str(err[1])
        }
        self.cases.append(info)


    def addSuccess(self, test: unittest.case.TestCase):
        """用例执行成功自动执行的测试用例"""
        super().addSuccess(test)
        self.success+=1
        info={
            "name":test._testMethodName,
            "res":"通过"
        }
        self.cases.append(info)

三、实现业务需要

1、目录结果：

在这里插入图片描述

2、测试文件

test_demo1.py

import time
import unittest
from ddt import ddt,data

@ddt
class DemoA1(unittest.TestCase):
    @data(1,2,3,6)
    def test_demoa1(self,item):
        time.sleep(1)
        print('---------test_demoa1--------',item)


@ddt
class DemoA2(unittest.TestCase):
    @data(1,2,3,6)
    def test_demoa2(self,item):
        time.sleep(1)
        print('---------test_demoa2--------',item)

test_demo2.py

import time
import unittest
from ddt import ddt,data

@ddt
class DemoB1(unittest.TestCase):
    @data(1,2,3,6)
    def test_demob1(self,item):
        time.sleep(1)
        print('---------test_demob1--------',item)


@ddt
class DemoB2(unittest.TestCase):
    @data(1,2,3,6)
    def test_demob2(self,item):
        time.sleep(1)
        print('---------test_demob1--------',item)

test_demo3.py

test_demo4.py

3、代码实现

import unittest
from concurrent.futures.thread import ThreadPoolExecutor
import time

class TestResult(unittest.TestResult):
    """测试结果记录器"""
    def __init__(self):
        super().__init__()
        self.success=0
        self.cases=[]

    def startTest(self, test: unittest.case.TestCase):
        """每次执行用例时调用的方法"""
        #统计用例的数量
        self.testsRun+=1
        print('开始执行用例:{}'.format(test))
        # 统计用例开始执行的时间
        test.start_time=time.time()


    def stopTest(self, test: unittest.case.TestCase):
        super().stopTest(test)
        """每次用例执行完后调用的方法"""
        #统计用例结束时间
        test.end_time=time.time()
        test.runTime=test.end_time-getattr(test,'start_time')
        print('用例执行时长:',test.runTime)


    def startTestRun(self):
        """所有用例开始执行之前执行的方法"""
        super().startTestRun()
        #用例执行的开始时间
        self.start_time=time.time()

    def stopTestRun(self):
        """所有用例执行完毕后执行的方法"""
        #super().stopTestRun()
        """统计和汇总执行结果"""
        result={
            "success":self.success,
            "errors":len(self.errors),
            "failures":len(self.failures),
            "skip":len(self.skipped),
            "RunTime":time.time()-self.start_time,
            "all_case":self.testsRun,
            "cases":self.cases
        }
        return result

    def addError(self, test, err):
        super().addError(test,err)
        info={
            "name":test._testMethodName,
            "res":"错误",
            "error":err[1]
        }
        self.cases.append(info)

    def addFailure(self, test, err):
        super().addFailure(test,err)
        info={
            "name":test._testMethodName,
            "res":"失败",
            "assertInfo":str(err[1])
        }
        self.cases.append(info)


    def addSuccess(self, test: unittest.case.TestCase):
        """用例执行成功自动执行的测试用例"""
        super().addSuccess(test)
        self.success+=1
        info={
            "name":test._testMethodName,
            "res":"通过"
        }
        self.cases.append(info)


class TestRunner:

    def __init__(self,suites,thread_count,task_unit='cls'):
        """
        :param suites: 套件对象
        :param task_unit: 并发执行的最小任务单元（cls、case、other）
        :param thread_count: 并发线程数
        """
        self.suites=suites
        self.task_unit=task_unit
        self.thread_count=thread_count

    def __parser_suite(self):
        """将套件拆分成并发执行的任务，放到一个列表中"""
        case_list=[]
        if self.task_unit=='case':
            for items in self.suites:
                for item in items:
                    for i in item:
                        case_list.append(i)
        elif self.task_unit=='cls':
            for items in self.suites:
                for i in items:
                    case_list.append(i)
        else:
            case_list=[i for i in self.suites]
        return case_list
    def run(self):
        """执行用例的启动方法"""

        tasks=self.__parser_suite()
        #result = unittest.TestResult()
        result = TestResult()
        result.startTestRun()
        with ThreadPoolExecutor(max_workers=self.thread_count) as tp:
            for i in tasks:
                tp.submit(i.run,result)
        result_info=result.stopTestRun()
        #获取用例执行的结果
        return result_info

if __name__ == '__main__':
    suites=unittest.defaultTestLoader.discover(r'D:\project_development\测试开发\day12\testcases')
    runner = TestRunner(suites,thread_count=8)
    result=runner.run()
    print(result)

    # result=TestResult()
    # suite = unittest.defaultTestLoader.discover(r'D:\project_development\测试开发\day12\testcases\test_A')
    # suite.run(result)

4、测试结果输出

result={'success': 24,
     'errors': 4,
     'failures': 4,
     'skip': 0,
     'RunTime': 4.041905879974365,
     'all_case': 32,
     'cases': [
         {'name': 'test_demob2_1_1', 'res': '通过'}, {'name': 'test_demoa2_1_1', 'res': '通过'},
         {'name': 'test_demod1_1_1', 'res': '通过'}, {'name': 'test_demod2_1_1', 'res': '通过'},
         {'name': 'test_democ1_1_1', 'res': '通过'}, {'name': 'test_democ2_1_1', 'res': '通过'},
         {'name': 'test_demoa1_1_1', 'res': '错误',
          'error': AttributeError("'DemoA1' object has no attribute 'assertEqual1'")},
         {'name': 'test_demob1_1_1', 'res': '失败', 'assertInfo': '1 != 2'},
         {'name': 'test_demob2_2_2', 'res': '通过'},
         {'name': 'test_demoa2_2_2', 'res': '通过'},
         {'name': 'test_demob1_2_2', 'res': '失败', 'assertInfo': '1 != 2'},
         {'name': 'test_democ2_2_2', 'res': '通过'},
         {'name': 'test_demoa1_2_2', 'res': '错误',
          'error': AttributeError("'DemoA1' object has no attribute 'assertEqual1'")},
         {'name': 'test_demod2_2_2', 'res': '通过'},
         {'name': 'test_democ1_2_2', 'res': '通过'},
         {'name': 'test_demod1_2_2', 'res': '通过'},
         {'name': 'test_demod2_3_3', 'res': '通过'},
         {'name': 'test_demoa1_3_3', 'res': '错误',
          'error': AttributeError("'DemoA1' object has no attribute 'assertEqual1'")},
         {'name': 'test_demob2_3_3', 'res': '通过'},
         {'name': 'test_demod1_3_3', 'res': '通过'},
         {'name': 'test_democ1_3_3', 'res': '通过'},
         {'name': 'test_demoa2_3_3', 'res': '通过'},
         {'name': 'test_democ2_3_3', 'res': '通过'},
         {'name': 'test_demob1_3_3', 'res': '失败', 'assertInfo': '1 != 2'},
         {'name': 'test_demod2_4_6', 'res': '通过'},
         {'name': 'test_democ1_4_6', 'res': '通过'},
         {'name': 'test_demod1_4_6', 'res': '通过'},
         {'name': 'test_demoa2_4_6', 'res': '通过'},
         {'name': 'test_demob1_4_6', 'res': '失败', 'assertInfo': '1 != 2'},
         {'name': 'test_democ2_4_6', 'res': '通过'},
         {'name': 'test_demoa1_4_6', 'res': '错误',
          'error': AttributeError("'DemoA1' object has no attribute 'assertEqual1'")},
         {'name': 'test_demob2_4_6', 'res': '通过'}
     ]
     }