diff --git a/images_build/client/dign_client/bin/client.py b/images_build/client/dign_client/bin/client.py index ab67b6d..1f37a2c 100644 --- a/images_build/client/dign_client/bin/client.py +++ b/images_build/client/dign_client/bin/client.py @@ -5,6 +5,7 @@ import pycurl import os import re import time +import io from io import BytesIO import getopt import ciunittest @@ -967,8 +968,6 @@ class SslUnitTest(unittest.TestCase): class TsgDiagnoseRun: def __init__(self): self.interval = 1 - self.format = "txt" - self.write = None self.loop = False self.count = 1 self.config = None @@ -979,14 +978,10 @@ class TsgDiagnoseRun: parser = argparse.ArgumentParser(description="Tsg Tools - tsg diagnose", epilog = "Example:help") parser.add_argument('-i','--interval', type = int, default = 30,help='Wait interval seconds between each tsg disagnose. The default is to wait for 30 seconds between each tsg diagnose.') parser.add_argument('-c','--count', type = int, default = 1, help='Specifies the count of tsg diagnoses ,range:1-65535') - parser.add_argument('-f','--format', type = str, default = 'txt',help='Specifies the result output format of the tsg diagnose. There two formats: json,txt, the default is txt.') - parser.add_argument('-w','--write', type = str, default = None,help='Write out result into file or NEZHA. Specifies the output file name or NEZHA.') parser.add_argument('-p','--configpath', type = str, default = '/opt/dign_client/etc/client.conf',help='Specifies the config file, default /opt/dign_client/etc/client.conf') parser.add_argument('-l','--loop', action='store_true', default = False, help='Tsg diagnose loop, exit when recv a signal') args = parser.parse_args() self.interval = args.interval - self.format = args.format - self.write = args.write self.loop = args.loop self.count = args.count self.config = args.configpath @@ -994,16 +989,6 @@ class TsgDiagnoseRun: print("Error: bad number of tsg diagnose and will exit") parser.print_help() sys.exit(1) - - if self.format not in ('json', 'txt'): - print("Error: bad output format of tsg diagnose and will exit") - parser.print_help() - sys.exit(1) - - def _set_telegraf(self): - # self.client = TelegrafClient(host='192.51.100.1', port=8100,tags={'app_name':'tsg-diagnose'}) - self.client = TelegrafClient(host=str(self.config_dict['telegraf']['host']), port=int(self.config_dict['telegraf']['port']),tags={str(self.config_dict['telegraf']['tags_key']):str(self.config_dict['telegraf']['tags_value'])}) - def _get_suite_config(self): global suite_test_config_dict @@ -1064,76 +1049,18 @@ class TsgDiagnoseRun: self._add_suite('test_ssl_firewall_deny_drop') self._add_suite('test_ssl_firewall_deny_rst') - def _write_suite_result_into_file(self): - resultDict = '/opt/dign_client/log/unittest/' - resultNewestPath = resultDict + self.write - resultPath = resultDict + self.write + "." + time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime()) - if self.format == 'txt': - with open(resultNewestPath,"w+") as f: - runner = unittest.TextTestRunner(stream=f,verbosity=2) - runner.run(self.suite) - f.close() - - if self.format == 'json': - self.suite = unittest.TestLoader().loadTestsFromTestCase(SslUnitTest) - result_json = ciunittest.JsonTestRunner().run(self.suite, formatted=True) - with open(resultNewestPath,"w+") as f: - f.write(result_json) - f.close() - - with open(resultPath,"w+") as f: - fn = open(resultNewestPath,'r') - f.write(fn.read()) - fn.close() - f.close() - - def _write_suite_result_into_NEZHA(self): - nzdict = {} - self.suite = unittest.TestLoader().loadTestsFromTestCase(SslUnitTest) - result_json = ciunittest.JsonTestRunner().run(self.suite, formatted=False) - result_dict = json.loads(result_json) - reuslt_list = result_dict['results'] - succsum = 0 - failsum = 0 - for reuslt in reuslt_list: - succkey = reuslt['name'].split()[0] + '_succ' - nzdict[succkey] = 0 - if reuslt['type'] == 'success': - nzdict[succkey] = 1 - succsum = succsum + 1 - if reuslt['type'] == 'failure': - failsum = failsum + 1 - - nzdict['succsum'] = succsum - self.client.metric('tsg_diagnose_result', nzdict) - result_dict['succsum'] = succsum - result_dict['failsum'] = failsum - result_stdout = json.dumps(result_dict) - print(result_stdout) - def _stdout_suite_result(self): print(format(("Test start time: " + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())),'#^120s')) - if self.format == 'txt': - runner = unittest.TextTestRunner(verbosity=2) - runner.run(self.suite) - if self.format == 'json': - self.suite = unittest.TestLoader().loadTestsFromTestCase(SslUnitTest) - result_json = ciunittest.JsonTestRunner().run(self.suite, formatted=True) - print(result_json) + runner = unittest.TextTestRunner(verbosity=2) + runner.run(self.suite) print(format(("Test end time: " + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())),'=^120s')) def _output_suite_result(self): - if self.write and self.write != 'NEZHA': - self._write_suite_result_into_file() - elif self.write == 'NEZHA': - self._write_suite_result_into_NEZHA() - else: - self._stdout_suite_result() + self._stdout_suite_result() def execute_suite_tsg_diagnose(self): self._get_suite_option() self._get_suite_config() - self._set_telegraf() self._init_suite() try: if int(self.config_dict['start_time_random_delay_range']['enabled']) == 1: