-以前用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
