michaelh/zephyr
1
0
Fork 0
Code Releases Activity

scripts: pytest: align adapters API

Browse source 
Select and place common code of three adapters (HardwareAdapter,
BinaryAdapterBase and QemuAdapter) into basic DeviceAdapter class.

Introduce new way of reading device output by run separate thread which
try to read device output and place it into internal python queue.
Thanks to this, now it is possible to create readline method for all
adapters, which can be unblock when timeout occur.

Collect all common steps which have to be done before setup device in
launch method. The same was done for teardown operations which were
placed into close method.

Additionally some protection mechanisms were introduced to prevent for
undesirable side-effects when user could try to launch to already
launched device or try to send some data to disconnected device.

iter_stdout method was replaced by two new methods: readline and
readlines. To make it possible to remove all read output from internal
buffer (queue), clear_buffer method was introduced.

Also unit tests were rewritten to work properly with current version
of adapters.

Signed-off-by: Piotr Golyzniak <piotr.golyzniak@nordicsemi.no>
This commit is contained in:
Piotr Golyzniak 2023-08-04 15:23:34 +02:00 • committed by Anas Nashif
commit f22c2d6388
17 changed files with 754 additions and 780 deletions
Download patch file Download diff file Expand all files Collapse all files

98
scripts/pylib/pytest-twister-harness/tests/device/qemu_adapter_test.py
View file

@ -3,98 +3,56 @@
# SPDX-License-Identifier: Apache-2.0
import os
import subprocess
from typing import Generator
from unittest import mock
from unittest.mock import patch
import pytest
from conftest import readlines_until
from twister_harness.device.qemu_adapter import QemuAdapter
from twister_harness.exceptions import TwisterHarnessException
from twister_harness.log_files.log_file import HandlerLogFile, NullLogFile
from twister_harness.twister_harness_config import DeviceConfig
@pytest.fixture(name='device')
def fixture_device_adapter(tmp_path) -> Generator[QemuAdapter, None, None]:
build_dir = tmp_path / 'build_dir'
adapter = QemuAdapter(DeviceConfig(build_dir=build_dir))
yield adapter
os.mkdir(build_dir)
device = QemuAdapter(DeviceConfig(build_dir=build_dir))
try:
adapter.stop() # to make sure all running processes are closed
except TwisterHarnessException:
pass
yield device
finally:
device.close() # to make sure all running processes are closed
@patch('shutil.which', return_value='/usr/bin/west')
def test_if_generate_command_creates_proper_command(patched_which):
adapter = QemuAdapter(DeviceConfig(build_dir='build_dir'))
adapter.generate_command()
assert adapter.command == ['/usr/bin/west', 'build', '-d', 'build_dir', '-t', 'run']
@patch('shutil.which', return_value='west')
def test_if_generate_command_creates_proper_command(patched_which, device: QemuAdapter):
device.device_config.build_dir = 'build_dir'
device.generate_command()
assert device.command == ['west', 'build', '-d', 'build_dir', '-t', 'run']
@patch('shutil.which', return_value=None)
def test_if_generate_command_creates_empty_listy_if_west_is_not_installed(patched_which):
adapter = QemuAdapter(DeviceConfig())
adapter.generate_command()
assert adapter.command == []
def test_if_qemu_adapter_raises_exception_for_empty_command(device) -> None:
device.command = []
exception_msg = 'Run simulation command is empty, please verify if it was generated properly.'
with pytest.raises(TwisterHarnessException, match=exception_msg):
device.flash_and_run()
def test_if_qemu_adapter_raises_exception_file_not_found(device) -> None:
device.command = ['dummy']
with pytest.raises(TwisterHarnessException, match='File not found: dummy'):
device.flash_and_run()
device.stop()
assert device._exc is not None
assert isinstance(device._exc, TwisterHarnessException)
@mock.patch('subprocess.Popen', side_effect=subprocess.SubprocessError(1, 'Exception message'))
def test_if_qemu_adapter_raises_exception_when_subprocess_raised_an_error(patched_run, device):
device.command = ['echo', 'TEST']
with pytest.raises(TwisterHarnessException, match='Exception message'):
device.flash_and_run()
device.stop()
def test_if_qemu_adapter_runs_without_errors(resources, tmp_path) -> None:
fifo_file_path = str(tmp_path / 'qemu-fifo')
def test_if_qemu_adapter_runs_without_errors(resources, device: QemuAdapter) -> None:
fifo_file_path = str(device.device_config.build_dir / 'qemu-fifo')
script_path = resources.joinpath('fifo_mock.py')
device = QemuAdapter(DeviceConfig(build_dir=str(tmp_path)))
device.connection_timeout = 1
device.booting_timeout_in_ms = 1000
# device.base_timeout = 1
# device.booting_timeout_in_ms = 1000
device.command = ['python', str(script_path), fifo_file_path]
device.connect()
device.initialize_log_files()
device.flash_and_run()
lines = list(device.iter_stdout)
device.launch()
lines = readlines_until(device=device, line_pattern='Namespaces are one honking great idea')
device.close()
assert 'Readability counts.' in lines
assert os.path.isfile(device.handler_log_file.filename)
with open(device.handler_log_file.filename, 'r') as file:
assert os.path.isfile(device.handler_log_path)
with open(device.handler_log_path, 'r') as file:
file_lines = [line.strip() for line in file.readlines()]
assert file_lines[-2:] == lines[-2:]
device.disconnect()
def test_if_qemu_adapter_finishes_after_timeout(device) -> None:
device.connection_timeout = 0.1
device.command = ['sleep', '0.3']
device.flash_and_run()
device.stop()
assert device._process_ended_with_timeout is True
def test_handler_and_device_log_correct_initialized_on_qemu(device, tmp_path) -> None:
device.device_config.build_dir = tmp_path
device.initialize_log_files()
assert isinstance(device.handler_log_file, HandlerLogFile)
assert isinstance(device.device_log_file, NullLogFile)
assert device.handler_log_file.filename.endswith('handler.log') # type: ignore[union-attr]
def test_if_qemu_adapter_raise_exception_due_to_no_fifo_connection(device: QemuAdapter) -> None:
device.base_timeout = 0.3
device.command = ['sleep', '1']
with pytest.raises(TwisterHarnessException, match='Problem with starting QEMU'):
device._flash_and_run()
device._close_device()
assert not os.path.exists(device._fifo_connection._fifo_in)
assert not os.path.exists(device._fifo_connection._fifo_out)