diff --git a/src/zoix.py b/src/zoix.py index b773e4e..cee8e10 100644 --- a/src/zoix.py +++ b/src/zoix.py @@ -7,6 +7,7 @@ import re import enum +######## TEMPORARY ######## log = logging.getLogger("testcrush logger") log.setLevel(logging.DEBUG) log_stream = logging.StreamHandler(stream = sys.stdout) @@ -17,6 +18,7 @@ log_file.setFormatter(logging.Formatter('%(lineno)d:[%(levelname)s|%(module)s|%(funcName)s]: %(message)s')) log.addHandler(log_stream) log.addHandler(log_file) +########################### class LogicSimulation(enum.Enum): TIMEOUT = 0 # Endless loop @@ -33,6 +35,10 @@ class Compilation(enum.Enum): ZOIX_ERROR = 0 # stderr contains text SUCCESS = 1 # None of the above +class FaultSimulation(enum.Enum): + TIMEOUT = 0 # Wall-clock + FSIM_ERROR = 1 # stderr contains text + SUCCESS = 2 # None of the above class ZoixInvoker(): def __init__(self, fsim_threads : int) -> 'ZoixInvoker': @@ -107,6 +113,12 @@ def logic_simulate(self, *instructions : str, **kwargs) -> LogicSimulation: - success_regexp (re.regexp): A regular expression which is used for matching in every line of the `stdout` stream to signify the sucessfull completion of the logic simulation. + - tat_regexp_capture_group (int): An integer which corresponds + to the index of the TaT value on the custom regexp (if provided). + By default is 1, mapping to the default `success_regexp` group. + - tat_value (list): An empty list to store the value of the + tat after being successfully matched with `success_regexp`. + Pass-by-reference style. - Returns: - LogicSimulation (enum): - TIMEOUT: if user defined timeout has been triggered. @@ -121,10 +133,25 @@ def logic_simulate(self, *instructions : str, **kwargs) -> LogicSimulation: # $finish at simulation time XXXXXXYs # where X = a digit and Y = time unit. # Capturing of the simulation duration - # done for possible TaT purposes. TODO + # done for possible TaT purposes. success : re.regexp = kwargs.get("success_regexp", re.compile(r"\$finish[^0-9]+([0-9]+)[m|u|n|p]s", re.DOTALL)) + # By default, a single capturing group + # is expected in the regexp, which maps + # to the TaT value. If a custom regexp + # is provided however, with >1 groups, + # then the user must specify which is + # the expected capture group. + tat_capture_group : int = kwargs.get("tat_regexp_capture_group", 1) + + # An empty mutable container is expected + # to store the TaT value matched from + # the regexp. It is used like that to + # mimic a pass-by-reference and not alter + # the function's return value. + tat_value : list = kwargs.get("tat_value", []) + simulation_status = None for cmd in instructions: @@ -134,37 +161,67 @@ def logic_simulate(self, *instructions : str, **kwargs) -> LogicSimulation: if stderr and stderr != "TimeoutExpired": log.debug(f"Error during execution of {cmd}\n\ - ---------[MESSAGE]---------\n\ - {'-'.join(stderr.splitlines(keepends=True))}\n\ + ------[STDERR STREAM]------\n\ + {'-'.join(stderr.splitlines(keepends = True))}\n\ ---------------------------\n") simulation_status = LogicSimulation.SIM_ERROR break elif stderr == stdout == "TimeoutExpired": + simulation_status = LogicSimulation.TIMEOUT break for line in stdout.splitlines(): + log.debug(f"{cmd}: {line.rstrip()}") - if re.match(success, line): - log.debug(f"SIMULATION SUCCESSFULL: {re.match(success, line).groups()}") + end_reached : re.Match = re.search(success, line) + + if end_reached: + + test_application_time = end_reached.group(tat_capture_group) + + try: + tat_value.append(int(test_application_time)) + + except ValueError: + raise LogicSimulationException(f"Test application time was not correctly captured \ +'{test_application_time=}' and could not be converted to an integer. Perhaps there is something wrong with\ +your regular expression '{success}' ?") + + log.debug(f"SIMULATION SUCCESSFULL: {end_reached.groups()}") simulation_status = LogicSimulation.SUCCESS break if not simulation_status: - raise LogicSimulationException(f"Simulation status was not set during\ - the execution of {instructions}. Check the debug log for more information!") + raise LogicSimulationException(f"Simulation status was not set during \ +the execution of {instructions}. Is your regular expression correct? Check the debug log for more information!") return simulation_status + def fault_simulate(self, *instructions : str, **kwargs) -> FaultSimulation: + ... # TODO + def main(): """Sandbox/Testing Env""" A = ZoixInvoker(16) #A.logic_simulate("for i in $(seq 100000); do echo $i; done", timeout = 0.1) - res = A.logic_simulate("cat oksimlog.log", timeout = 60.0) + + tat_dict = { + "success_regexp" : re.compile(r"test application time = ([0-9]+)"), + "tat_regexp_capture_group" : 1, + "tat_value" : [] + } + + res = A.logic_simulate("cat logic_sim_excerpt.log", + timeout = 60.0, + **tat_dict) + + print(tat_dict['tat_value']) print(res) + if __name__ == "__main__": main()