Skip to content

Commit

Permalink
doc: add get_filtered_output method documentation
Browse files Browse the repository at this point in the history
Add documentation for the get_filtered_output method in both the Python
implementation and the Integration with pytest chapter. The method is used
to filter command outputs by removing prompts and log messages, making it
easier to process shell command results.

Signed-off-by: Thomas Günther <[email protected]>

removed traling whitespace
  • Loading branch information
tguenth authored and nashif committed Dec 6, 2024
1 parent 9e908b1 commit 20409ca
Show file tree
Hide file tree
Showing 2 changed files with 40 additions and 11 deletions.
2 changes: 2 additions & 0 deletions doc/develop/test/pytest.rst
Original file line number Diff line number Diff line change
Expand Up @@ -219,6 +219,8 @@ Shell

.. automethod:: wait_for_prompt

.. automethod:: get_filtered_output


Examples of pytest tests in the Zephyr project
**********************************************
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,9 @@ class Shell:
Helper class that provides methods used to interact with shell application.
"""

def __init__(self, device: DeviceAdapter, prompt: str = 'uart:~$', timeout: float | None = None) -> None:
def __init__(
self, device: DeviceAdapter, prompt: str = 'uart:~$', timeout: float | None = None
) -> None:
self._device: DeviceAdapter = device
self.prompt: str = prompt
self.base_timeout: float = timeout or device.base_timeout
Expand All @@ -48,7 +50,9 @@ def wait_for_prompt(self, timeout: float | None = None) -> bool:
return True
return False

def exec_command(self, command: str, timeout: float | None = None, print_output: bool = True) -> list[str]:
def exec_command(
self, command: str, timeout: float | None = None, print_output: bool = True
) -> list[str]:
"""
Send shell command to a device and return response. Passed command
is extended by double enter sings - first one to execute this command
Expand All @@ -63,20 +67,42 @@ def exec_command(self, command: str, timeout: float | None = None, print_output:
self._device.write(command_ext.encode())
lines: list[str] = []
# wait for device command print - it should be done immediately after sending command to device
lines.extend(self._device.readlines_until(regex=regex_command, timeout=1.0, print_output=print_output))
lines.extend(
self._device.readlines_until(
regex=regex_command, timeout=1.0, print_output=print_output
)
)
# wait for device command execution
lines.extend(self._device.readlines_until(regex=regex_prompt, timeout=timeout, print_output=print_output))
lines.extend(
self._device.readlines_until(
regex=regex_prompt, timeout=timeout, print_output=print_output
)
)
return lines

def get_filtered_output(self, command_lines: list[str]) -> list[str]:
"""
Filter out prompts and log messages
Take the output of exec_command, which can contain log messages and command prompts,
and filter them to obtain only the command output.
Example:
>>> # equivalent to `lines = shell.exec_command("kernel version")`
>>> lines = [
>>> 'uart:~$', # filter prompts
>>> 'Zephyr version 3.6.0', # keep this line
>>> 'uart:~$ <dbg> debug message' # filter log messages
>>> ]
>>> filtered_output = shell.get_filtered_output(output)
>>> filtered_output
['Zephyr version 3.6.0']
:param command_lines: List of strings i.e. the output of `exec_command`.
:return: A list of strings containing, excluding prompts and log messages.
"""
regex_filter = re.compile(
'|'.join([
re.escape(self.prompt),
'<dbg>',
'<inf>',
'<wrn>',
'<err>'
])
'|'.join([re.escape(self.prompt), '<dbg>', '<inf>', '<wrn>', '<err>'])
)
return list(filter(lambda l: not regex_filter.search(l), command_lines))

Expand Down Expand Up @@ -106,6 +132,7 @@ class ShellMCUbootCommandParsed:
"""
Helper class to keep data from `mcuboot` shell command.
"""

areas: list[ShellMCUbootArea] = field(default_factory=list)

@classmethod
Expand Down

0 comments on commit 20409ca

Please sign in to comment.