Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- class TestHardPDAQCrash(TestLiveControlWithPDAQ):
- def test(self):
- assert 'OK' == livecmd("config set -i default")
- assert 'OK' == livecmd("altconfigs add -i alt1")
- assert 'OK' == livecmd("start daq")
- # Simulate hard crash of pDAQ:
- self.toy.close()
- # Lexical closure to collect complete set of log messages:
- def message_updater():
- msgs = []
- def update_msgs():
- msgs.extend(drain_queue(self.sink))
- def messages_received_so_far():
- update_msgs()
- return msgs
- return messages_received_so_far
- allmsgs = message_updater()
- def correct_recovery_failure_messages_generated():
- all = allmsgs()
- moni = [x["payload"] for x in all if x["cmd"] == "moni"]
- logs = [x["value"] for x in moni if x["varname"] == "log"]
- recovery_fail_msgs = filter(
- lambda x: "failed and did not generate any events" in x,
- logs)
- failed_configs = [re.findall("config (\S+) failed", m)[0]
- for m in recovery_fail_msgs]
- return failed_configs == (["default" for _ in range(4)] +
- ["alt1" for _ in range(4)])
- misc.wait_until(correct_recovery_failure_messages_generated)
Add Comment
Please, Sign In to add comment