Skip to content

Pytest embedded

pytest_embedded.app

App

Built binary files base class

Source code in pytest_embedded/app.py
class App:
    """
    Built binary files base class
    """

    def __init__(
        self,
        app_path: Optional[str] = None,
        build_dir: Optional[str] = None,
        **kwargs,
    ):
        """
        Args:
            app_path: App path
            build_dir: Build directory (where binaries reside)
        """
        if app_path is None:
            app_path = os.getcwd()

        self.app_path = os.path.realpath(app_path)
        self.build_dir = build_dir
        self.binary_path = self._get_binary_path()

        for k, v in kwargs.items():
            setattr(self, k, v)

    def _get_binary_path(self) -> Optional[str]:
        if not self.build_dir:
            return None

        if os.path.isdir(self.build_dir):
            return os.path.realpath(self.build_dir)

        logging.debug(f'{self.build_dir} doesn\'t exist. Treat it as a relative path...')
        path = os.path.join(self.app_path, self.build_dir)
        if os.path.isdir(path):
            return path

        logging.debug(f'{path} doesn\'t exist.')
        return None

__init__(app_path=None, build_dir=None, **kwargs)

Parameters:

Name Type Description Default
app_path Optional[str]

App path

None
build_dir Optional[str]

Build directory (where binaries reside)

None
Source code in pytest_embedded/app.py
def __init__(
    self,
    app_path: Optional[str] = None,
    build_dir: Optional[str] = None,
    **kwargs,
):
    """
    Args:
        app_path: App path
        build_dir: Build directory (where binaries reside)
    """
    if app_path is None:
        app_path = os.getcwd()

    self.app_path = os.path.realpath(app_path)
    self.build_dir = build_dir
    self.binary_path = self._get_binary_path()

    for k, v in kwargs.items():
        setattr(self, k, v)

pytest_embedded.dut

Dut

Device under test (DUT) base class

Source code in pytest_embedded/dut.py
class Dut:
    """
    Device under test (DUT) base class
    """

    def __init__(
        self, pexpect_proc: PexpectProcess, app: App, pexpect_logfile: str, test_case_name: str, **kwargs
    ) -> None:
        """
        Args:
            pexpect_proc: `PexpectProcess` instance
            app: `App` instance
        """
        self.pexpect_proc = pexpect_proc
        self.app = app

        self.logfile = pexpect_logfile
        self.logdir = os.path.dirname(self.logfile)
        logging.info(f'Logs recorded under folder: {self.logdir}')

        self.test_case_name = test_case_name
        self.dut_name = os.path.splitext(os.path.basename(pexpect_logfile))[0]

        for k, v in kwargs.items():
            setattr(self, k, v)

        # junit related
        # TODO: if request.option.xmlpath
        self.testsuite = TestSuite(self.test_case_name)

    def close(self) -> None:
        if self.testsuite.testcases:
            junit_report = os.path.join(self.logdir, f'{self.dut_name}.xml')
            self.testsuite.dump(junit_report)
            logging.info(f'Created unity output junit report: {junit_report}')

    def write(self, *args, **kwargs) -> None:
        """
        Write to `pexpect_proc`. All arguments would pass to `pexpect.spawn.write()`
        """
        self.pexpect_proc.write(*args, **kwargs)

    def _pexpect_func(func) -> Callable[..., Union[Match, AnyStr]]:  # noqa
        @functools.wraps(func)  # noqa
        def wrapper(
            self, pattern, *args, expect_all: bool = False, **kwargs
        ) -> Union[Union[Match, AnyStr], List[Union[Match, AnyStr]]]:
            patterns = to_list(pattern)
            res = []
            while patterns:
                try:
                    index = func(self, pattern, *args, **kwargs)  # noqa
                except (pexpect.EOF, pexpect.TIMEOUT) as e:
                    wrapped_buffer_bytes = textwrap.shorten(
                        remove_asci_color_code(to_str(self.pexpect_proc.buffer)),
                        width=200,
                        placeholder=f'... (total {len(self.pexpect_proc.buffer)} bytes)',
                    )
                    debug_str = (
                        f'Not found "{str(pattern)}"\n'
                        f'Bytes in current buffer (color code eliminated): {wrapped_buffer_bytes}\n'
                        f'Please check the full log here: {self.logfile}'
                    )
                    raise e.__class__(debug_str) from e
                else:
                    if self.pexpect_proc.match in [pexpect.EOF, pexpect.TIMEOUT]:
                        res.append(self.pexpect_proc.before.rstrip())
                    else:
                        res.append(self.pexpect_proc.match)

                if expect_all:
                    patterns.pop(index)
                else:
                    break  # one succeeded. leave the loop

            if len(res) == 1:
                return res[0]

            return res

        return wrapper

    @_pexpect_func  # noqa
    def expect(self, pattern, **kwargs) -> Match:  # noqa
        """
        Expect from `pexpect_proc`. All the arguments would pass to `pexpect.expect()`.

        Returns:
            AnyStr: if you're matching pexpect.EOF or pexpect.TIMEOUT to get all the current buffers.

        Returns:
            re.Match: if matched given string.
        """
        return self.pexpect_proc.expect(pattern, **kwargs)

    @_pexpect_func  # noqa
    def expect_exact(self, pattern, **kwargs) -> Match:  # noqa
        """
        Expect from `pexpect_proc`. All the arguments would pass to `pexpect.expect_exact()`.

        Returns:
            AnyStr: if you're matching pexpect.EOF or pexpect.TIMEOUT to get all the current buffers.

        Returns:
            re.Match: if matched given string.
        """
        return self.pexpect_proc.expect_exact(pattern, **kwargs)

    def expect_unity_test_output(
        self, remove_asci_escape_code: bool = True, timeout: int = 60, extra_before: Optional[AnyStr] = None
    ) -> None:
        """
        Expect a unity test summary block and parse the output into junit report.

        Would combine the junit report into the main one if you use `pytest --junitxml` feature.

        Args:
            remove_asci_escape_code: remove asci escape code in the message field. (default: True)
            timeout: timeout. (default: 60 seconds)
            extra_before: would append before the expected bytes.
                Use this argument when need to run `expect` functions between one unity test call.

        Notes:
            Would raise AssertionError at the end of the test if any unity test case result is "FAIL"
        """
        self.expect(UNITY_SUMMARY_LINE_REGEX, timeout=timeout)

        if extra_before:
            log = to_bytes(extra_before) + self.pexpect_proc.before
        else:
            log = self.pexpect_proc.before

        if remove_asci_escape_code:
            log = remove_asci_color_code(log)

        self.testsuite.add_unity_test_cases(log)

__init__(pexpect_proc, app, pexpect_logfile, test_case_name, **kwargs)

Parameters:

Name Type Description Default
pexpect_proc PexpectProcess

PexpectProcess instance

required
app App

App instance

required
Source code in pytest_embedded/dut.py
def __init__(
    self, pexpect_proc: PexpectProcess, app: App, pexpect_logfile: str, test_case_name: str, **kwargs
) -> None:
    """
    Args:
        pexpect_proc: `PexpectProcess` instance
        app: `App` instance
    """
    self.pexpect_proc = pexpect_proc
    self.app = app

    self.logfile = pexpect_logfile
    self.logdir = os.path.dirname(self.logfile)
    logging.info(f'Logs recorded under folder: {self.logdir}')

    self.test_case_name = test_case_name
    self.dut_name = os.path.splitext(os.path.basename(pexpect_logfile))[0]

    for k, v in kwargs.items():
        setattr(self, k, v)

    # junit related
    # TODO: if request.option.xmlpath
    self.testsuite = TestSuite(self.test_case_name)

write(*args, **kwargs)

Write to pexpect_proc. All arguments would pass to pexpect.spawn.write()

Source code in pytest_embedded/dut.py
def write(self, *args, **kwargs) -> None:
    """
    Write to `pexpect_proc`. All arguments would pass to `pexpect.spawn.write()`
    """
    self.pexpect_proc.write(*args, **kwargs)

expect(pattern, **kwargs)

Expect from pexpect_proc. All the arguments would pass to pexpect.expect().

Returns:

Name Type Description
AnyStr Match

if you're matching pexpect.EOF or pexpect.TIMEOUT to get all the current buffers.

Returns:

Type Description
Match

re.Match: if matched given string.

Source code in pytest_embedded/dut.py
@_pexpect_func  # noqa
def expect(self, pattern, **kwargs) -> Match:  # noqa
    """
    Expect from `pexpect_proc`. All the arguments would pass to `pexpect.expect()`.

    Returns:
        AnyStr: if you're matching pexpect.EOF or pexpect.TIMEOUT to get all the current buffers.

    Returns:
        re.Match: if matched given string.
    """
    return self.pexpect_proc.expect(pattern, **kwargs)

expect_exact(pattern, **kwargs)

Expect from pexpect_proc. All the arguments would pass to pexpect.expect_exact().

Returns:

Name Type Description
AnyStr Match

if you're matching pexpect.EOF or pexpect.TIMEOUT to get all the current buffers.

Returns:

Type Description
Match

re.Match: if matched given string.

Source code in pytest_embedded/dut.py
@_pexpect_func  # noqa
def expect_exact(self, pattern, **kwargs) -> Match:  # noqa
    """
    Expect from `pexpect_proc`. All the arguments would pass to `pexpect.expect_exact()`.

    Returns:
        AnyStr: if you're matching pexpect.EOF or pexpect.TIMEOUT to get all the current buffers.

    Returns:
        re.Match: if matched given string.
    """
    return self.pexpect_proc.expect_exact(pattern, **kwargs)

expect_unity_test_output(remove_asci_escape_code=True, timeout=60, extra_before=None)

Expect a unity test summary block and parse the output into junit report.

Would combine the junit report into the main one if you use pytest --junitxml feature.

Parameters:

Name Type Description Default
remove_asci_escape_code bool

remove asci escape code in the message field. (default: True)

True
timeout int

timeout. (default: 60 seconds)

60
extra_before Optional[AnyStr]

would append before the expected bytes. Use this argument when need to run expect functions between one unity test call.

None
Notes

Would raise AssertionError at the end of the test if any unity test case result is "FAIL"

Source code in pytest_embedded/dut.py
def expect_unity_test_output(
    self, remove_asci_escape_code: bool = True, timeout: int = 60, extra_before: Optional[AnyStr] = None
) -> None:
    """
    Expect a unity test summary block and parse the output into junit report.

    Would combine the junit report into the main one if you use `pytest --junitxml` feature.

    Args:
        remove_asci_escape_code: remove asci escape code in the message field. (default: True)
        timeout: timeout. (default: 60 seconds)
        extra_before: would append before the expected bytes.
            Use this argument when need to run `expect` functions between one unity test call.

    Notes:
        Would raise AssertionError at the end of the test if any unity test case result is "FAIL"
    """
    self.expect(UNITY_SUMMARY_LINE_REGEX, timeout=timeout)

    if extra_before:
        log = to_bytes(extra_before) + self.pexpect_proc.before
    else:
        log = self.pexpect_proc.before

    if remove_asci_escape_code:
        log = remove_asci_color_code(log)

    self.testsuite.add_unity_test_cases(log)

pytest_embedded.log

PexpectProcess

Bases: pexpect.fdpexpect.fdspawn

Use a temp file to gather multiple inputs into one output, and do pexpect.expect() from one place.

Source code in pytest_embedded/log.py
class PexpectProcess(pexpect.fdpexpect.fdspawn):
    """
    Use a temp file to gather multiple inputs into one output, and do `pexpect.expect()` from one place.
    """

    STDOUT = sys.stdout

    def __init__(
        self,
        pexpect_fr: BinaryIO,
        pexpect_fw: BinaryIO,
        with_timestamp: bool = True,
        count: int = 1,
        total: int = 1,
        **kwargs,
    ):
        self._count = count
        self._total = total

        if self._total > 1:
            self.source = f'dut-{self._count}'
        else:
            self.source = None

        super().__init__(pexpect_fr, **kwargs)

        self._fr = pexpect_fr
        self._fw = pexpect_fw
        self._with_timestamp = with_timestamp
        self._write_lock = threading.Lock()

        self._added_prefix = False

    def send(self, s: AnyStr) -> int:
        """
        Write to the pexpect process and log.

        Args:
            s: bytes or str

        Returns:
            number of written bytes.
        """
        if not s:
            return 0

        s = self._coerce_send_string(s)
        self._log(s, 'send')

        # for pytest logging
        _temp = sys.stdout
        sys.stdout = self.STDOUT  # ensure the following print uses system sys.stdout

        _s = to_str(s)
        prefix = ''
        if self.source:
            prefix = f'[{self.source}] ' + prefix
        if self._with_timestamp:
            prefix = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + ' ' + prefix

        if not self._added_prefix:
            _s = prefix + _s
            self._added_prefix = True
        _s = _s.replace('\n', '\n' + prefix)
        if _s.endswith(prefix):
            _s = _s.rsplit(prefix, maxsplit=1)[0]
            self._added_prefix = False

        sys.stdout.write(_s)
        sys.stdout.flush()
        sys.stdout = _temp

        # write the bytes into the pexpect process
        b = self._encoder.encode(s, final=False)
        try:
            written = self._fw.write(b)
            self._fw.flush()
        except ValueError:  # write to closed file. since this function would be run in daemon thread, would happen
            return 0

        return written

    def write(self, s: AnyStr) -> None:
        with self._write_lock:
            self.send(s)

    def read_nonblocking(self, size=1, timeout=-1) -> bytes:
        """
        Since we're using real file stream, here we only raise an EOF error only when the file stream has been closed.
        This could solve the `os.read()` blocked issue.

        Args:
            size: most read bytes
            timeout: timeout

        Returns:
            read bytes
        """
        try:
            if os.name == 'posix':
                if timeout == -1:
                    timeout = self.timeout
                rlist = [self.child_fd]
                wlist = []
                xlist = []
                if self.use_poll:
                    rlist = poll_ignore_interrupts(rlist, timeout)
                else:
                    rlist, wlist, xlist = select_ignore_interrupts(rlist, wlist, xlist, timeout)
                if self.child_fd not in rlist:
                    raise TIMEOUT('Timeout exceeded.')

            s = os.read(self.child_fd, size)
        except OSError as err:
            if err.args[0] == errno.EIO:  # Linux-style EOF
                pass
            if err.args[0] == errno.EBADF:  # Bad file descriptor
                raise EOF('Bad File Descriptor')
            raise

        s = self._decoder.decode(s, final=False)
        self._log(s, 'read')
        return s

    def terminate(self, force=False):
        """
        Close the temporary file streams
        """
        try:
            self._fr.close()
            self._fw.close()
        except:  # noqa
            pass

send(s)

Write to the pexpect process and log.

Parameters:

Name Type Description Default
s AnyStr

bytes or str

required

Returns:

Type Description
int

number of written bytes.

Source code in pytest_embedded/log.py
def send(self, s: AnyStr) -> int:
    """
    Write to the pexpect process and log.

    Args:
        s: bytes or str

    Returns:
        number of written bytes.
    """
    if not s:
        return 0

    s = self._coerce_send_string(s)
    self._log(s, 'send')

    # for pytest logging
    _temp = sys.stdout
    sys.stdout = self.STDOUT  # ensure the following print uses system sys.stdout

    _s = to_str(s)
    prefix = ''
    if self.source:
        prefix = f'[{self.source}] ' + prefix
    if self._with_timestamp:
        prefix = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + ' ' + prefix

    if not self._added_prefix:
        _s = prefix + _s
        self._added_prefix = True
    _s = _s.replace('\n', '\n' + prefix)
    if _s.endswith(prefix):
        _s = _s.rsplit(prefix, maxsplit=1)[0]
        self._added_prefix = False

    sys.stdout.write(_s)
    sys.stdout.flush()
    sys.stdout = _temp

    # write the bytes into the pexpect process
    b = self._encoder.encode(s, final=False)
    try:
        written = self._fw.write(b)
        self._fw.flush()
    except ValueError:  # write to closed file. since this function would be run in daemon thread, would happen
        return 0

    return written

read_nonblocking(size=1, timeout=-1)

Since we're using real file stream, here we only raise an EOF error only when the file stream has been closed. This could solve the os.read() blocked issue.

Parameters:

Name Type Description Default
size

most read bytes

1
timeout

timeout

-1

Returns:

Type Description
bytes

read bytes

Source code in pytest_embedded/log.py
def read_nonblocking(self, size=1, timeout=-1) -> bytes:
    """
    Since we're using real file stream, here we only raise an EOF error only when the file stream has been closed.
    This could solve the `os.read()` blocked issue.

    Args:
        size: most read bytes
        timeout: timeout

    Returns:
        read bytes
    """
    try:
        if os.name == 'posix':
            if timeout == -1:
                timeout = self.timeout
            rlist = [self.child_fd]
            wlist = []
            xlist = []
            if self.use_poll:
                rlist = poll_ignore_interrupts(rlist, timeout)
            else:
                rlist, wlist, xlist = select_ignore_interrupts(rlist, wlist, xlist, timeout)
            if self.child_fd not in rlist:
                raise TIMEOUT('Timeout exceeded.')

        s = os.read(self.child_fd, size)
    except OSError as err:
        if err.args[0] == errno.EIO:  # Linux-style EOF
            pass
        if err.args[0] == errno.EBADF:  # Bad file descriptor
            raise EOF('Bad File Descriptor')
        raise

    s = self._decoder.decode(s, final=False)
    self._log(s, 'read')
    return s

terminate(force=False)

Close the temporary file streams

Source code in pytest_embedded/log.py
def terminate(self, force=False):
    """
    Close the temporary file streams
    """
    try:
        self._fr.close()
        self._fw.close()
    except:  # noqa
        pass

DuplicateStdout

Bases: TextIOWrapper

A context manager to duplicate sys.stdout to pexpect_proc.

Warning
  • Within this context manager, the print() would be redirected to self.write(). All the args and kwargs passed to print() would be ignored and might not work as expected.
  • The context manager replacement of sys.stdout is NOT thread-safe. DO NOT use it in a thread.
Source code in pytest_embedded/log.py
class DuplicateStdout(TextIOWrapper):
    """
    A context manager to duplicate `sys.stdout` to `pexpect_proc`.

    Warning:
        - Within this context manager, the `print()` would be redirected to `self.write()`.
        All the `args` and `kwargs` passed to `print()` would be ignored and might not work as expected.
        - The context manager replacement of `sys.stdout` is NOT thread-safe. DO NOT use it in a thread.
    """

    STDOUT = sys.stdout

    def __init__(self, pexpect_proc: PexpectProcess):  # noqa
        """
        Args:
            pexpect_proc: `PexpectProcess` instance
        """
        # DO NOT call super().__init__(), use TextIOWrapper as parent class only for types and functions
        self.pexpect_proc = pexpect_proc
        self.before = None

    def __enter__(self):
        if sys.stdout != self.STDOUT:
            self.before = sys.stdout
        sys.stdout = self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def __del__(self):
        self.close()

    def write(self, data: bytes) -> None:
        """
        Call `pexpect_proc.write()` instead of `sys.stdout.write()`
        """
        if not data:
            return

        self.pexpect_proc.write(data)

    def flush(self) -> None:
        """
        Don't need to flush anymore since the `flush` method would be called inside `pexpect_proc`.
        """
        pass

    def close(self) -> None:
        """
        Stop redirecting `sys.stdout`.
        """
        if self.before:
            sys.stdout = self.before
        else:
            sys.stdout = self.STDOUT

    def isatty(self) -> bool:
        """
        Returns:
            True since it has `write()`.
        """
        return True

__init__(pexpect_proc)

Parameters:

Name Type Description Default
pexpect_proc PexpectProcess

PexpectProcess instance

required
Source code in pytest_embedded/log.py
def __init__(self, pexpect_proc: PexpectProcess):  # noqa
    """
    Args:
        pexpect_proc: `PexpectProcess` instance
    """
    # DO NOT call super().__init__(), use TextIOWrapper as parent class only for types and functions
    self.pexpect_proc = pexpect_proc
    self.before = None

write(data)

Call pexpect_proc.write() instead of sys.stdout.write()

Source code in pytest_embedded/log.py
def write(self, data: bytes) -> None:
    """
    Call `pexpect_proc.write()` instead of `sys.stdout.write()`
    """
    if not data:
        return

    self.pexpect_proc.write(data)

flush()

Don't need to flush anymore since the flush method would be called inside pexpect_proc.

Source code in pytest_embedded/log.py
def flush(self) -> None:
    """
    Don't need to flush anymore since the `flush` method would be called inside `pexpect_proc`.
    """
    pass

close()

Stop redirecting sys.stdout.

Source code in pytest_embedded/log.py
def close(self) -> None:
    """
    Stop redirecting `sys.stdout`.
    """
    if self.before:
        sys.stdout = self.before
    else:
        sys.stdout = self.STDOUT

isatty()

Returns:

Type Description
bool

True since it has write().

Source code in pytest_embedded/log.py
def isatty(self) -> bool:
    """
    Returns:
        True since it has `write()`.
    """
    return True

DuplicateStdoutMixin

A mixin class which provides function create_forward_io_thread to create a forward io thread.

Notes

_forward_io() should be implemented in subclasses, the function should be something like:

def _forward_io(self, pexpect_proc: PexpectProcess) -> None:
    pexpect_proc.write(...)
Source code in pytest_embedded/log.py
class DuplicateStdoutMixin:
    """
    A mixin class which provides function `create_forward_io_thread` to create a forward io thread.

    Notes:
        `_forward_io()` should be implemented in subclasses, the function should be something like:

        ```python
        def _forward_io(self, pexpect_proc: PexpectProcess) -> None:
            pexpect_proc.write(...)
        ```
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self._forward_io_thread: threading.Thread = None  # type: ignore

    def create_forward_io_thread(self, pexpect_proc: PexpectProcess) -> None:
        """
        Create a forward io daemon thread if it doesn't exist.

        Args:
            pexpect_proc: `PexpectProcess` instance
        """
        if self._forward_io_thread and self._forward_io_thread.is_alive():
            return

        self._forward_io_thread = threading.Thread(target=self._forward_io, args=(pexpect_proc,), daemon=True)
        self._forward_io_thread.start()

    def _forward_io(self, pexpect_proc: PexpectProcess) -> None:
        raise NotImplementedError('should be implemented by subclasses')

create_forward_io_thread(pexpect_proc)

Create a forward io daemon thread if it doesn't exist.

Parameters:

Name Type Description Default
pexpect_proc PexpectProcess

PexpectProcess instance

required
Source code in pytest_embedded/log.py
def create_forward_io_thread(self, pexpect_proc: PexpectProcess) -> None:
    """
    Create a forward io daemon thread if it doesn't exist.

    Args:
        pexpect_proc: `PexpectProcess` instance
    """
    if self._forward_io_thread and self._forward_io_thread.is_alive():
        return

    self._forward_io_thread = threading.Thread(target=self._forward_io, args=(pexpect_proc,), daemon=True)
    self._forward_io_thread.start()

DuplicateStdoutPopen

Bases: DuplicateStdoutMixin, subprocess.Popen

subprocess.Popen with DuplicateStdoutMixin mixed with default popen kwargs.

Source code in pytest_embedded/log.py
class DuplicateStdoutPopen(DuplicateStdoutMixin, subprocess.Popen):
    """
    `subprocess.Popen` with `DuplicateStdoutMixin` mixed with default popen kwargs.
    """

    def __init__(self, cmd: Union[str, List[str]], **kwargs):
        # we use real log file to record output, pipe-like file object won't be non-blocking.
        _log_file = os.path.join(
            tempfile.gettempdir(),
            datetime.datetime.utcnow().strftime('%Y-%m-%d_%H-%M-%S'),
            f'{uuid.uuid4()}.log',
        )
        parent_dir = os.path.dirname(_log_file)
        if parent_dir:  # in case value is a single file under the current dir
            os.makedirs(os.path.dirname(_log_file), exist_ok=True)
        self._fw = open(_log_file, 'w')
        self._fr = open(_log_file, 'r')
        logging.debug(f'temp log file: {_log_file}')

        kwargs.update(
            {
                'bufsize': 0,
                'stdin': subprocess.PIPE,
                'stdout': self._fw,
                'stderr': self._fw,
            }
        )

        super().__init__(cmd, **kwargs)

    def __del__(self):
        self._fw.close()
        self._fr.close()

    def send(self, s: AnyStr) -> None:
        """
        Write to `stdin` via `stdin.write`.

        If the input is `str`, will encode to `bytes` and add a b'\\n' automatically in the end.

        if the input is `bytes`, will pass this directly.

        Args:
            s: bytes or str
        """
        self.stdin.write(to_bytes(s, '\n'))

    def _forward_io(self, pexpect_proc: PexpectProcess) -> None:
        while self.poll() is None:
            pexpect_proc.write(self._fr.read())
            sleep(0.1)  # set interval

send(s)

Write to stdin via stdin.write.

If the input is str, will encode to bytes and add a b'\n' automatically in the end.

if the input is bytes, will pass this directly.

Parameters:

Name Type Description Default
s AnyStr

bytes or str

required
Source code in pytest_embedded/log.py
def send(self, s: AnyStr) -> None:
    """
    Write to `stdin` via `stdin.write`.

    If the input is `str`, will encode to `bytes` and add a b'\\n' automatically in the end.

    if the input is `bytes`, will pass this directly.

    Args:
        s: bytes or str
    """
    self.stdin.write(to_bytes(s, '\n'))

live_print_call(*args, **kwargs)

live print the subprocess.Popen process.

Notes

This function behaves the same as subprocess.call(), it would block your current process.

Source code in pytest_embedded/log.py
def live_print_call(*args, **kwargs):
    """
    live print the `subprocess.Popen` process.

    Notes:
        This function behaves the same as `subprocess.call()`, it would block your current process.
    """
    default_kwargs = {
        'stdout': subprocess.PIPE,
        'stderr': subprocess.STDOUT,
    }
    default_kwargs.update(kwargs)

    process = subprocess.Popen(*args, **default_kwargs)
    while process.poll() is None:
        print(to_str(process.stdout.read()))

pytest_embedded.utils

to_str(bytes_str)

Turn bytes or str to str

Parameters:

Name Type Description Default
bytes_str AnyStr

bytes or str

required

Returns:

Type Description
str

utf8-decoded string

Source code in pytest_embedded/utils.py
def to_str(bytes_str: AnyStr) -> str:
    """
    Turn `bytes` or `str` to `str`

    Args:
        bytes_str: `bytes` or `str`

    Returns:
        utf8-decoded string
    """
    if isinstance(bytes_str, bytes):
        return bytes_str.decode('utf-8', errors='ignore')
    return bytes_str

to_bytes(bytes_str, ending=None)

Turn bytes or str to bytes

Parameters:

Name Type Description Default
bytes_str AnyStr

bytes or str

required
ending Optional[AnyStr]

bytes or str, will add to the end of the result. Only works when the bytes_str is str

None

Returns:

Type Description
bytes

utf8-encoded bytes

Source code in pytest_embedded/utils.py
def to_bytes(bytes_str: AnyStr, ending: Optional[AnyStr] = None) -> bytes:
    """
    Turn `bytes` or `str` to `bytes`

    Args:
        bytes_str: `bytes` or `str`
        ending: `bytes` or `str`, will add to the end of the result.
            Only works when the `bytes_str` is `str`

    Returns:
        utf8-encoded bytes
    """
    if isinstance(bytes_str, str):
        bytes_str = bytes_str.encode()

        if ending:
            if isinstance(ending, str):
                ending = ending.encode()
            return bytes_str + ending

    return bytes_str

to_list(s)

Parameters:

Name Type Description Default
s _T

Anything

required

Returns:

Type Description
List[_T]

list(s). If s is a tuple or a set.

List[_T]

itself. If s is a list

List[_T]

[s]. If s is other types.

Source code in pytest_embedded/utils.py
def to_list(s: _T) -> List[_T]:
    """
    Args:
        s: Anything

    Returns:
        `list(s)`. If `s` is a tuple or a set.

        itself. If `s` is a list

        `[s]`. If `s` is other types.
    """
    if not s:
        return s

    if isinstance(s, set) or isinstance(s, tuple):
        return list(s)
    elif isinstance(s, list):
        return s
    else:
        return [s]

pytest_embedded.plugin

count(request)

Enable parametrization for the same cli option. Inject to global variable COUNT.

Source code in pytest_embedded/plugin.py
@pytest.fixture(autouse=True)
def count(request):
    """
    Enable parametrization for the same cli option. Inject to global variable `COUNT`.
    """
    global _COUNT
    _COUNT = _gte_one_int(getattr(request, 'param', request.config.option.count))

parse_multi_dut_args(count, s)

Parse multi-dut argument by the following rules:

  • When the return value is a string, split the string by |.
  • If the configuration value only has one item, duplicate it by the "count" amount.
  • If the configuration value item amount is the same as the "count" amount, return it directly.

Parameters:

Name Type Description Default
count int

Multi-Dut count

required
s str

argument string

required

Returns:

Type Description
Union[Any, Tuple[Any]]

The argument itself. if count is 1.

Union[Any, Tuple[Any]]

The tuple of the parsed argument. if count is greater than 1.

Raises:

Type Description
ValueError

when a configuration has multi values but the amount is different from the count amount.

Source code in pytest_embedded/plugin.py
def parse_multi_dut_args(count: int, s: str) -> Union[Any, Tuple[Any]]:
    """
    Parse multi-dut argument by the following rules:

    - When the return value is a string, split the string by `|`.
    - If the configuration value only has one item, duplicate it by the "count" amount.
    - If the configuration value item amount is the same as the "count" amount, return it directly.

    Args:
        count: Multi-Dut count
        s: argument string

    Returns:
        The argument itself. if `count` is 1.
        The tuple of the parsed argument. if `count` is greater than 1.

    Raises:
        ValueError: when a configuration has multi values but the amount is different from the `count` amount.
    """
    if isinstance(s, str):
        res = s.split('|')
    else:
        res = [s]

    if len(res) == 1:
        if count == 1:
            return _str_bool(res[0])
        else:
            return tuple([_str_bool(res[0])] * count)
    else:  # len(res) > 1
        if len(res) != count:
            raise ValueError('The configuration has multi values but the amount is different from the "count" amount.')
        else:
            return tuple(_str_bool(item) for item in res)

multi_dut_argument(func)

Used for parse the multi-dut argument according to the count amount.

Source code in pytest_embedded/plugin.py
def multi_dut_argument(func) -> Callable[..., Union[Optional[str], Tuple[Optional[str]]]]:
    """
    Used for parse the multi-dut argument according to the `count` amount.
    """

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return parse_multi_dut_args(_COUNT, func(*args, **kwargs))

    return wrapper

multi_dut_fixture(func)

Apply the multi-dut arguments to each fixture.

Notes

Run the func(*args, **kwargs) for multiple times by iterating all kwargs via itemgetter

For example:

  • input: {key1: (v1, v2), key2: (v1, v2)}
  • output: (func(**{key1: v1, key2: v1}), func(**{key1: v2, key2: v2}))

Returns:

Type Description
Callable[..., Union[Any, Tuple[Any]]]

The return value, if count is 1.

Callable[..., Union[Any, Tuple[Any]]]

The tuple of return values, if count is greater than 1.

Source code in pytest_embedded/plugin.py
def multi_dut_fixture(func) -> Callable[..., Union[Any, Tuple[Any]]]:
    """
    Apply the multi-dut arguments to each fixture.

    Notes:
        Run the `func(*args, **kwargs)` for multiple times by iterating all `kwargs` via `itemgetter`

        For example:

        - input: `{key1: (v1, v2), key2: (v1, v2)}`
        - output: `(func(**{key1: v1, key2: v1}), func(**{key1: v2, key2: v2}))`

    Returns:
        The return value, if `count` is 1.
        The tuple of return values, if `count` is greater than 1.
    """

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        if _COUNT == 1:
            return func(*args, **kwargs)

        res = tuple()
        for i in range(_COUNT):
            getter = itemgetter(i)
            current_kwargs = {}
            for k, v in kwargs.items():
                if isinstance(v, list) or isinstance(v, tuple):
                    current_kwargs[k] = getter(v)
                else:
                    current_kwargs[k] = v
            res = tuple(list(res) + [func(*args, **current_kwargs)])

        return res

    return wrapper

multi_dut_generator_fixture(func)

Apply the multi-dut arguments to each fixture.

Notes

Run the func() for multiple times by iterating all kwargs via itemgetter. Auto call close() or terminate() method of the object after it yield back.

Yields:

Type Description
Callable[..., Generator[Union[Any, Tuple[Any]], Any, None]]

The return value, if count is 1.

Callable[..., Generator[Union[Any, Tuple[Any]], Any, None]]

The tuple of return values, if count is greater than 1.

Source code in pytest_embedded/plugin.py
def multi_dut_generator_fixture(func) -> Callable[..., Generator[Union[Any, Tuple[Any]], Any, None]]:
    """
    Apply the multi-dut arguments to each fixture.

    Notes:
        Run the `func()` for multiple times by iterating all `kwargs` via `itemgetter`. Auto call `close()` or
        `terminate()` method of the object after it yield back.

    Yields:
        The return value, if `count` is 1.
        The tuple of return values, if `count` is greater than 1.
    """

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        def _close_or_terminate(obj):
            try:
                obj.close()
            except OSError as e:
                logging.debug(str(e))
                pass
            except AttributeError:
                try:
                    obj.terminate()
                except AttributeError as e:
                    logging.debug(str(e))
                    pass
            except Exception as e:
                logging.debug(str(e))
                pass  # swallow up all error
            finally:
                del obj

        if _COUNT == 1:
            res = None
            try:
                res = func(*args, **kwargs)
                yield res
            finally:
                if res:
                    _close_or_terminate(res)
        else:
            res = []
            for i in range(_COUNT):
                getter = itemgetter(i)
                current_kwargs = {}
                for k, v in kwargs.items():
                    if isinstance(v, list) or isinstance(v, tuple):
                        current_kwargs[k] = getter(v)
                    else:
                        current_kwargs[k] = v
                if func.__name__ in ['_pexpect_logfile', 'pexpect_proc']:
                    current_kwargs['count'] = i
                    current_kwargs['total'] = _COUNT
                res.append(func(*args, **current_kwargs))
            try:
                yield res
            finally:
                if res:
                    for item in res:
                        _close_or_terminate(item)

    return wrapper

session_tempdir()

Session scoped temp dir for pytest-embedded

Source code in pytest_embedded/plugin.py
@pytest.fixture(scope='session', autouse=True)
def session_tempdir() -> str:
    """Session scoped temp dir for pytest-embedded"""
    _tmpdir = os.path.join(
        tempfile.gettempdir(),
        'pytest-embedded',
        datetime.datetime.utcnow().strftime('%Y-%m-%d_%H-%M-%S'),
    )
    os.makedirs(_tmpdir, exist_ok=True)
    return _tmpdir

test_file_path(request)

Current test script file path

Source code in pytest_embedded/plugin.py
@pytest.fixture
def test_file_path(request: FixtureRequest) -> str:
    """Current test script file path"""
    return request.module.__file__

test_case_name(request)

Current test case function name

Source code in pytest_embedded/plugin.py
@pytest.fixture
def test_case_name(request: FixtureRequest) -> str:
    """Current test case function name"""
    return request.node.name

test_case_tempdir(test_case_name, session_tempdir)

Function scoped temp dir for pytest-embedded

Source code in pytest_embedded/plugin.py
@pytest.fixture
def test_case_tempdir(test_case_name: str, session_tempdir: str) -> str:
    """Function scoped temp dir for pytest-embedded"""
    return os.path.join(session_tempdir, test_case_name)

with_timestamp(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture()
@multi_dut_argument
def with_timestamp(request: FixtureRequest) -> bool:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'with_timestamp', None)

pexpect_proc(_pexpect_fr, _pexpect_fw, with_timestamp, **kwargs)

Pexpect process that run the expect functions on

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_generator_fixture
def pexpect_proc(
    _pexpect_fr, _pexpect_fw, with_timestamp, **kwargs  # kwargs passed by `multi_dut_generator_fixture()`
) -> PexpectProcess:
    """Pexpect process that run the expect functions on"""
    kwargs.update({'pexpect_fr': _pexpect_fr, 'pexpect_fw': _pexpect_fw, 'with_timestamp': with_timestamp})
    return PexpectProcess(**_drop_none_kwargs(kwargs))

redirect(pexpect_proc)

A context manager that could help duplicate all the sys.stdout to dut.pexpect_proc.

with redirect():
    print('this should be logged and sent to pexpect_proc')
Warning

This is NOT thread-safe, DO NOT use this in a thread. If you want to redirect the stdout of a thread to the pexpect process and log it, please use pexpect_proc.write() instead.

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_generator_fixture
def redirect(pexpect_proc: PexpectProcess) -> Callable[..., DuplicateStdout]:
    """
    A context manager that could help duplicate all the `sys.stdout` to `dut.pexpect_proc`.

    ```python
    with redirect():
        print('this should be logged and sent to pexpect_proc')
    ```

    Warning:
        This is NOT thread-safe, DO NOT use this in a thread. If you want to redirect the stdout of a thread to the
        pexpect process and log it, please use `pexpect_proc.write()` instead.
    """

    def _inner():
        return DuplicateStdout(pexpect_proc)

    return _inner

embedded_services(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def embedded_services(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'embedded_services', None)

app_path(request, test_file_path)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def app_path(request: FixtureRequest, test_file_path: str) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'app_path', os.path.dirname(test_file_path))

build_dir(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def build_dir(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'build_dir', 'build')

port(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def port(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'port', None)

baud(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def baud(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'baud', None)

target(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def target(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'target', None)

skip_autoflash(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def skip_autoflash(request: FixtureRequest) -> Optional[bool]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'skip_autoflash', None)

erase_all(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def erase_all(request: FixtureRequest) -> Optional[bool]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'erase_all', None)

esptool_baud(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def esptool_baud(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'esptool_baud', None)

part_tool(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def part_tool(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'part_tool', None)

confirm_target_elf_sha256(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def confirm_target_elf_sha256(request: FixtureRequest) -> Optional[bool]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'confirm_target_elf_sha256', None)

erase_nvs(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def erase_nvs(request: FixtureRequest) -> Optional[bool]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'erase_nvs', None)

skip_check_coredump(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def skip_check_coredump(request: FixtureRequest) -> Optional[bool]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'skip_check_coredump', None)

gdb_prog_path(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def gdb_prog_path(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'gdb_prog_path', None)

gdb_cli_args(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def gdb_cli_args(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'gdb_cli_args', None)

openocd_prog_path(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def openocd_prog_path(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'openocd_prog_path', None)

openocd_cli_args(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def openocd_cli_args(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'openocd_cli_args', None)

qemu_image_path(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def qemu_image_path(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'qemu_image_path', None)

qemu_prog_path(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def qemu_prog_path(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'qemu_prog_path', None)

qemu_cli_args(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def qemu_cli_args(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'qemu_cli_args', None)

qemu_extra_args(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def qemu_extra_args(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'qemu_extra_args', None)

skip_regenerate_image(request)

Enable parametrization for the same cli option

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_argument
def skip_regenerate_image(request: FixtureRequest) -> Optional[str]:
    """Enable parametrization for the same cli option"""
    return _request_param_or_config_option_or_default(request, 'skip_regenerate_image', None)

app(_fixture_classes_and_options)

A pytest fixture to gather information from the specified built binary folder

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_fixture
def app(_fixture_classes_and_options: ClassCliOptions) -> App:
    """A pytest fixture to gather information from the specified built binary folder"""
    cls = _fixture_classes_and_options.classes['app']
    kwargs = _fixture_classes_and_options.kwargs['app']
    return cls(**_drop_none_kwargs(kwargs))

serial(_fixture_classes_and_options, app)

A serial subprocess that could read/redirect/write

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_generator_fixture
def serial(_fixture_classes_and_options, app) -> Optional['Serial']:
    """A serial subprocess that could read/redirect/write"""
    if 'serial' not in _fixture_classes_and_options.classes:
        return None

    cls = _fixture_classes_and_options.classes['serial']
    kwargs = _fixture_classes_and_options.kwargs['serial']
    if 'app' in kwargs and kwargs['app'] is None:
        kwargs['app'] = app
    return cls(**_drop_none_kwargs(kwargs))

openocd(_fixture_classes_and_options)

An openocd subprocess that could read/redirect/write

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_generator_fixture
def openocd(_fixture_classes_and_options: ClassCliOptions) -> Optional['OpenOcd']:
    """An openocd subprocess that could read/redirect/write"""
    if 'openocd' not in _fixture_classes_and_options.classes:
        return None

    cls = _fixture_classes_and_options.classes['openocd']
    kwargs = _fixture_classes_and_options.kwargs['openocd']
    return cls(**_drop_none_kwargs(kwargs))

gdb(_fixture_classes_and_options)

A gdb subprocess that could read/redirect/write

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_generator_fixture
def gdb(_fixture_classes_and_options: ClassCliOptions) -> Optional['Gdb']:
    """A gdb subprocess that could read/redirect/write"""
    if 'gdb' not in _fixture_classes_and_options.classes:
        return None

    cls = _fixture_classes_and_options.classes['gdb']
    kwargs = _fixture_classes_and_options.kwargs['gdb']
    return cls(**_drop_none_kwargs(kwargs))

qemu(_fixture_classes_and_options)

A qemu subprocess that could read/redirect/write

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_generator_fixture
def qemu(_fixture_classes_and_options: ClassCliOptions) -> Optional['Qemu']:
    """A qemu subprocess that could read/redirect/write"""
    if 'qemu' not in _fixture_classes_and_options.classes:
        return None

    cls = _fixture_classes_and_options.classes['qemu']
    kwargs = _fixture_classes_and_options.kwargs['qemu']
    return cls(**_drop_none_kwargs(kwargs))

dut(_fixture_classes_and_options, app, serial, openocd, gdb, qemu)

A device under test (DUT) object that could gather output from various sources and redirect them to the pexpect process, and run expect() via its pexpect process.

Source code in pytest_embedded/plugin.py
@pytest.fixture
@multi_dut_generator_fixture
def dut(
    _fixture_classes_and_options: ClassCliOptions,
    app: App,
    serial: Optional['Serial'],
    openocd: Optional['OpenOcd'],
    gdb: Optional['Gdb'],
    qemu: Optional['Qemu'],
) -> Dut:
    """
    A device under test (DUT) object that could gather output from various sources and redirect them to the pexpect
    process, and run `expect()` via its pexpect process.
    """
    cls = _fixture_classes_and_options.classes['dut']
    kwargs = _fixture_classes_and_options.kwargs['dut']

    for k, v in kwargs.items():
        if v is None:
            if k == 'app':
                kwargs[k] = app
            elif k == 'serial':
                kwargs[k] = serial
            elif k == 'openocd':
                kwargs[k] = openocd
            elif k == 'gdb':
                kwargs[k] = gdb
            elif k == 'qemu':
                kwargs[k] = qemu
    return cls(**_drop_none_kwargs(kwargs))