feature:新增DignTextTestResult处理自检结果输出

images_build/client/dign_client/bin/client.py

@@ -150,6 +150,143 @@ DnsAAAARequestFirewallAllow         = "Dns rdtype AAAA request data is sucess"
 DnsCNAMERequestFirewallAllow        = "Dns rdtype CNAME request data is sucess"    
 
 
+
+
+class _WritelnDecorator(object):
+    """Used to decorate file-like objects with a handy 'writeln' method"""
+    def __init__(self,stream):
+        self.stream = stream
+
+    def __getattr__(self, attr):
+        if attr in ('stream', '__getstate__'):
+            raise AttributeError(attr)
+        return getattr(self.stream,attr)
+
+    def writeln(self, arg=None):
+        if arg:
+            self.write(arg)
+        self.write('\n') # text-mode streams translate to \r\n if needed
+
+
+class DignTextTestResult(unittest.result.TestResult):
+    """A test result class that can print formatted text results to a stream.
+    Used by TextTestRunner.
+    """
+    separator1 = '=' * 70
+    separator2 = '-' * 70
+
+    def __init__(self, stream, descriptions, verbosity):
+        super(DignTextTestResult, self).__init__(stream, descriptions, verbosity)
+        self.stream = stream
+        self.dignStream = _WritelnDecorator(sys.stderr)
+        self.showAll = verbosity > 1
+        self.dots = verbosity == 1
+        self.descriptions = descriptions
+
+    def getDescription(self, test):
+        doc_first_line = test.shortDescription()
+        if self.descriptions and doc_first_line:
+            return '\n'.join((str(test), doc_first_line))
+        else:
+            return str(test)
+
+    def startTest(self, test):
+        super(DignTextTestResult, self).startTest(test)
+        if self.showAll:
+            self.stream.write(self.getDescription(test))
+            self.dignStream.write(self.getDescription(test))
+            self.stream.write(" ... ")
+            self.dignStream.write(" ... ")
+            self.stream.flush()
+            self.dignStream.flush()
+
+    def addSuccess(self, test):
+        super(DignTextTestResult, self).addSuccess(test)
+        if self.showAll:
+            self.stream.writeln("ok")
+            self.dignStream.writeln("ok")
+        elif self.dots:
+            self.stream.write('.')
+            self.dignStream.write('.')
+            self.stream.flush()
+            self.dignStream.flush()
+
+    def addError(self, test, err):
+        super(DignTextTestResult, self).addError(test, err)
+        if self.showAll:
+            self.stream.writeln("ERROR")
+            self.dignStream.writeln("ERROR")
+        elif self.dots:
+            self.stream.write('E')
+            self.dignStream.write('E')
+            self.stream.flush()
+            self.dignStream.flush()
+
+    def addFailure(self, test, err):
+        super(DignTextTestResult, self).addFailure(test, err)
+        if self.showAll:
+            self.stream.writeln("FAIL")
+            self.dignStream.writeln("FAIL")
+        elif self.dots:
+            self.stream.write('F')
+            self.dignStream.write('F')
+            self.stream.flush()
+            self.dignStream.flush()
+
+    def addSkip(self, test, reason):
+        super(DignTextTestResult, self).addSkip(test, reason)
+        if self.showAll:
+            self.stream.writeln("skipped {0!r}".format(reason))
+            self.dignStream.writeln("skipped {0!r}".format(reason))
+        elif self.dots:
+            self.stream.write("s")
+            self.dignStream.write("s")
+            self.stream.flush()
+            self.dignStream.flush()
+
+    def addExpectedFailure(self, test, err):
+        super(DignTextTestResult, self).addExpectedFailure(test, err)
+        if self.showAll:
+            self.stream.writeln("expected failure")
+            self.dignStream.writeln("expected failure")
+        elif self.dots:
+            self.stream.write("x")
+            self.dignStream.write("x")
+            self.stream.flush()
+            self.dignStream.flush()
+
+    def addUnexpectedSuccess(self, test):
+        super(DignTextTestResult, self).addUnexpectedSuccess(test)
+        if self.showAll:
+            self.stream.writeln("unexpected success")
+            self.dignStream.writeln("unexpected success")
+        elif self.dots:
+            self.stream.write("u")
+            self.dignStream.write("u")
+            self.stream.flush()
+            self.dignStream.flush()
+
+    def printErrors(self):
+        if self.dots or self.showAll:
+            self.stream.writeln()
+            self.dignStream.writeln()
+        self.printErrorList('ERROR', self.errors)
+        self.printErrorList('FAIL', self.failures)
+
+    def printErrorList(self, flavour, errors):
+        for test, err in errors:
+            self.stream.writeln(self.separator1)
+            self.dignStream.writeln(self.separator1)
+            self.stream.writeln("%s: %s" % (flavour,self.getDescription(test)))
+            self.dignStream.writeln("%s: %s" % (flavour,self.getDescription(test)))
+            self.stream.writeln(self.separator2)
+            self.dignStream.writeln(self.separator2)
+            self.stream.writeln("%s" % err)
+            self.dignStream.writeln("%s" % err)
+
+
+
+
 def get_logger(logPath,enableConsole=True):
     logger = logging.getLogger()
     fileHandler = logging.FileHandler(logPath)
@@ -1056,13 +1193,15 @@ class TsgDiagnose:
         self._add_suite('test_ssl_firewall_deny_rst')
 
     def _dign_running(self):
+        print(format(("Test start time: " + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())),'#^120s'))
         runningLogPath =  "/opt/dign_client/log/tsg-diagnose.log" + '.' + time.strftime("%Y-%m-%d", time.localtime())
-        runningLogger = get_logger(runningLogPath, True)
+        runningLogger = get_logger(runningLogPath, False)
         runningLogger.debug("Diagnose Start,the It will take up to %d seconds" %(self.dign_duration))
         result_stream = io.StringIO()
-        runner = unittest.TextTestRunner(stream=result_stream,verbosity=2)
+        runner = unittest.TextTestRunner(stream=result_stream,verbosity=2,resultclass=DignTextTestResult)
         runner.run(self.suite)
         runningLogger.debug("Diagnose end, Testing results:" + "\n" + result_stream.getvalue())
+        print(format(("Test  end  time: " + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())),'=^120s'))
 
     def dign_exec(self):
         self._get_dign_option()