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

feature: support export summary info.

Browse Source
This commit is contained in:
fumingwei
2024-06-04 09:51:42 +08:00
parent 78f1f97553
commit 476ed0aab0
1 changed files with 21 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -356,13 +356,13 @@ class ServerAddressBuilder:
return [f'{self.IPv4_1_TO_3TH_OCTET}.{self._ipv4_4th_octet}'] return [f'{self.IPv4_1_TO_3TH_OCTET}.{self._ipv4_4th_octet}']
class TcpPacketsCapture: class TcpPacketsCapture:
IFACE = "net1"
DSCP_KEY = ["DSCP"] DSCP_KEY = ["DSCP"]
def __init__(self, server_ip, server_port): def __init__(self, server_ip, server_port):
self._server_ip = server_ip self._server_ip = server_ip
self._server_port = server_port self._server_port = server_port
self._quadruple_to_dscp_table = {} self._quadruple_to_dscp_table = {}
self._build_filter() self._build_filter()
self._iface_names = get_if_list()
# def _read_server_ip_and_port(self): # def _read_server_ip_and_port(self):
@@ -370,7 +370,7 @@ class TcpPacketsCapture:
self._filter = f"tcp and src host {self._server_ip} and src port {self._server_port}" self._filter = f"tcp and src host {self._server_ip} and src port {self._server_port}"
def start(self): def start(self):
self._sniff_thread = AsyncSniffer(iface=self.IFACE, prn=self._packet_callback, filter=self._filter) self._sniff_thread = AsyncSniffer(iface=self._iface_names, prn=self._packet_callback, filter=self._filter)
self._sniff_thread.start() self._sniff_thread.start()
def _packet_callback(self, packet): def _packet_callback(self, packet):
@@ -1152,6 +1152,13 @@ class FirewallCasesRunner:
return True, None return True, None
class ResultExportBuilder: class ResultExportBuilder:
COLUMN_0 = "Test cases"
COLUMN_1 = "Result"
COLUMN_2 = "Fail Reason"
RESULT_OK = "ok"
RESULT_FAIL = "FAIL"
DEFAULT_INFO = "-"
def __init__(self): def __init__(self):
self._exporter = None self._exporter = None
self._create_exporter() self._create_exporter()
@@ -1159,25 +1166,25 @@ class ResultExportBuilder:
def _create_exporter(self): def _create_exporter(self):
self._exporter = PrettyTable() self._exporter = PrettyTable()
self._exporter.vrules = NONE self._exporter.vrules = NONE
self._exporter.field_names = ["Case name", "Status", "Description"] self._exporter.field_names = [self.COLUMN_0, self.COLUMN_1, self.COLUMN_2]
self._exporter.align["Case name"] = "l" self._exporter.align[self.COLUMN_0] = "l"
self._exporter.align["Status"] = "l" self._exporter.align[self.COLUMN_1] = "l"
self._exporter.align["Description"] = "l" self._exporter.align[self.COLUMN_2] = "l"
def append_case_result(self, case_name, case_result, enable_ouptut_by_case=True): def append_case_result(self, case_name, case_result, enable_ouptut_by_case=True):
status, description = self._convert_case_result(case_result) status, description = self._convert_case_result(case_result)
if enable_ouptut_by_case: if enable_ouptut_by_case:
self._output_by_case_name(case_name, status) self._output_by_case_name(case_name, status)
if not case_result[0]: #if not case_result[0]:
self._add_exporter_row(case_name, status, description) self._add_exporter_row(case_name, status, description)
def _convert_case_result(self, case_result): def _convert_case_result(self, case_result):
status = "ok" status = self.RESULT_OK
if not case_result[0]: if not case_result[0]:
status = "FAIL" status = self.RESULT_FAIL
description = "-" description = self.DEFAULT_INFO
if case_result[1] is not None: if case_result[1] is not None:
description = case_result[1] description = case_result[1]
@@ -1186,11 +1193,12 @@ class ResultExportBuilder:
def _output_by_case_name(self, case_name, case_status): def _output_by_case_name(self, case_name, case_status):
print(f"{case_name} ... {case_status}") print(f"{case_name} ... {case_status}")
def _add_exporter_row(self, case_name: str, status: str, description: str): def _add_exporter_row(self, column0: str, column1: str, column2: str):
self._exporter.add_row([case_name, status, description]) self._exporter.add_row([column0, column1, column2])
def export(self): def export(self):
if len(self._exporter._rows) > 0: if len(self._exporter._rows) > 0:
print("\nSummary:")
print(self._exporter) print(self._exporter)