scripts/pylib/pytest-twister-harness/tests/device/binary_adapter_test.py

# Copyright (c) 2023 Nordic Semiconductor ASA
#
# SPDX-License-Identifier: Apache-2.0

import logging
import os
import subprocess
import time
from pathlib import Path
from typing import Generator
from unittest import mock

import pytest

from twister_harness.device.binary_adapter import (
    CustomSimulatorAdapter,
    NativeSimulatorAdapter,
    UnitSimulatorAdapter,
)
from twister_harness.exceptions import TwisterHarnessException, TwisterHarnessTimeoutException
from twister_harness.twister_harness_config import DeviceConfig


@pytest.fixture
def script_path(resources: Path) -> str:
    return str(resources.joinpath('mock_script.py'))


@pytest.fixture(name='device')
def fixture_device_adapter(tmp_path: Path) -> Generator[NativeSimulatorAdapter, None, None]:
    build_dir = tmp_path / 'build_dir'
    os.mkdir(build_dir)
    device = NativeSimulatorAdapter(DeviceConfig(build_dir=build_dir, type='native', base_timeout=5.0))
    try:
        yield device
    finally:
        device.close()  # to make sure all running processes are closed


@pytest.fixture(name='launched_device')
def fixture_launched_device_adapter(
    device: NativeSimulatorAdapter, script_path: str
) -> Generator[NativeSimulatorAdapter, None, None]:
    device.command = ['python3', script_path]
    try:
        device.launch()
        yield device
    finally:
        device.close()  # to make sure all running processes are closed


def test_if_binary_adapter_runs_without_errors(launched_device: NativeSimulatorAdapter) -> None:
    """
    Run script which prints text line by line and ends without errors.
    Verify if subprocess was ended without errors, and without timeout.
    """
    device = launched_device
    lines = device.readlines_until(regex='Returns with code')
    device.close()
    assert 'Readability counts.' in lines
    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:]


def test_if_binary_adapter_finishes_after_timeout_while_there_is_no_data_from_subprocess(
    device: NativeSimulatorAdapter, script_path: str
) -> None:
    """Test if thread finishes after timeout when there is no data on stdout, but subprocess is still running"""
    device.base_timeout = 0.3
    device.command = ['python3', script_path, '--long-sleep', '--sleep=5']
    device.launch()
    with pytest.raises(TwisterHarnessTimeoutException, match='Read from device timeout occurred'):
        device.readlines_until(regex='Returns with code')
    device.close()
    assert device._process is None
    with open(device.handler_log_path, 'r') as file:
        file_lines = [line.strip() for line in file.readlines()]
    # this message should not be printed because script has been terminated due to timeout
    assert 'End of script' not in file_lines, 'Script has not been terminated before end'


def test_if_binary_adapter_raises_exception_empty_command(device: NativeSimulatorAdapter) -> None:
    device.command = []
    exception_msg = 'Run command is empty, please verify if it was generated properly.'
    with pytest.raises(TwisterHarnessException, match=exception_msg):
        device._flash_and_run()


@mock.patch('subprocess.Popen', side_effect=subprocess.SubprocessError(1, 'Exception message'))
def test_if_binary_adapter_raises_exception_when_subprocess_raised_subprocess_error(
    patched_popen, device: NativeSimulatorAdapter
) -> None:
    device.command = ['echo', 'TEST']
    with pytest.raises(TwisterHarnessException, match='Exception message'):
        device._flash_and_run()


@mock.patch('subprocess.Popen', side_effect=FileNotFoundError(1, 'File not found', 'fake_file.txt'))
def test_if_binary_adapter_raises_exception_file_not_found(
    patched_popen, device: NativeSimulatorAdapter
) -> None:
    device.command = ['echo', 'TEST']
    with pytest.raises(TwisterHarnessException, match='fake_file.txt'):
        device._flash_and_run()


@mock.patch('subprocess.Popen', side_effect=Exception(1, 'Raised other exception'))
def test_if_binary_adapter_raises_exception_when_subprocess_raised_an_error(
    patched_popen, device: NativeSimulatorAdapter
) -> None:
    device.command = ['echo', 'TEST']
    with pytest.raises(TwisterHarnessException, match='Raised other exception'):
        device._flash_and_run()


def test_if_binary_adapter_connect_disconnect_print_warnings_properly(
    caplog: pytest.LogCaptureFixture, launched_device: NativeSimulatorAdapter
) -> None:
    device = launched_device
    assert device._device_connected.is_set() and device.is_device_connected()
    caplog.set_level(logging.DEBUG)
    device.connect()
    warning_msg = 'Device already connected'
    assert warning_msg in caplog.text
    for record in caplog.records:
        if record.message == warning_msg:
            assert record.levelname == 'DEBUG'
            break
    device.disconnect()
    assert not device._device_connected.is_set() and not device.is_device_connected()
    device.disconnect()
    warning_msg = 'Device already disconnected'
    assert warning_msg in caplog.text
    for record in caplog.records:
        if record.message == warning_msg:
            assert record.levelname == 'DEBUG'
            break


def test_if_binary_adapter_raise_exc_during_connect_read_and_write_after_close(
    launched_device: NativeSimulatorAdapter
) -> None:
    device = launched_device
    assert device._device_run.is_set() and device.is_device_running()
    device.close()
    assert not device._device_run.is_set() and not device.is_device_running()
    with pytest.raises(TwisterHarnessException, match='Cannot connect to not working device'):
        device.connect()
    with pytest.raises(TwisterHarnessException, match='No connection to the device'):
        device.write(b'')
    device.clear_buffer()
    with pytest.raises(TwisterHarnessException, match='No connection to the device and no more data to read.'):
        device.readline()


def test_if_binary_adapter_raise_exc_during_read_and_write_after_close(
    launched_device: NativeSimulatorAdapter
) -> None:
    device = launched_device
    device.disconnect()
    with pytest.raises(TwisterHarnessException, match='No connection to the device'):
        device.write(b'')
    device.clear_buffer()
    with pytest.raises(TwisterHarnessException, match='No connection to the device and no more data to read.'):
        device.readline()


def test_if_binary_adapter_is_able_to_read_leftovers_after_disconnect_or_close(
    device: NativeSimulatorAdapter, script_path: str
) -> None:
    device.command = ['python3', script_path, '--sleep=0.05']
    device.launch()
    device.readlines_until(regex='Beautiful is better than ugly.')
    time.sleep(0.1)
    device.disconnect()
    assert len(device.readlines()) > 0
    device.connect()
    device.readlines_until(regex='Flat is better than nested.')
    time.sleep(0.1)
    device.close()
    assert len(device.readlines()) > 0


def test_if_binary_adapter_properly_send_data_to_subprocess(
    shell_simulator_adapter: NativeSimulatorAdapter
) -> None:
    """Run shell_simulator.py program, send "zen" command and verify output."""
    device = shell_simulator_adapter
    time.sleep(0.1)
    device.write(b'zen\n')
    time.sleep(1)
    lines = device.readlines_until(regex='Namespaces are one honking great idea')
    assert 'The Zen of Python, by Tim Peters' in lines


def test_if_native_binary_adapter_get_command_returns_proper_string(device: NativeSimulatorAdapter) -> None:
    device.generate_command()
    assert isinstance(device.command, list)
    assert device.command == [str(device.device_config.build_dir / 'zephyr' / 'zephyr.exe')]


@mock.patch('shutil.which', return_value='west')
def test_if_custom_binary_adapter_get_command_returns_proper_string(patched_which, tmp_path: Path) -> None:
    device = CustomSimulatorAdapter(DeviceConfig(build_dir=tmp_path, type='custom'))
    device.generate_command()
    assert isinstance(device.command, list)
    assert device.command == ['west', 'build', '-d', str(tmp_path), '-t', 'run']


@mock.patch('shutil.which', return_value=None)
def test_if_custom_binary_adapter_raise_exception_when_west_not_found(patched_which, tmp_path: Path) -> None:
    device = CustomSimulatorAdapter(DeviceConfig(build_dir=tmp_path, type='custom'))
    with pytest.raises(TwisterHarnessException, match='west not found'):
        device.generate_command()


def test_if_unit_binary_adapter_get_command_returns_proper_string(tmp_path: Path) -> None:
    device = UnitSimulatorAdapter(DeviceConfig(build_dir=tmp_path, type='unit'))
    device.generate_command()
    assert isinstance(device.command, list)
    assert device.command == [str(tmp_path / 'testbinary')]