gfwleak/tsg-tsg-diagnose
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor:删除tsg-diagnose.py的write和format配置项

Browse Source
This commit is contained in:
fumingwei
2021-09-03 15:12:29 +08:00
parent 5a291e32ca
commit b7064c97b9
1 changed files with 4 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
images_build/client/dign_client/bin/client.py
View File

@@ -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: