unittest用例实现重跑机制 最佳实现

-以前用java的时候实现过testng的失败重跑，今天看见有人问python，unittest的重跑，而且是完成测试结束后下次运行时，可以只运行失败的用例。
-他这个需求其实不难实现，网上搜索的到的方案，大都是使用pickle进行序列化，然后下次运行时反序列化失败的testcase重新运行。
-不过今天的博文并不是实现他的这个需求。因为在搜索过程中看到了一些讲解python失败retry机制的实现，不过很多都存在问题的。
-下面我会贴出两段由本人编写并经过本人测试的代码
-这是一段完全由本人自行编写的重试机制的实现源码

# coding=utf-8
import unittest
import sys
import warnings
from unittest.case import _ExpectedFailure, SkipTest, _UnexpectedSuccess


class MyTestCase(unittest.TestCase):
    def __init__(self, methodName='runTest', retryMax=3):
        super(MyTestCase, self).__init__(methodName)
        self.retry = 0  # 当前重试序号
        self.retryMax = retryMax  # 最大重试次数，不包括必要的第一次执行

    def run(self, result=None):
        is_need_stoptest = True  # 修改了失败时的处理，增加了is_need_*来控制tearDown和stopTest
        is_need_teardown = True
        orig_result = result
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            if startTestRun is not None:
                startTestRun()

        self._resultForDoCleanups = result
        result.startTest(self)

        testMethod = getattr(self, self._testMethodName)
        if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
            # If the class or method was skipped.
            try:
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                self._addSkip(result, skip_why)
            finally:
                result.stopTest(self)
            return
        try:
            success = False

            try:
                self.setUp()
            except SkipTest as e:
                self._addSkip(result, str(e))
            except KeyboardInterrupt:
                raise
            except:
                result.addError(self, sys.exc_info())
            else:
                try:
                    testMethod()
                except KeyboardInterrupt:
                    raise
                except self.failureException:
                    # 此处是失败时的处理
                    if self.retry <= self.retryMax:
                        result.addSkip(self, result._exc_info_to_string(sys.exc_info(), self))
                        self.retry += 1
                        self.tearDown()
                        result.stopTest(self)
                        is_need_teardown = False
                        is_need_stoptest = False
                        self.run(result)
                    else:
                        result.addFailure(self, sys.exc_info())
                except _ExpectedFailure as e:
                    addExpectedFailure = getattr(result, 'addExpectedFailure', None)
                    if addExpectedFailure is not None:
                        addExpectedFailure(self, e.exc_info)
                    else:
                        warnings.warn("TestResult has no addExpectedFailure method, reporting as passes",
                                      RuntimeWarning)
                        result.addSuccess(self)
                except _UnexpectedSuccess:
                    addUnexpectedSuccess = getattr(result, 'addUnexpectedSuccess', None)
                    if addUnexpectedSuccess is not None:
                        addUnexpectedSuccess(self)
                    else:
                        warnings.warn("TestResult has no addUnexpectedSuccess method, reporting as failures",
                                      RuntimeWarning)
                        result.addFailure(self, sys.exc_info())
                except SkipTest as e:
                    self._addSkip(result, str(e))
                except:
                    result.addError(self, sys.exc_info())
                else:
                    success = True

                try:
                    if is_need_teardown:  # 此处增加条件判断
                        self.tearDown()
                except KeyboardInterrupt:
                    raise
                except:
                    result.addError(self, sys.exc_info())
                    success = False

            cleanUpSuccess = self.doCleanups()
            success = success and cleanUpSuccess
            if success:
                result.addSuccess(self)
        finally:
            if is_need_stoptest:  # 此处增加判断
                result.stopTest(self)
            if orig_result is None:
                stopTestRun = getattr(result, 'stopTestRun', None)
                if stopTestRun is not None:
                    stopTestRun()

-上述代码更改处均已说明，核心是继承了TestCase并重写了run方法，run方法的主要流程仍沿用原代码，只是更改了测试失败时的处理。因为采用的递归实现重复，为了使其代码逻辑保持正确，增加了对tearDown和stopTest的处理。该源码可完整保持测试用例中setUpClass,teardownClass,setUp,tearDowm的功能，重试机制中，未达到最大重试次数的case均标记为skip，达到最大重试次数的case如仍抱错，则fail，若运行过程中，先出现执行断言错误，后续重试时又执行成功，则该用例通过。通过unittest. TestCase=MyTestCase可使所有测试类均使用修改后的测试类。同时以继承方式可以继续支持二次开发，以达到向html报告中输出更多信息，这是使用装饰器所不能达到的。
下面给出一段使用示例：

import unittest
import ddt

unittest.TestCase = MyTestCase
flag = 0

data = [
    [1, 2, 3],
    ["a", "b", "c"],
    [(1, 2, 3), "s", "k"]
]


@ddt.ddt
class MyClass(unittest.TestCase):
    def setUp(self):
        print "setup"

    def tearDown(self):
        print "tearDown"

    @ddt.data(*data)
    def test_001(self, value):
        global flag
        print (value, flag)
        flag += 1
        if flag % 4 != 0:
            assert False

    def test_002(self):
        assert False


class MyClass1(unittest.TestCase):
    def test_001(self):
        assert False

    def test_002(self):
        print "test_002"

-下面是一段由本人编写的使用装饰器来实现重试机制的源码

# coding=utf-8
import sys
import functools
import traceback


def retry_method(n=0):  # n为重试次数,不包括必要的第一次执行
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            num = 0
            while num <= n:
                try:
                    num += 1
                    func(*args, **kwargs)
                    return
                except AssertionError:
                    if num <= n:
                        trace = sys.exc_info()
                        traceback_info = ""
                        for trace_line in traceback.format_exception(trace[0], trace[1], trace[2], 3):
                            traceback_info += trace_line
                        print traceback_info
                        args[0].tearDown()
                        args[0].setUp()
                    else:
                        raise

        return wrapper

    return decorator

将该装饰器装饰在可能运行过程中出现AssertionError的测试方法上，则可以实现assert False时自动重新运行。该装饰器支持普通测试方法，支持ddt，支持parmunittest，效果非常好。与继承unittest.TestCase方式下相比，在后期扩展性稍显不足，但功能上是一样的。超过最大重试次数是抛出异常。
下面给出一段使用示例：

class TestA(unittest.TestCase):
    @retry_method(3)
    def test_001(self):
        assert False

下面再提供一个用于类的装饰器重试机制实现

# coding=utf-8
import functools
import sys
import traceback


def retry_class(n=0, prefix="test"):  # n为重试次数,不包括必要的第一次执行
    def retry(cls):
        for name, func in list(cls.__dict__.items()):
            if hasattr(func, "__call__") and name.startswith(prefix):
                setattr(cls, name, retry_method(n)(func))

        return cls

    return retry

该类装饰器会查找该类下以test开头的方法，并自动包装这些测试方法，以达到实现重试机制的目的。请注意，该类装饰器是以方法装饰器为基础进行设计的。
下面给出一个使用示例

@retry_class(3)
class TestA(unittest.TestCase):
    def test_001(self):
        assert False

一个小提示：本例的retry_class与retry_method同时使用时，是会同时生效的。

2020.7.3更新：
最近工作上的事情较少，本人回顾一些以前写的内容，发现有很大的改进空间，特意对本文进行了补充。
以下包含两个实现用例重跑机制的装饰器，一个是函数式装饰器，一个是类式装饰器，它们的功能是一样的，仅仅实现方式有所差异而已。它们的具体用法已在doc中完整示例。

# coding=utf-8
import sys
import functools
import traceback
import inspect
import unittest


def retry(target=None, max_n=1, func_prefix="test"):
    """
    一个装饰器，用于unittest执行测试用例出现失败后，自动重试执行

# example_1: test_001默认重试1次
class ClassA(unittest.TestCase):
    @retry
    def test_001(self):
        raise AttributeError


# example_2: max_n=2,test_001重试2次
class ClassB(unittest.TestCase):
    @retry(max_n=2)
    def test_001(self):
        raise AttributeError


# example_3: test_001重试3次; test_002重试3次
@retry(max_n=3)
class ClassC(unittest.TestCase):
    def test_001(self):
        raise AttributeError

    def test_002(self):
        raise AttributeError


# example_4: test_102重试2次, test_001不参与重试机制
@retry(max_n=2, func_prefix="test_1")
class ClassD(unittest.TestCase):
    def test_001(self):
        raise AttributeError

    def test_102(self):
        raise AttributeError


    :param target: 被装饰的对象，可以是class, function
    :param max_n: 重试次数，没有包含必须有的第一次执行
    :param func_prefix: 当装饰class时，可以用于标记哪些测试方法会被自动装饰
    :return: wrapped class 或 wrapped function
    """

    def decorator(func_or_cls):
        if inspect.isfunction(func_or_cls):
            @functools.wraps(func_or_cls)
            def wrapper(*args, **kwargs):
                n = 0
                while n <= max_n:
                    try:
                        n += 1
                        func_or_cls(*args, **kwargs)
                        return
                    except Exception:  # 可以修改要捕获的异常类型
                        if n <= max_n:
                            trace = sys.exc_info()
                            traceback_info = str()
                            for trace_line in traceback.format_exception(trace[0], trace[1], trace[2], 3):
                                traceback_info += trace_line
                            print(traceback_info)  # 输出组装的错误信息
                            args[0].tearDown()
                            args[0].setUp()
                        else:
                            raise

            return wrapper
        elif inspect.isclass(func_or_cls):
            for name, func in list(func_or_cls.__dict__.items()):
                if inspect.isfunction(func) and name.startswith(func_prefix):
                    setattr(func_or_cls, name, decorator(func))
            return func_or_cls
        else:
            raise AttributeError

    if target:
        return decorator(target)
    else:
        return decorator


class Retry(object):
    """
    类装饰器, 功能与Retry一样


# example_1: test_001默认重试1次
class ClassA(unittest.TestCase):
    @Retry
    def test_001(self):
        raise AttributeError


# example_2: max_n=2,test_001重试2次
class ClassB(unittest.TestCase):
    @Retry(max_n=2)
    def test_001(self):
        raise AttributeError


# example_3: test_001重试3次; test_002重试3次
@Retry(max_n=3)
class ClassC(unittest.TestCase):
    def test_001(self):
        raise AttributeError

    def test_002(self):
        raise AttributeError


# example_4: test_102重试2次, test_001不参与重试机制
@Retry(max_n=2, func_prefix="test_1")
class ClassD(unittest.TestCase):
    def test_001(self):
        raise AttributeError

    def test_102(self):
        raise AttributeError

    """

    def __new__(cls, func_or_cls=None, max_n=1, func_prefix="test"):
        self = object.__new__(cls)
        if func_or_cls:
            self.__init__(func_or_cls, max_n, func_prefix)
            return self(func_or_cls)
        else:
            return self

    def __init__(self, func_or_cls=None, max_n=1, func_prefix="test"):
        self._prefix = func_prefix
        self._max_n = max_n

    def __call__(self, func_or_cls=None):
        if inspect.isfunction(func_or_cls):
            @functools.wraps(func_or_cls)
            def wrapper(*args, **kwargs):
                n = 0
                while n <= self._max_n:
                    try:
                        n += 1
                        func_or_cls(*args, **kwargs)
                        return
                    except Exception:  # 可以修改要捕获的异常类型
                        if n <= self._max_n:
                            trace = sys.exc_info()
                            traceback_info = str()
                            for trace_line in traceback.format_exception(trace[0], trace[1], trace[2], 3):
                                traceback_info += trace_line
                            print(traceback_info)  # 输出组装的错误信息
                            args[0].tearDown()
                            args[0].setUp()
                        else:
                            raise

            return wrapper
        elif inspect.isclass(func_or_cls):
            for name, func in list(func_or_cls.__dict__.items()):
                if inspect.isfunction(func) and name.startswith(self._prefix):
                    setattr(func_or_cls, name, self(func))
            return func_or_cls
        else:
            raise AttributeError