teaching/mosis24Statecharts
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix tester wrt. idempotency

Browse source
This commit is contained in:
Joeri Exelmans 2024-12-02 13:38:43 +01:00
parent d92587ac42
commit f6af9d01ae
2 changed files with 118 additions and 62 deletions
Download patch file Download diff file Expand all files Collapse all files

88
StartingPoint/lib/test.py
View file

@ -1,42 +1,54 @@
import abc
from difflib import ndiff
from lib.controller import Controller, pretty_time
from lib.tracer import Tracer
from lib.yakindu_helpers import YakinduTimerServiceAdapter, trace_output_events
# Can we ignore event in 'trace' at position 'idx' with respect to idempotency?
def can_ignore(trace, idx, IDEMPOTENT):
(timestamp, event_name, value) = trace[idx]
if event_name in IDEMPOTENT:
# If the same event occurred earlier, with the same parameter value, then this event can be ignored:
for (earlier_timestamp, earlier_event_name, earlier_value) in reversed(trace[0:idx]):
if (earlier_event_name, earlier_value) == (event_name, value):
# same event name and same parameter value (timestamps allowed to differ)
return True
elif event_name == earlier_event_name:
# same event name, but different parameter value:
# stop looking into the past:
break
# If the same event occurs later event, but with the same timestamp, this event is overwritten and can be ignored:
for (later_timestamp, later_event_name, later_value) in trace[idx+1:]:
if (later_timestamp, later_event_name) == (timestamp, event_name):
# if a later event with same name and timestamp occurs, ours will be overwritten:
return True
if later_timestamp != timestamp:
# no need to look further into the future:
break
return False
class AbstractEnvironmentState:
# should return the new state after handling the event
@abc.abstractmethod
def handle_event(self, event_name, param):
pass
# should compare states *by value*
@abc.abstractmethod
def __eq__(self, other):
pass
def postprocess_trace(trace, INITIAL, IDEMPOTENT):
# Prepend trace with events that set assumed initial state:
result = [(0, event_name, value) for (event_name, value) in INITIAL] + trace
# # Can we ignore event in 'trace' at position 'idx' with respect to idempotency?
# def can_ignore(trace, idx):
# (timestamp, event_name, value) = trace[idx]
# if event_name in IDEMPOTENT:
# # If the same event occurred earlier, with the same parameter value, then this event can be ignored:
# for (earlier_timestamp, earlier_event_name, earlier_value) in reversed(trace[0:idx]):
# if (earlier_event_name, earlier_value) == (event_name, value):
# # same event name and same parameter value (timestamps allowed to differ)
# return True
# elif event_name == earlier_event_name:
# # same event name, but different parameter value:
# # stop looking into the past:
# break
# # If the same event occurs later event, but with the same timestamp, this event is overwritten and can be ignored:
# for (later_timestamp, later_event_name, later_value) in trace[idx+1:]:
# if (later_timestamp, later_event_name) == (timestamp, event_name):
# # if a later event with same name and timestamp occurs, ours will be overwritten:
# return True
# if later_timestamp != timestamp:
# # no need to look further into the future:
# break
# return False
def postprocess_trace(trace, environment_class):
env_state = environment_class()
filtered_trace = []
# Remove events that have no effect:
while True:
filtered = [tup for (idx, tup) in enumerate(result) if not can_ignore(result, idx, IDEMPOTENT)]
# Keep on filtering until no more events could be removed:
if len(filtered) == len(result):
return filtered
result = filtered
for timestamp, event_name, param in trace:
new_env_state = env_state.handle_event(event_name, param)
if new_env_state != env_state:
# event had an effect
filtered_trace.append((timestamp, event_name, param))
env_state = new_env_state
return filtered_trace
def compare_traces(expected, actual):
i = 0
@ -56,7 +68,7 @@ def compare_traces(expected, actual):
print("Traces match.")
return True
def run_scenario(input_trace, expected_output_trace, statechart_class, INITIAL, IDEMPOTENT, verbose=False):
def run_scenario(input_trace, expected_output_trace, statechart_class, environment_class, verbose=False):
controller = Controller()
sc = statechart_class()
tracer = Tracer(verbose=False)
@ -81,11 +93,8 @@ def run_scenario(input_trace, expected_output_trace, statechart_class, INITIAL,
actual_output_trace = tracer.output_events
clean_expected = postprocess_trace(expected_output_trace, INITIAL, IDEMPOTENT)
clean_actual = postprocess_trace(actual_output_trace, INITIAL, IDEMPOTENT)
# clean_expected = expected_output_trace
# clean_actual = actual_output_trace
clean_expected = postprocess_trace(expected_output_trace, environment_class)
clean_actual = postprocess_trace(actual_output_trace, environment_class)
def print_diff():
# The diff printed will be a diff of the 'raw' traces, not of the cleaned up traces
@ -128,6 +137,7 @@ def run_scenario(input_trace, expected_output_trace, statechart_class, INITIAL,
print("\n\"Useless events\" are ignored by the comparison algorithm, and will never cause your test to fail. In this assignment, your solution is allowed to contain useless events.")
if not compare_traces(clean_expected, clean_actual):
# even though we compared the 'normalized' traces, we print the *raw* traces, not to confuse the user!
print("Raw diff between expected and actual output event trace:")
print_diff()
return False
@ -135,11 +145,11 @@ def run_scenario(input_trace, expected_output_trace, statechart_class, INITIAL,
print_diff()
return True
def run_scenarios(scenarios, statechart_class, initial, idempotent, verbose=True):
def run_scenarios(scenarios, statechart_class, environment_class, verbose=True):
ok = True
for scenario in scenarios:
print(f"Running scenario: {scenario["name"]}")
ok = run_scenario(scenario["input_events"], scenario["output_events"], statechart_class, initial, idempotent, verbose=verbose) and ok
ok = run_scenario(scenario["input_events"], scenario["output_events"], statechart_class, environment_class, verbose=verbose) and ok
print("--------")
if ok:
print("All scenarios passed.")

92
StartingPoint/runner_tests.py
View file

@ -1,5 +1,6 @@
import functools
from lib.test import run_scenarios
import dataclasses
from lib.test import run_scenarios, AbstractEnvironmentState
from srcgen.lock_controller import LockController
# from srcgen.solution import Solution as LockController # Teacher's solution
@ -206,27 +207,72 @@ SCENARIOS = [
}
]
# The following output events are safe to repeat: (with same value)
# This will be taken into account while comparing traces.
# Do not change this:
IDEMPOTENT = [
"open_doors",
"close_doors",
"red_light",
"green_light",
"set_request_pending",
"open_flow",
"close_flow",
]
# We pretend that initially, these events occur:
# Do not change this:
INITIAL = [
("open_doors", 0),
("close_doors", 1),
("green_light", 0),
("red_light", 1),
("set_request_pending", False)
]
LOW = 0
HIGH = 1
# Simulated state of the 'plant'.
# This is used for checking whether an event has any effect wrt. idempotency
@dataclasses.dataclass
class PlantState(AbstractEnvironmentState):
# initial state of the plant
door_low_open: bool = False
door_high_open: bool = False
flow_low_open: bool = False
flow_high_open: bool = False
light_low: str = "RED"
light_high: str = "RED"
request_is_pending: bool = False
sensor_is_broken: bool = False
def handle_event(self, event_name, param):
if event_name == "open_doors":
if param == LOW:
return dataclasses.replace(self, door_low_open=True)
elif param == HIGH:
return dataclasses.replace(self, door_high_open=True)
else:
raise Exception(f"invalid param for event '{event_name}': {param}")
elif event_name == "close_doors":
if param == LOW:
return dataclasses.replace(self, door_low_open=False)
elif param == HIGH:
return dataclasses.replace(self, door_high_open=False)
else:
raise Exception(f"invalid param for event '{event_name}': {param}")
elif event_name == "open_flow":
if param == LOW:
return dataclasses.replace(self, flow_low_open=True)
elif param == HIGH:
return dataclasses.replace(self, flow_high_open=True)
else:
raise Exception(f"invalid param for event '{event_name}': {param}")
elif event_name == "close_flow":
if param == LOW:
return dataclasses.replace(self, flow_low_open=False)
elif param == HIGH:
return dataclasses.replace(self, flow_high_open=False)
else:
raise Exception(f"invalid param for event '{event_name}': {param}")
elif event_name == "green_light":
if param == LOW:
return dataclasses.replace(self, light_low="GREEN")
elif param == HIGH:
return dataclasses.replace(self, light_high="GREEN")
else:
raise Exception(f"invalid param for event '{event_name}': {param}")
elif event_name == "red_light":
if param == LOW:
return dataclasses.replace(self, light_low="RED")
elif param == HIGH:
return dataclasses.replace(self, light_high="RED")
else:
raise Exception(f"invalid param for event '{event_name}': {param}")
elif event_name == "set_request_pending":
return dataclasses.replace(self, request_is_pending=param)
elif event_name == "set_sensor_broken":
return dataclasses.replace(self, sensor_is_broken=param)
else:
raise Exception("don't know how to handle event:", event_name)
if __name__ == "__main__":
run_scenarios(SCENARIOS, LockController, INITIAL, IDEMPOTENT, verbose=False)
run_scenarios(SCENARIOS, LockController, PlantState, verbose=False)