From b015aec9db56bcd3bef151bd7ed01e35276450ce Mon Sep 17 00:00:00 2001 From: Leo Liu Date: Tue, 26 Mar 2024 10:45:38 +0800 Subject: [PATCH] add an option for abort control. --- ReadMe.md | 8 +++++--- fakecall.py | 12 ++++++++---- fakesniff.py | 12 ++++++++---- 3 files changed, 21 insertions(+), 11 deletions(-) diff --git a/ReadMe.md b/ReadMe.md index bf88668..4d2ad75 100644 --- a/ReadMe.md +++ b/ReadMe.md @@ -34,9 +34,9 @@ Note: **ptftpd** package should be installed (before running). fakecall.py, a companion for generic testbed ```sh -usage: fakecall.py [-h] [-v] [-a] [-l] [-e extra_trailing] [-t intermittent] - [-n name] [-d directory] [-f filename] [-i interpreted] - [-o oriented] [-r report] +usage: fakecall.py [-h] [-v] [-a] [-b] [-l] [-e extra_trailing] + [-t intermittent] [-n name] [-d directory] [-f filename] + [-i interpreted] [-o oriented] [-r report] CLI argument parsing @@ -44,6 +44,8 @@ optional arguments: -h, --help show this help message and exit -v, --verbose verbosity -a, --auto auto-mode + -b, --abort abort while exception encountered by default, + specified for persist (not abort) -l, --linefeed LF only mode (instead of both CR and LF) -e extra_trailing, --extra_trailing extra_trailing extra trailing string diff --git a/fakecall.py b/fakecall.py index e31d251..88259d3 100644 --- a/fakecall.py +++ b/fakecall.py @@ -9,12 +9,12 @@ from fakesniff import FakeSniff class FakeCall(FakeSniff): - def __init__(self, lf: bool = False, et: bool = False) -> None: + def __init__(self, lf: bool = False, et: bool = False, ab: bool = True) -> None: super().__init__() #variables for pattern matching self.patt["deli_lf"] = lf self.patt["deli_et"] = et - self.patt["abort"] = True + self.patt["abort"] = ab self.patt["capi"]["traffic_agent_reset"] = self.__silence self.patt["capi"]["traffic_agent_config"] = self.__silence self.patt["capi"]["traffic_agent_receive_start"] = self.__silence @@ -141,6 +141,10 @@ def find_interpreting_handle(dir: str = ".", alias: str = None) -> tuple: "--auto", action="store_true", help="auto-mode") + my_parser.add_argument("-b", + "--abort", + action="store_false", + help="abort while exception encountered by default, specified for persist (not abort)") my_parser.add_argument("-l", "--linefeed", action="store_true", @@ -207,7 +211,7 @@ def find_interpreting_handle(dir: str = ".", alias: str = None) -> tuple: logging.info("name: " + args.name) rpt = open(args.report, "w") fldr = FakeSniff.find_interpreting_directory(args.directory) - fc = FakeCall(lf = args.linefeed, et = args.extra_trailing) + fc = FakeCall(lf = args.linefeed, et = args.extra_trailing, ab = args.abort) for f in fldr: (hdl, filename) = FakeCall.find_interpreting_handle(f, args.name) for h in hdl: @@ -236,7 +240,7 @@ def find_interpreting_handle(dir: str = ".", alias: str = None) -> tuple: handle_invoke = args.oriented else: handle_invoke = args.oriented + ":" + args.interpreted.split(":")[1] - fc = FakeCall(lf = args.linefeed, et = args.extra_trailing) + fc = FakeCall(lf = args.linefeed, et = args.extra_trailing, ab = args.abort) time_begin = time.time() (ret, stat) = fc.interpret(dir = args.directory, fn = args.filename, handle = args.interpreted, handle_invoke = handle_invoke) time_end = time.time() diff --git a/fakesniff.py b/fakesniff.py index c9d16f5..5cbebef 100644 --- a/fakesniff.py +++ b/fakesniff.py @@ -168,6 +168,7 @@ def __invoke(self, argv: list) -> bool: #last state depends on the CAPI invocation result invoke_running_tmo: int = self.cfg["tmo_running"] invoke_result_tmo: int = self.cfg["tmo_result"] + capi: str = None ret: bool = False if self.cfg["telnet"] is True: try: @@ -178,13 +179,13 @@ def __invoke(self, argv: list) -> bool: self.cfg["object_invoke"].write(bytes(capi, "UTF-8")) rcv = self.cfg["object_invoke"].read_until((b"\r\n" if self.patt["deli_lf"] is False else b"\n"), invoke_running_tmo) if len(rcv) == 0: - raise Exception("Empty") + raise Exception("Empty (synchronous)") rsp = rcv.decode("UTF-8").rstrip().split(self.patt["deli_arg"]) if (rsp[0] == "status") and ("RUNNING" in rsp[1]): #status running shall be hidden rcv = self.cfg["object_invoke"].read_until((b"\r\n" if self.patt["deli_lf"] is False else b"\n"), invoke_result_tmo) if len(rcv) == 0: - raise Exception("Empty") + raise Exception("Empty (asynchronous)") rsp = rcv.decode("UTF-8").rstrip().split(self.patt["deli_arg"]) if len(rsp) >= 2: self.status["invoked"] = argv[0] @@ -193,7 +194,8 @@ def __invoke(self, argv: list) -> bool: ret = True ret = True except Exception as e: - logging.exception(e) + pass + #logging.exception(e) finally: if self.cfg["reuse"] is False: self.cfg["object_invoke"].close() @@ -201,7 +203,9 @@ def __invoke(self, argv: list) -> bool: else: pass if ret is False: - logging.error("INVOKE: " + argv[0]) + self.status["invoked"] = capi.strip() + self.status["returned"] = "\"\"" + self.status["silenced"] = False if self.patt["abort"] is False: ret = True else: