From 479ed2e2a501ab0a7ccab5bd83eeb01f2c9f89b4 Mon Sep 17 00:00:00 2001 From: fumingwei Date: Tue, 7 Sep 2021 10:10:00 +0800 Subject: [PATCH] =?UTF-8?q?feature:=E6=96=B0=E5=A2=9EDignTextTestResult?= =?UTF-8?q?=E5=A4=84=E7=90=86=E8=87=AA=E6=A3=80=E7=BB=93=E6=9E=9C=E8=BE=93?= =?UTF-8?q?=E5=87=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- images_build/client/dign_client/bin/client.py | 143 +++++++++++++++++- 1 file changed, 141 insertions(+), 2 deletions(-) diff --git a/images_build/client/dign_client/bin/client.py b/images_build/client/dign_client/bin/client.py index 6011849..63eeb02 100644 --- a/images_build/client/dign_client/bin/client.py +++ b/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()