Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Allow test API command dispatcher override. Fixes nasa/fprime#2442 #158

Merged
merged 1 commit into from
Dec 20, 2023
Merged

Allow test API command dispatcher override. Fixes nasa/fprime#2442 #158

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions src/fprime_gds/common/testing_fw/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -402,7 +402,7 @@ def send_and_await_event(self, command, args=None, events=None, timeout=5):
return self.await_event_sequence(events, start=start, timeout=timeout)
return self.await_event(events, start=start, timeout=timeout)

def send_and_assert_command(self, command, args=[], max_delay=None, timeout=5, events=None):
def send_and_assert_command(self, command, args=[], max_delay=None, timeout=5, events=None, commander="cmdDisp"):
"""
This helper will send a command and verify that the command was dispatched and completed
within the F' deployment. This helper can retroactively check that the delay between
Expand All @@ -414,12 +414,13 @@ def send_and_assert_command(self, command, args=[], max_delay=None, timeout=5, e
max_delay: the maximum allowable delay between dispatch and completion (int/float)
timeout: the number of seconds to wait before terminating the search (int)
events: extra event predicates to check between dispatch and complete
commander: the command dispatching component. Defaults to cmdDisp
Return:
returns a list of the EventData objects found by the search
"""
cmd_id = self.translate_command_name(command)
dispatch = [self.get_event_pred("cmdDisp.OpCodeDispatched", [cmd_id, None])]
complete = [self.get_event_pred("cmdDisp.OpCodeCompleted", [cmd_id])]
dispatch = [self.get_event_pred(f"{commander}.OpCodeDispatched", [cmd_id, None])]
complete = [self.get_event_pred(f"{commander}.OpCodeCompleted", [cmd_id])]
events = dispatch + (events if events else []) + complete
results = self.send_and_assert_event(command, args, events, timeout=timeout)
if max_delay is not None:
Expand Down
Loading