| # Copyright (c) 2023 Nordic Semiconductor ASA |
| # |
| # SPDX-License-Identifier: Apache-2.0 |
| |
| from __future__ import annotations |
| |
| import logging |
| import os |
| import shutil |
| import signal |
| import subprocess |
| import threading |
| import time |
| from datetime import datetime |
| from pathlib import Path |
| from queue import Empty, Queue |
| from typing import Generator |
| |
| import psutil |
| |
| from twister_harness.constants import QEMU_FIFO_FILE_NAME |
| from twister_harness.device.device_abstract import DeviceAbstract |
| from twister_harness.device.fifo_handler import FifoHandler |
| from twister_harness.exceptions import TwisterHarnessException |
| from twister_harness.helper import log_command |
| from twister_harness.log_files.log_file import HandlerLogFile |
| from twister_harness.twister_harness_config import DeviceConfig |
| |
| logger = logging.getLogger(__name__) |
| |
| |
| class QemuAdapter(DeviceAbstract): |
| """Adapter for Qemu simulator""" |
| |
| def __init__(self, device_config: DeviceConfig, **kwargs) -> None: |
| super().__init__(device_config, **kwargs) |
| self._process: subprocess.Popen | None = None |
| self._process_ended_with_timeout: bool = False |
| self._exc: Exception | None = None #: store any exception which appeared running this thread |
| self._thread: threading.Thread | None = None |
| self._emulation_was_finished: bool = False |
| self.connection = FifoHandler(Path(self.device_config.build_dir).joinpath(QEMU_FIFO_FILE_NAME)) |
| self.command: list[str] = [] |
| self.timeout: float = 60 # running timeout in seconds |
| self.booting_timeout_in_ms: int = 10_000 #: wait time for booting Qemu in milliseconds |
| |
| def generate_command(self) -> None: |
| """Return command to flash.""" |
| if (west := shutil.which('west')) is None: |
| logger.error('west not found') |
| self.command = [] |
| else: |
| self.command = [west, 'build', '-d', str(self.device_config.build_dir), '-t', 'run'] |
| |
| def connect(self, timeout: float = 1) -> None: |
| logger.debug('Opening connection') |
| self.connection.connect() |
| |
| def flash_and_run(self, timeout: float = 60.0) -> None: |
| self.timeout = timeout |
| if not self.command: |
| msg = 'Run simulation command is empty, please verify if it was generated properly.' |
| logger.error(msg) |
| raise TwisterHarnessException(msg) |
| |
| self._thread = threading.Thread(target=self._run_command, args=(self.timeout,), daemon=True) |
| self._thread.start() |
| # Give a time to start subprocess before test is executed |
| time.sleep(0.1) |
| # Check if subprocess (simulation) has started without errors |
| if self._exc is not None: |
| logger.error('Simulation failed due to an exception: %s', self._exc) |
| raise self._exc |
| |
| def _run_command(self, timeout: float) -> None: |
| log_command(logger, 'Running command', self.command, level=logging.INFO) |
| try: |
| self._process = subprocess.Popen( |
| self.command, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.STDOUT, |
| env=self.env |
| ) |
| stdout, _ = self._process.communicate(timeout=timeout) |
| return_code: int = self._process.returncode |
| except subprocess.TimeoutExpired: |
| logger.error('Running simulation finished after timeout: %s seconds', timeout) |
| self._process_ended_with_timeout = True |
| # we don't want to raise Timeout exception, but allowed a test to parse the output |
| # and set proper status |
| except subprocess.SubprocessError as e: |
| logger.error('Running simulation failed due to subprocess error %s', e) |
| self._exc = TwisterHarnessException(e.args) |
| except FileNotFoundError as e: |
| logger.error(f'Running simulation failed due to file not found: {e.filename}') |
| self._exc = TwisterHarnessException(f'File not found: {e.filename}') |
| except Exception as e: |
| logger.error('Running simulation failed: %s', e) |
| self._exc = TwisterHarnessException(e.args) |
| else: |
| if return_code == 0: |
| logger.info('Running simulation finished with return code %s', return_code) |
| elif return_code == -15: |
| logger.info('Running simulation terminated') |
| else: |
| logger.warning('Running simulation finished with return code %s', return_code) |
| for line in stdout.decode('utf-8').split('\n'): |
| logger.info(line) |
| finally: |
| self._emulation_was_finished = True |
| |
| def disconnect(self): |
| logger.debug('Closing connection') |
| self.connection.disconnect() |
| |
| def stop(self) -> None: |
| """Stop device.""" |
| time.sleep(0.1) # give a time to end while loop in running simulation |
| if self._process is not None and self._process.returncode is None: |
| logger.debug('Stopping all running processes for PID %s', self._process.pid) |
| # kill all child subprocesses |
| for child in psutil.Process(self._process.pid).children(recursive=True): |
| try: |
| os.kill(child.pid, signal.SIGTERM) |
| except ProcessLookupError: |
| pass |
| # kill subprocess if it is still running |
| os.kill(self._process.pid, signal.SIGTERM) |
| if self._thread is not None: |
| self._thread.join(timeout=1) # Should end immediately, but just in case we set timeout for 1 sec |
| if self._exc: |
| raise self._exc |
| |
| def _wait_for_fifo(self): |
| for _ in range(int(self.booting_timeout_in_ms / 10) or 1): |
| if self.connection.is_open: |
| break |
| elif self._emulation_was_finished: |
| msg = 'Problem with starting QEMU' |
| logger.error(msg) |
| raise TwisterHarnessException(msg) |
| time.sleep(0.1) |
| else: |
| msg = 'Problem with starting QEMU - fifo file was not created yet' |
| logger.error(msg) |
| raise TwisterHarnessException(msg) |
| |
| @property |
| def iter_stdout(self) -> Generator[str, None, None]: |
| if not self.connection: |
| return |
| # fifo file can be not create yet, so we need to wait for a while |
| self._wait_for_fifo() |
| |
| # create unblocking reading from fifo file |
| q: Queue = Queue() |
| |
| def read_lines(): |
| while self.connection and self.connection.is_open: |
| try: |
| line = self.connection.readline().decode('UTF-8').strip() |
| except (OSError, ValueError): |
| # file could be closed already so we should stop reading |
| break |
| if len(line) != 0: |
| q.put(line) |
| |
| t = threading.Thread(target=read_lines, daemon=True) |
| t.start() |
| |
| end_time = time.time() + self.timeout |
| try: |
| while True: |
| try: |
| stream = q.get(timeout=0.1) |
| self.handler_log_file.handle(data=stream + '\n') |
| yield stream |
| except Empty: # timeout appeared |
| pass |
| if time.time() > end_time: |
| break |
| except KeyboardInterrupt: |
| # let thread to finish smoothly |
| pass |
| finally: |
| t.join(1) |
| |
| def write(self, data: bytes) -> None: |
| """Write data to serial""" |
| if self.connection: |
| self.connection.write(data) |
| |
| def initialize_log_files(self): |
| self.handler_log_file = HandlerLogFile.create(build_dir=self.device_config.build_dir) |
| start_msg = f'\n==== Logging started at {datetime.now()} ====\n' |
| self.handler_log_file.handle(start_msg) |
