twister: pytest: Add support for mcumgr go tool with BLE (c6b3b934) · Commits · Wenxi XU / zephyr

scripts/pylib/pytest-twister-harness/src/twister_harness/init.py

+2 −2

Original line number	Diff line number	Diff line
		@@ -5,9 +5,9 @@
		# flake8: noqa

		from twister_harness.device.device_adapter import DeviceAdapter
		from twister_harness.helpers.mcumgr import MCUmgr
		from twister_harness.helpers.mcumgr import MCUmgr, MCUmgrBle
		from twister_harness.helpers.shell import Shell

		__all__ = ['DeviceAdapter', 'MCUmgr', 'Shell']
		__all__ = ['DeviceAdapter', 'MCUmgr', 'MCUmgrBle', 'Shell']

		__version__ = '0.0.1'

scripts/pylib/pytest-twister-harness/src/twister_harness/fixtures.py

+31 −6

Original line number	Diff line number	Diff line
		@@ -13,7 +13,7 @@ from twister_harness.device.device_adapter import DeviceAdapter
		from twister_harness.device.factory import DeviceFactory
		from twister_harness.twister_harness_config import DeviceConfig, TwisterHarnessConfig
		from twister_harness.helpers.shell import Shell
		from twister_harness.helpers.mcumgr import MCUmgr
		from twister_harness.helpers.mcumgr import MCUmgr, MCUmgrBle
		from twister_harness.helpers.utils import find_in_config

		logger = logging.getLogger(__name__)
		@@ -46,7 +46,9 @@ def determine_scope(fixture_name, config):


		@pytest.fixture(scope=determine_scope)
		def unlaunched_dut(request: pytest.FixtureRequest, device_object: DeviceAdapter) -> Generator[DeviceAdapter, None, None]:
		def unlaunched_dut(
		request: pytest.FixtureRequest, device_object: DeviceAdapter
		) -> Generator[DeviceAdapter, None, None]:
		"""Return device object - with logs connected, but not run"""
		device_object.initialize_log_files(request.node.name)
		try:
		@@ -54,6 +56,7 @@ def unlaunched_dut(request: pytest.FixtureRequest, device_object: DeviceAdapter)
		finally: # to make sure we close all running processes execution
		device_object.close()


		@pytest.fixture(scope=determine_scope)
		def dut(request: pytest.FixtureRequest, device_object: DeviceAdapter) -> Generator[DeviceAdapter, None, None]:
		"""Return launched device - with run application."""
		@@ -83,12 +86,34 @@ def shell(dut: DeviceAdapter) -> Shell:
		return shell


		@pytest.fixture(scope='session')
		def is_mcumgr_available() -> None:
		@pytest.fixture()
		def mcumgr(device_object: DeviceAdapter) -> Generator[MCUmgr, None, None]:
		"""Fixture to create an MCUmgr instance for serial connection."""
		if not MCUmgr.is_available():
		pytest.skip('mcumgr not available')
		yield MCUmgr.create_for_serial(device_object.device_config.serial)


		@pytest.fixture()
		def mcumgr(is_mcumgr_available: None, dut: DeviceAdapter) -> Generator[MCUmgr, None, None]:
		yield MCUmgr.create_for_serial(dut.device_config.serial)
		def mcumgr_ble(device_object: DeviceAdapter) -> Generator[MCUmgrBle, None, None]:
		"""Fixture to create an MCUmgr instance for BLE connection."""
		if not MCUmgrBle.is_available():
		pytest.skip('mcumgr for ble not available')

		for fixture in device_object.device_config.fixtures:
		if fixture.startswith('usb_hci:'):
		hci_name = fixture.split(':', 1)[1]
		break
		else:
		pytest.skip('usb_hci fixture not found')

		try:
		hci_index = int(hci_name.split('hci')[-1])
		except ValueError:
		pytest.skip(f'Invalid HCI name: {hci_name}. Expected format is "hciX".')

		peer_name = find_in_config(
		Path(device_object.device_config.app_build_dir) / 'zephyr' / '.config', 'CONFIG_BT_DEVICE_NAME'
		) or 'Zephyr'

		yield MCUmgrBle.create_for_ble(hci_index, peer_name)

scripts/pylib/pytest-twister-harness/src/twister_harness/helpers/mcumgr.py

+23 −0

Original line number	Diff line number	Diff line
		@@ -4,8 +4,10 @@
		from __future__ import annotations

		import logging
		import os
		import re
		import shlex
		import shutil
		from dataclasses import dataclass
		from pathlib import Path
		from subprocess import check_output, getstatusoutput
		@@ -113,3 +115,24 @@ class MCUmgr:
		if not hash:
		hash = self.get_hash_to_confirm()
		self.run_command(f'image confirm {hash}')


		class MCUmgrBle(MCUmgr):
		"""MCUmgr wrapper for BLE connection"""

		@classmethod
		def create_for_ble(cls, hci_index: int, peer_name: str) -> MCUmgr:
		"""Create MCUmgr instance for BLE connection"""
		connection_string = (
		f'--conntype ble --hci {hci_index} '
		f'--connstring peer_name="{peer_name}"'
		)
		return cls(connection_options=connection_string)

		@classmethod
		def is_available(cls) -> bool:
		"""Check if mcumgr is available. For BLE, it requires root privileges."""
		if os.getuid() != 0 and 'sudo' not in cls.mcumgr_exec:
		mcumgr_path = shutil.which(cls.mcumgr_exec)
		cls.mcumgr_exec = f'sudo {mcumgr_path}'
		return super().is_available()

Admin message