python unittest 运行失败后重试

# -*- coding:utf-8 -*-
import unittest,time
from unittest.suite import _isnotsuite


class Suite(unittest.TestSuite):
    def run(self, result, debug=False):
        fail_count = 1
        class_num = 1
        topLevel = False
        if getattr(result, ‘_testRunEntered‘, False) is False:
            result._testRunEntered = topLevel = True

        for test in self:
            case_num = 1
            if result.shouldStop:
                break
            success_flag = True
            while success_flag:
                if _isnotsuite(test):
                    self._tearDownPreviousClass(test, result)
                    self._handleModuleFixture(test, result)
                    self._handleClassSetUp(test, result)
                    result._previousTestClass = test.__class__
                    if (getattr(test.__class__, ‘_classSetupFailed‘, False) or
                            getattr(result, ‘_moduleSetUpFailed‘, False)):
                        if class_num > fail_count:
                            success_flag = False
                        else:
                            time.sleep(5)
                            result._previousTestClass = None
                            print ‘%s Retrying init for %s times...‘ % (test.__class__, class_num)
                            class_num += 1
                        continue

                if not debug:
                    test(result)
                    print ‘1‘
                else:
                    test.debug()

                if result.result[-1][0] == 1 or result.result[-1][0] == 2:  # 1为Failed，2为Error
                    if case_num > fail_count:
                        success_flag = False
                    else:
                        print ‘%s is Failed ! Retrying %s times...‘ % (test, case_num)
                        case_num += 1
                else:
                    success_flag = False

            if topLevel:
                self._tearDownPreviousClass(None, result)
                self._handleModuleTearDown(result)
                result._testRunEntered = False
        return result

    def removeTest(self, test):
        self._tests.remove(test)

# -*- coding:utf-8 -*-
import unittest,time,settings
from utilities.send_mail import send_mail,new_report
import HTMLTestRunner,suite,os
from BeautifulSoup import BeautifulSoup


def get_result_html(file_dir):
    L = []
    for root, dirs, files in os.walk(file_dir):
        for file in files:
            if os.path.splitext(file)[1] == ‘.html‘:
                L.append(os.path.join(root, file))
    return L


def get_result_text():
    name = get_result_html(settings.RESULT_HTML_DIR)[-1]
    htmlfile = open(name, ‘r‘)
    htmlpage = htmlfile.read()
    soup = BeautifulSoup(htmlpage)

    pass_div = soup.findAll(‘tr‘, attrs={‘class‘: ‘passClass‘})
    pass_case_num = str(len(pass_div))

    file_obj = open(settings.RESULT_TEXT_DIR + ‘/test_result.txt‘, ‘wb+‘)
    file_obj.writelines(‘API TEST RESULT!‘)
    file_obj.writelines(‘\n\n\n\n\n\n\n\n\n‘)
    file_obj.writelines(‘SUCCESS:‘ + pass_case_num)
    file_obj.writelines(‘\n\n\n\n\n\n\n\n\n‘)
    file_obj.writelines(‘FAILED:‘ + str(int(len(case_name_list)) - int(pass_case_num)))
    file_obj.writelines(‘\n\n\n\n\n\n\n\n\n‘)
    file_obj.writelines(‘START_TIME:‘ + str(start_time))
    file_obj.writelines(‘\n\n\n\n\n\n\n\n\n‘)
    file_obj.writelines(‘END_TIME:‘ + str(end_time))
    file_obj.close()


if __name__ == ‘__main__‘:
    start_time = time.strftime("%Y-%m-%d %H:%M:%S")
    test_dir = settings.API_TEST_CASE_DIR
    suite = suite.Suite()
    file_name = ‘./result/‘ + time.strftime("%Y_%m_%d_%H_%M_%S") + ‘_‘ + settings.RESULT_FILE_NAME
    fp = open(file_name, ‘wb+‘)
    runner = HTMLTestRunner.HTMLTestRunner(stream=fp, title=settings.RESULT_TITLE,
                                               description=settings.RESULT_DESCRIPTION, verbosity=2)
    for case in test_dir:
        test_case = unittest.defaultTestLoader.discover(case,pattern=‘test_*‘,top_level_dir=os.path.abspath(‘./test_case/api_test_case/‘))
        case_name_list = []

        for x in [i._tests for i in test_case]:
            if not len(x) == 0:
                if len(x) == 1:
                    tds1 = x[0]
                    if not len(tds1._tests) == 0:
                        for y in tds1._tests:
                            case_name_list.append(y)

                elif len(x) == 2:
                    tds1 = x[0]
                    if not len(tds1._tests) == 0:
                        for y in tds1._tests:
                            case_name_list.append(y)
                    tds2 = x[1]
                    if not len(tds2._tests) == 0:
                        for y in tds2._tests:
                            case_name_list.append(y)

        for case_one in test_case._tests[0]._tests[1]:
            suite.addTest(case_one)
            runner.run(suite)
            suite.removeTest(case_one)
    fp.close()
    end_time = time.strftime("%Y-%m-%d %H:%M:%S")
    time.sleep(0.5)
    get_result_text()

    if not settings.ENV == ‘dev‘:
        new_report = new_report(settings.RESULT_TEXT_DIR)
        time.sleep(0.5)
        send_mail(new_report)