Skip to content

Pytest embedded qemu

pytest_embedded_qemu.app

IdfFlashImageMaker

Create a single image for QEMU based on the IdfApp's partition table and all the flash files.

Source code in pytest_embedded_qemu/app.py
class IdfFlashImageMaker:
    """
    Create a single image for QEMU based on the `IdfApp`'s partition table and all the flash files.
    """

    def __init__(self, app: IdfApp, image_path: str):
        """
        Args:
            app: `IdfApp` instance
            image_path: output image path
        """
        self.app = app
        self.image_path = image_path

    def make_bin(self) -> None:
        """
        Create a single image file for qemu.
        """
        # flash_files is sorted, if the first offset is not 0x0, we need to fill it with empty bin
        if self.app.flash_files[0][0] != 0x0:
            self._write_empty_bin(count=self.app.flash_files[0][0])
        for offset, file_path, encrypted in self.app.flash_files:
            if encrypted:
                raise NotImplementedError('will implement later')
            else:
                self._write_bin(file_path, seek=offset)

    def _write_empty_bin(self, count: int, bs: int = 1024, seek: int = 0):
        live_print_call(
            f'dd if=/dev/zero bs={bs} count={count} seek={seek} of={self.image_path}',
            shell=True,
        )

    def _write_bin(self, binary_filepath, bs: int = 1, seek: int = 0):
        live_print_call(
            f'dd if={binary_filepath} bs={bs} seek={seek} of={self.image_path} conv=notrunc',
            shell=True,
        )

    def _write_encrypted_bin(self, binary_filepath, bs: int = 1, seek: int = 0):
        live_print_call(
            'dd if=/dev/zero bs=1 count=32 of=key.bin',
            shell=True,
        )  # generate a fake key bin
        live_print_call(
            f'espsecure.py encrypt_flash_data --keyfile key.bin --output decrypted.bin --address {seek} '
            f'{binary_filepath}',
            shell=True,
        )
        self._write_bin('decrypted.bin', bs=bs, seek=seek)

    def _burn_efuse(self):
        pass

__init__(app, image_path)

Parameters:

Name Type Description Default
app IdfApp

IdfApp instance

required
image_path str

output image path

required
Source code in pytest_embedded_qemu/app.py
def __init__(self, app: IdfApp, image_path: str):
    """
    Args:
        app: `IdfApp` instance
        image_path: output image path
    """
    self.app = app
    self.image_path = image_path

make_bin()

Create a single image file for qemu.

Source code in pytest_embedded_qemu/app.py
def make_bin(self) -> None:
    """
    Create a single image file for qemu.
    """
    # flash_files is sorted, if the first offset is not 0x0, we need to fill it with empty bin
    if self.app.flash_files[0][0] != 0x0:
        self._write_empty_bin(count=self.app.flash_files[0][0])
    for offset, file_path, encrypted in self.app.flash_files:
        if encrypted:
            raise NotImplementedError('will implement later')
        else:
            self._write_bin(file_path, seek=offset)

QemuApp

Bases: IdfApp

QEMU App class

Attributes:

Name Type Description
pexpect_proc PexpectProcess

pexpect process

image_path str

QEMU flash-able bin path

Source code in pytest_embedded_qemu/app.py
class QemuApp(IdfApp):
    """
    QEMU App class

    Attributes:
        pexpect_proc (PexpectProcess): pexpect process
        image_path (str): QEMU flash-able bin path
    """

    def __init__(
        self,
        pexpect_proc: PexpectProcess,
        app_path: Optional[str] = None,
        build_dir: Optional[str] = None,
        part_tool: Optional[str] = None,
        qemu_image_path: Optional[str] = None,
        skip_regenerate_image: Optional[bool] = False,
        **kwargs,
    ):
        """
        Args:
            pexpect_proc: pexpect process
            app_path: App path
            build_dir: Build directory
            part_tool: Partition tool path
            qemu_image_path: QEMU flashable bin path
            skip_regenerate_image: skip regenerate QEMU image
        """
        super().__init__(app_path, build_dir=build_dir, part_tool=part_tool, **kwargs)
        self.pexpect_proc = pexpect_proc
        self.image_path = qemu_image_path or os.path.join(self.app_path, self.build_dir, DEFAULT_IMAGE_FN)
        self.skip_regenerate_image = skip_regenerate_image

        if self.target != 'esp32':
            raise ValueError('For now on QEMU we only support ESP32')

        self.create_image()

    def create_image(self) -> None:
        """
        Create the image, if it doesn't exist.
        """
        if os.path.exists(self.image_path) and self.skip_regenerate_image:
            logging.info(f'Using existing image: {self.image_path}')
        else:
            with DuplicateStdout(self.pexpect_proc):
                image_maker = IdfFlashImageMaker(self, self.image_path)
                image_maker.make_bin()

__init__(pexpect_proc, app_path=None, build_dir=None, part_tool=None, qemu_image_path=None, skip_regenerate_image=False, **kwargs)

Parameters:

Name Type Description Default
pexpect_proc PexpectProcess

pexpect process

required
app_path Optional[str]

App path

None
build_dir Optional[str]

Build directory

None
part_tool Optional[str]

Partition tool path

None
qemu_image_path Optional[str]

QEMU flashable bin path

None
skip_regenerate_image Optional[bool]

skip regenerate QEMU image

False
Source code in pytest_embedded_qemu/app.py
def __init__(
    self,
    pexpect_proc: PexpectProcess,
    app_path: Optional[str] = None,
    build_dir: Optional[str] = None,
    part_tool: Optional[str] = None,
    qemu_image_path: Optional[str] = None,
    skip_regenerate_image: Optional[bool] = False,
    **kwargs,
):
    """
    Args:
        pexpect_proc: pexpect process
        app_path: App path
        build_dir: Build directory
        part_tool: Partition tool path
        qemu_image_path: QEMU flashable bin path
        skip_regenerate_image: skip regenerate QEMU image
    """
    super().__init__(app_path, build_dir=build_dir, part_tool=part_tool, **kwargs)
    self.pexpect_proc = pexpect_proc
    self.image_path = qemu_image_path or os.path.join(self.app_path, self.build_dir, DEFAULT_IMAGE_FN)
    self.skip_regenerate_image = skip_regenerate_image

    if self.target != 'esp32':
        raise ValueError('For now on QEMU we only support ESP32')

    self.create_image()

create_image()

Create the image, if it doesn't exist.

Source code in pytest_embedded_qemu/app.py
def create_image(self) -> None:
    """
    Create the image, if it doesn't exist.
    """
    if os.path.exists(self.image_path) and self.skip_regenerate_image:
        logging.info(f'Using existing image: {self.image_path}')
    else:
        with DuplicateStdout(self.pexpect_proc):
            image_maker = IdfFlashImageMaker(self, self.image_path)
            image_maker.make_bin()

pytest_embedded_qemu.qemu

Qemu

Bases: DuplicateStdoutPopen

QEMU class

Source code in pytest_embedded_qemu/qemu.py
class Qemu(DuplicateStdoutPopen):
    """
    QEMU class
    """

    QEMU_PROG_PATH = 'qemu-system-xtensa'
    QEMU_DEFAULT_ARGS = '-nographic -no-reboot -machine esp32'

    QEMU_STRAP_MODE_FMT = '-global driver=esp32.gpio,property=strap_mode,value={}'
    QEMU_SERIAL_TCP_FMT = '-serial tcp::{},server,nowait'

    def __init__(
        self,
        qemu_image_path: Optional[str] = None,
        qemu_prog_path: Optional[str] = None,
        qemu_cli_args: Optional[str] = None,
        qemu_extra_args: Optional[str] = None,
        **kwargs,
    ):
        """
        Args:
            qemu_image_path: QEMU image path
            qemu_prog_path: QEMU program path
            qemu_cli_args: QEMU CLI arguments
            qemu_extra_args: QEMU CLI extra arguments, will be appended to `qemu_cli_args`
        """
        image_path = qemu_image_path or DEFAULT_IMAGE_FN
        if not os.path.exists(image_path):
            raise ValueError(f'QEMU image path doesn\'t exist: {image_path}')

        qemu_prog_path = qemu_prog_path or self.QEMU_PROG_PATH
        qemu_cli_args = qemu_cli_args or self.QEMU_DEFAULT_ARGS

        qemu_extra_args = qemu_extra_args.replace('"', '') if qemu_extra_args else qemu_extra_args
        qemu_extra_args = [qemu_extra_args] if qemu_extra_args else []
        qemu_extra_args.append(f'-drive file={image_path},if=mtd,format=raw')

        cmd = f'{qemu_prog_path} {qemu_cli_args} {" ".join(qemu_extra_args)}'
        logging.debug(cmd)

        super().__init__(cmd, shell=True, **kwargs)

__init__(qemu_image_path=None, qemu_prog_path=None, qemu_cli_args=None, qemu_extra_args=None, **kwargs)

Parameters:

Name Type Description Default
qemu_image_path Optional[str]

QEMU image path

None
qemu_prog_path Optional[str]

QEMU program path

None
qemu_cli_args Optional[str]

QEMU CLI arguments

None
qemu_extra_args Optional[str]

QEMU CLI extra arguments, will be appended to qemu_cli_args

None
Source code in pytest_embedded_qemu/qemu.py
def __init__(
    self,
    qemu_image_path: Optional[str] = None,
    qemu_prog_path: Optional[str] = None,
    qemu_cli_args: Optional[str] = None,
    qemu_extra_args: Optional[str] = None,
    **kwargs,
):
    """
    Args:
        qemu_image_path: QEMU image path
        qemu_prog_path: QEMU program path
        qemu_cli_args: QEMU CLI arguments
        qemu_extra_args: QEMU CLI extra arguments, will be appended to `qemu_cli_args`
    """
    image_path = qemu_image_path or DEFAULT_IMAGE_FN
    if not os.path.exists(image_path):
        raise ValueError(f'QEMU image path doesn\'t exist: {image_path}')

    qemu_prog_path = qemu_prog_path or self.QEMU_PROG_PATH
    qemu_cli_args = qemu_cli_args or self.QEMU_DEFAULT_ARGS

    qemu_extra_args = qemu_extra_args.replace('"', '') if qemu_extra_args else qemu_extra_args
    qemu_extra_args = [qemu_extra_args] if qemu_extra_args else []
    qemu_extra_args.append(f'-drive file={image_path},if=mtd,format=raw')

    cmd = f'{qemu_prog_path} {qemu_cli_args} {" ".join(qemu_extra_args)}'
    logging.debug(cmd)

    super().__init__(cmd, shell=True, **kwargs)

pytest_embedded_qemu.dut

QemuDut

Bases: Dut

QEMU dut class

Source code in pytest_embedded_qemu/dut.py
class QemuDut(Dut):
    """
    QEMU dut class
    """

    def __init__(
        self,
        qemu: Qemu,
        app: App,
        pexpect_proc: PexpectProcess,
        **kwargs,
    ) -> None:
        """
        Args:
            pexpect_proc: `PexpectProcess` instance
            app: `App` instance
            qemu: `Qemu` instance
        """
        super().__init__(pexpect_proc, app, **kwargs)
        self.qemu = qemu

        self.qemu.create_forward_io_thread(self.pexpect_proc)

    def write(self, s: AnyStr) -> None:
        """
        Write to qemu process.
        """
        self.qemu.send(s)

    def close(self) -> None:
        self.qemu.terminate()

        super(QemuDut, self).close()

__init__(qemu, app, pexpect_proc, **kwargs)

Parameters:

Name Type Description Default
pexpect_proc PexpectProcess

PexpectProcess instance

required
app App

App instance

required
qemu Qemu

Qemu instance

required
Source code in pytest_embedded_qemu/dut.py
def __init__(
    self,
    qemu: Qemu,
    app: App,
    pexpect_proc: PexpectProcess,
    **kwargs,
) -> None:
    """
    Args:
        pexpect_proc: `PexpectProcess` instance
        app: `App` instance
        qemu: `Qemu` instance
    """
    super().__init__(pexpect_proc, app, **kwargs)
    self.qemu = qemu

    self.qemu.create_forward_io_thread(self.pexpect_proc)

write(s)

Write to qemu process.

Source code in pytest_embedded_qemu/dut.py
def write(self, s: AnyStr) -> None:
    """
    Write to qemu process.
    """
    self.qemu.send(s)