diff --git a/doc/configuration.rst b/doc/configuration.rst index 57b23f13c5..35ab3ab3b9 100644 --- a/doc/configuration.rst +++ b/doc/configuration.rst @@ -991,6 +991,7 @@ Arguments: Used by: - `OpenOCDDriver`_ + - `PyOCDDriver`_ NetworkUSBDebugger ~~~~~~~~~~~~~~~~~~ @@ -1971,7 +1972,6 @@ Binds to: Implements: - :any:`ConsoleProtocol` - - :any:`ResetProtocol` Arguments: - txdelay (float, default=0.0): time in seconds to wait before sending a chunk @@ -2331,6 +2331,40 @@ Arguments: - board_config (str): optional, board config in the ``openocd/scripts/board/`` directory - load_commands (list of str): optional, load commands to use instead of ``init``, ``bootstrap {filename}``, ``shutdown`` +PyOCDDriver +~~~~~~~~~~~~~ +An :any:`PyOCDDriver` controls *PyOCD* to bootstrap a target with a bootloader. + +Binds to: + interface: + - `USBDebugger`_ + - `NetworkUSBDebugger`_ + +Implements: + - :any:`BootstrapProtocol` + - :any:`ResetProtocol` + +.. code-block:: yaml + + PyOCDDriver: + image: 'bitstream' + target_name: 'nrf52832' + frequency: "400khz" + config: 'pyocd.yaml' + serial: '1234567890' + load_commands: + - 'init' + - 'svf -quiet {filename}' + - 'exit' + +Arguments: + - image (str): optional, name of the image to bootstrap onto the device + - target_name (str): optional, set the target type. Should be one of the targets listed by 'pyocd list --targets'. + - frequency (str): optional, SWD/JTAG clock frequency in Hz + - config (str): optional, PyOCD configuration file + - serial (str): optional, USB-Serial to identify device if multiple are present. + - load_commands (list of str): optional, additional command line parameters for load to use instead of ``-e``, ``sector`` + QuartusHPSDriver ~~~~~~~~~~~~~~~~ A :any:`QuartusHPSDriver` controls the "Quartus Prime Programmer and Tools" to diff --git a/doc/man/device-config.rst b/doc/man/device-config.rst index d0e5ff9148..d3ca9e0f1a 100644 --- a/doc/man/device-config.rst +++ b/doc/man/device-config.rst @@ -110,6 +110,10 @@ TOOLS KEYS Path to the openocd binary, used by the OpenOCDDriver. See: https://openocd.org/ +``pyocd`` + Path to the pyocd binary, used by the PyOCDDriver. + See: https://pyocd.io/ + ``quartus_hps`` Path to the quartus_hps binary, used by the QuartusHPSDriver. See: https://www.intel.com/content/www/us/en/docs/programmable/683039/22-3/hps-flash-programmer.html diff --git a/labgrid/driver/__init__.py b/labgrid/driver/__init__.py index d3cd6f55ef..75cb54afb4 100644 --- a/labgrid/driver/__init__.py +++ b/labgrid/driver/__init__.py @@ -9,6 +9,7 @@ from .fastbootdriver import AndroidFastbootDriver from .dfudriver import DFUDriver from .openocddriver import OpenOCDDriver +from .pyocddriver import PyOCDDriver from .quartushpsdriver import QuartusHPSDriver from .flashromdriver import FlashromDriver from .onewiredriver import OneWirePIODriver diff --git a/labgrid/driver/pyocddriver.py b/labgrid/driver/pyocddriver.py new file mode 100644 index 0000000000..cfeb8f1ef8 --- /dev/null +++ b/labgrid/driver/pyocddriver.py @@ -0,0 +1,108 @@ +import attr + +from ..factory import target_factory +from ..protocol import BootstrapProtocol, ResetProtocol +from ..step import step +from ..util.managedfile import ManagedFile +from ..util.helper import processwrapper +from .common import Driver + + +@target_factory.reg_driver +@attr.s(eq=False) +class PyOCDDriver(Driver, BootstrapProtocol, ResetProtocol): + + priorities = {ResetProtocol: 5} + + bindings = { + "interface": { + "USBDebugger", + "NetworkUSBDebugger", + }, + } + + image = attr.ib( + default=None, + validator=attr.validators.optional(attr.validators.instance_of(str)), + ) + load_commands = attr.ib( + default=None, + validator=attr.validators.optional(attr.validators.instance_of((str, list))), + ) + target_name = attr.ib( + default=None, + validator=attr.validators.optional(attr.validators.instance_of(str)), + ) + frequency = attr.ib( + default=None, + validator=attr.validators.optional(attr.validators.instance_of(str)), + ) + config = attr.ib( + default=None, + validator=attr.validators.optional(attr.validators.instance_of(str)), + ) + serial = attr.ib( + default=None, + validator=attr.validators.optional(attr.validators.instance_of(str)), + ) + + def __attrs_post_init__(self): + super().__attrs_post_init__() + + # FIXME make sure we always have an environment or config + if self.target.env: + self.tool = self.target.env.config.get_tool("pyocd") + if self.config: + self.config = self.target.env.config.resolve_path(self.config) + else: + self.tool = "pyocd" + + def _run_commands(self, subcommand: str, commands: list | None = None): + cmd = [self.tool, subcommand] + if self.serial: + cmd += ["--uid", self.serial] + if self.target_name is not None and (not commands or "--target" not in commands): + cmd += ["--target", self.target_name] + if self.frequency is not None and (not commands or "--frequency" not in commands): + cmd += ["--frequency", self.frequency] + + if self.config is not None: + mconfig = ManagedFile(self.config, self.interface) + mconfig.sync_to_resource() + cmd += ["--config", mconfig.get_remote_path()] + else: + cmd += ["--no-config"] + + if commands: + cmd += commands + processwrapper.check_output( + command=self.interface.wrap_command(cmd), + print_on_silent_log=True, + ) + + @Driver.check_active + @step(args=["filename"]) + def load(self, filename=None): + + if filename is None and self.image is not None and self.target.env: + filename = self.target.env.config.get_image_path(self.image) + + mf = ManagedFile(filename, self.interface) + mf.sync_to_resource() + + if self.load_commands: + if isinstance(self.load_commands, str): + commands = self.load_commands.split() + else: + commands = self.load_commands.copy() + else: + commands = ["-e", "sector"] + + commands.append(mf.get_remote_path()) + + self._run_commands("load", commands) + + @Driver.check_active + @step() + def reset(self): + self._run_commands("reset") diff --git a/labgrid/remote/client.py b/labgrid/remote/client.py index 069240b9e2..41f4391ca7 100755 --- a/labgrid/remote/client.py +++ b/labgrid/remote/client.py @@ -932,40 +932,90 @@ def _get_driver_or_new(self, target, cls, *, name=None, activate=True): target.activate(drv) return drv + def _get_power_driver_for_resource(self, target, name=None, priority_cls=None): + from ..resource.power import NetworkPowerPort, PDUDaemonPort + from ..resource.remote import NetworkUSBPowerPort, NetworkSiSPMPowerPort, NetworkSysfsGPIO + from ..resource import TasmotaPowerPort, NetworkYKUSHPowerPort + + drv = None + for resource in target.resources: + if name and resource.name != name: + continue + temp_drv = None + if isinstance(resource, NetworkPowerPort): + temp_drv = self._get_driver_or_new(target, "NetworkPowerDriver", name=name) + elif isinstance(resource, NetworkUSBPowerPort): + temp_drv = self._get_driver_or_new(target, "USBPowerDriver", name=name) + elif isinstance(resource, NetworkSiSPMPowerPort): + temp_drv = self._get_driver_or_new(target, "SiSPMPowerDriver", name=name) + elif isinstance(resource, PDUDaemonPort): + temp_drv = self._get_driver_or_new(target, "PDUDaemonDriver", name=name) + elif isinstance(resource, TasmotaPowerPort): + temp_drv = self._get_driver_or_new(target, "TasmotaPowerDriver", name=name) + elif isinstance(resource, NetworkYKUSHPowerPort): + temp_drv = self._get_driver_or_new(target, "YKUSHPowerDriver", name=name) + elif isinstance(resource, NetworkSysfsGPIO): + self._get_driver_or_new(target, "GpioDigitalOutputDriver", name=name) + temp_drv = self._get_driver_or_new(target, "DigitalOutputPowerDriver", name=name) + if temp_drv: + if not priority_cls: + return temp_drv + if not drv or temp_drv.get_priority(priority_cls) > drv.get_priority(priority_cls): + drv = temp_drv + return drv + + def _get_io_driver_for_resource(self, target, name=None, priority_cls=None): + from ..resource import ( + ModbusTCPCoil, + OneWirePIO, + HttpDigitalOutput, + WaveshareModbusTCPCoil, + Eth008DigitalOutput, + ) + from ..resource.remote import NetworkDeditecRelais8, NetworkSysfsGPIO, NetworkLXAIOBusPIO, NetworkHIDRelay + + drv = None + for resource in target.resources: + if name and resource.name != name: + continue + temp_drv = None + if isinstance(resource, WaveshareModbusTCPCoil): + temp_drv = self._get_driver_or_new(target, "WaveShareModbusCoilDriver", name=name) + elif isinstance(resource, ModbusTCPCoil): + temp_drv = self._get_driver_or_new(target, "ModbusCoilDriver", name=name) + elif isinstance(resource, Eth008DigitalOutput): + temp_drv = self._get_driver_or_new(target, "Eth008DigitalOutputDriver", name=name) + elif isinstance(resource, OneWirePIO): + temp_drv = self._get_driver_or_new(target, "OneWirePIODriver", name=name) + elif isinstance(resource, HttpDigitalOutput): + temp_drv = self._get_driver_or_new(target, "HttpDigitalOutputDriver", name=name) + elif isinstance(resource, NetworkDeditecRelais8): + temp_drv = self._get_driver_or_new(target, "DeditecRelaisDriver", name=name) + elif isinstance(resource, NetworkSysfsGPIO): + temp_drv = self._get_driver_or_new(target, "GpioDigitalOutputDriver", name=name) + elif isinstance(resource, NetworkLXAIOBusPIO): + temp_drv = self._get_driver_or_new(target, "LXAIOBusPIODriver", name=name) + elif isinstance(resource, NetworkHIDRelay): + temp_drv = self._get_driver_or_new(target, "HIDRelayDriver", name=name) + if temp_drv: + if not priority_cls: + return temp_drv + if not drv or temp_drv.get_priority(priority_cls) > drv.get_priority(priority_cls): + drv = temp_drv + return drv + def power(self): place = self.get_acquired_place() action = self.args.action delay = self.args.delay name = self.args.name target = self._get_target(place) - from ..resource.power import NetworkPowerPort, PDUDaemonPort - from ..resource.remote import NetworkUSBPowerPort, NetworkSiSPMPowerPort, NetworkSysfsGPIO - from ..resource import TasmotaPowerPort, NetworkYKUSHPowerPort drv = None try: drv = target.get_driver("PowerProtocol", name=name) except NoDriverFoundError: - for resource in target.resources: - if name and resource.name != name: - continue - if isinstance(resource, NetworkPowerPort): - drv = self._get_driver_or_new(target, "NetworkPowerDriver", name=name) - elif isinstance(resource, NetworkUSBPowerPort): - drv = self._get_driver_or_new(target, "USBPowerDriver", name=name) - elif isinstance(resource, NetworkSiSPMPowerPort): - drv = self._get_driver_or_new(target, "SiSPMPowerDriver", name=name) - elif isinstance(resource, PDUDaemonPort): - drv = self._get_driver_or_new(target, "PDUDaemonDriver", name=name) - elif isinstance(resource, TasmotaPowerPort): - drv = self._get_driver_or_new(target, "TasmotaPowerDriver", name=name) - elif isinstance(resource, NetworkYKUSHPowerPort): - drv = self._get_driver_or_new(target, "YKUSHPowerDriver", name=name) - elif isinstance(resource, NetworkSysfsGPIO): - self._get_driver_or_new(target, "GpioDigitalOutputDriver", name=name) - drv = self._get_driver_or_new(target, "DigitalOutputPowerDriver", name=name) - if drv: - break + drv = self._get_power_driver_for_resource(target, name, None) if not drv: raise UserError("target has no compatible resource available") @@ -980,42 +1030,12 @@ def digital_io(self): action = self.args.action name = self.args.name target = self._get_target(place) - from ..resource import ( - ModbusTCPCoil, - OneWirePIO, - HttpDigitalOutput, - WaveshareModbusTCPCoil, - Eth008DigitalOutput, - ) - from ..resource.remote import NetworkDeditecRelais8, NetworkSysfsGPIO, NetworkLXAIOBusPIO, NetworkHIDRelay drv = None try: drv = target.get_driver("DigitalOutputProtocol", name=name) except NoDriverFoundError: - for resource in target.resources: - if name and resource.name != name: - continue - if isinstance(resource, WaveshareModbusTCPCoil): - drv = self._get_driver_or_new(target, "WaveShareModbusCoilDriver", name=name) - elif isinstance(resource, ModbusTCPCoil): - drv = self._get_driver_or_new(target, "ModbusCoilDriver", name=name) - elif isinstance(resource, Eth008DigitalOutput): - drv = self._get_driver_or_new(target, "Eth008DigitalOutputDriver", name=name) - elif isinstance(resource, OneWirePIO): - drv = self._get_driver_or_new(target, "OneWirePIODriver", name=name) - elif isinstance(resource, HttpDigitalOutput): - drv = self._get_driver_or_new(target, "HttpDigitalOutputDriver", name=name) - elif isinstance(resource, NetworkDeditecRelais8): - drv = self._get_driver_or_new(target, "DeditecRelaisDriver", name=name) - elif isinstance(resource, NetworkSysfsGPIO): - drv = self._get_driver_or_new(target, "GpioDigitalOutputDriver", name=name) - elif isinstance(resource, NetworkLXAIOBusPIO): - drv = self._get_driver_or_new(target, "LXAIOBusPIODriver", name=name) - elif isinstance(resource, NetworkHIDRelay): - drv = self._get_driver_or_new(target, "HIDRelayDriver", name=name) - if drv: - break + drv = self._get_io_driver_for_resource(target, name, None) if not drv: raise UserError("target has no compatible resource available") @@ -1026,6 +1046,37 @@ def digital_io(self): elif action == "low": drv.set(False) + def reset(self): + place = self.get_acquired_place() + name = self.args.name + target = self._get_target(place) + from ..resource.remote import NetworkUSBDebugger + from ..protocol import ResetProtocol + + drv = None + try: + drv = target.get_driver(ResetProtocol, name=name) + except NoDriverFoundError: + drv = self._get_power_driver_for_resource(target, name, ResetProtocol) + io_drv = self._get_io_driver_for_resource(target, name, None) + if io_drv: + temp_drv = self._get_driver_or_new(target, "DigitalOutputResetDriver", name=name) + if temp_drv and (not drv or temp_drv.get_priority(ResetProtocol) > drv.get_priority(ResetProtocol)): + drv = temp_drv + for resource in target.resources: + if name and resource.name != name: + continue + temp_drv = None + if isinstance(resource, NetworkUSBDebugger): + temp_drv = self._get_driver_or_new(target, "PyOCDDriver", name=name) + if temp_drv: + if not drv or temp_drv.get_priority(ResetProtocol) > drv.get_priority(ResetProtocol): + drv = temp_drv + + if not drv: + raise UserError("target has no compatible resource available") + drv.reset() + async def _console(self, place, target, timeout, *, logfile=None, loop=False, listen_only=False): name = self.args.name from ..resource import NetworkSerialPort @@ -1176,8 +1227,9 @@ def bootstrap(self): NetworkIMXUSBLoader, NetworkRKUSBLoader, NetworkAlteraUSBBlaster, + NetworkUSBDebugger, ) - from ..driver import OpenOCDDriver + from ..driver import OpenOCDDriver, PyOCDDriver drv = None try: @@ -1199,6 +1251,13 @@ def bootstrap(self): except NoDriverFoundError: drv = OpenOCDDriver(target, name=name, **args) drv.interface.timeout = self.args.wait + elif isinstance(resource, NetworkUSBDebugger): + args = dict(arg.split("=", 1) for arg in self.args.bootstrap_args) + try: + drv = target.get_driver("PyOCDDriver", activate=False, name=name) + except NoDriverFoundError: + drv = PyOCDDriver(target, name=name, **args) + drv.interface.timeout = self.args.wait elif isinstance(resource, NetworkRKUSBLoader): drv = self._get_driver_or_new(target, "RKUSBDriver", activate=False, name=name) drv.loader.timeout = self.args.wait @@ -1989,6 +2048,10 @@ def get_parser(auto_doc_mode=False) -> "argparse.ArgumentParser | AutoProgramArg subparser.add_argument("--name", "-n", help="optional resource name") subparser.set_defaults(func=ClientSession.power) + subparser = subparsers.add_parser("reset", help="reset the target") + subparser.add_argument("--name", "-n", help="optional resource name") + subparser.set_defaults(func=ClientSession.reset) + subparser = subparsers.add_parser("io", help="change (or get) a digital IO status") subparser.add_argument("action", choices=["high", "low", "get"], help="action") subparser.add_argument("name", help="optional resource name", nargs="?") diff --git a/labgrid/resource/udev.py b/labgrid/resource/udev.py index 7ecd4b01a7..517b00bd96 100644 --- a/labgrid/resource/udev.py +++ b/labgrid/resource/udev.py @@ -932,6 +932,7 @@ def filter_match(self, device): ("064b", "2507"), # Analog Devices Onboard Debug Agent ("064b", "2508"), # Analog Devices Onboard Debug Agent ("064b", "250A"), # Analog Devices Onboard Debug Agent + ("c251", "f001"), # Keil CMSIS-DAP-Link ]: return False diff --git a/man/labgrid-client.1 b/man/labgrid-client.1 index 45533af05d..befda6053a 100644 --- a/man/labgrid-client.1 +++ b/man/labgrid-client.1 @@ -651,6 +651,22 @@ format output as shell variables .B \-\-prio priority relative to other reservations (default 0) .UNINDENT +.SS labgrid\-client reset +.sp +reset the target +.INDENT 0.0 +.INDENT 3.5 +.sp +.EX +usage: labgrid\-client reset [\-\-name NAME] +.EE +.UNINDENT +.UNINDENT +.INDENT 0.0 +.TP +.B \-\-name , \-n +optional resource name +.UNINDENT .SS labgrid\-client resources|r .sp list available resources diff --git a/man/labgrid-device-config.5 b/man/labgrid-device-config.5 index 0755927deb..c989e4fe9f 100644 --- a/man/labgrid-device-config.5 +++ b/man/labgrid-device-config.5 @@ -125,6 +125,10 @@ See: \X'tty: link https://git.pengutronix.de/cgit/barebox/tree/scripts/mxs-usb-l Path to the openocd binary, used by the OpenOCDDriver. See: \X'tty: link https://openocd.org/'\fI\%https://openocd.org/\fP\X'tty: link' .TP +.B \fBpyocd\fP +Path to the pyocd binary, used by the PyOCDDriver. +See: \X'tty: link https://pyocd.io/'\fI\%https://pyocd.io/\fP\X'tty: link' +.TP .B \fBquartus_hps\fP Path to the quartus_hps binary, used by the QuartusHPSDriver. See: \X'tty: link https://www.intel.com/content/www/us/en/docs/programmable/683039/22-3/hps-flash-programmer.html'\fI\%https://www.intel.com/content/www/us/en/docs/programmable/683039/22\-3/hps\-flash\-programmer.html\fP\X'tty: link' diff --git a/tests/test_client_commands.py b/tests/test_client_commands.py new file mode 100644 index 0000000000..2433824037 --- /dev/null +++ b/tests/test_client_commands.py @@ -0,0 +1,176 @@ +import pytest + +from labgrid.exceptions import NoSupplierFoundError +from labgrid.remote.client import ensure_event_loop, ClientSession, UserError +from labgrid.driver import PyOCDDriver, NetworkPowerDriver, LXAIOBusPIODriver +from labgrid.resource.remote import NetworkUSBDebugger, NetworkLXAIOBusPIO +from labgrid.resource.power import NetworkPowerPort + + +class PseudoArgs: + def __getattr__(self, attr): + if attr == "action": + return "get" + if attr == "bootstrap_args": + return "" + return None + + +class PseudePlace: + def __getattr__(self, attr): + if attr == "name": + return "PseudoPlace" + return None + + +@pytest.fixture(scope="function") +def client(mocker, target): + loop = ensure_event_loop() + client = ClientSession(address="", loop=loop) + client.args = PseudoArgs() + place = PseudePlace() + + def get_acquired_place(): + return place + place_mock = mocker.patch("labgrid.remote.client.ClientSession.get_acquired_place") + place_mock.side_effect = get_acquired_place + + def _get_target(_): + return target + target_mock = mocker.patch("labgrid.remote.client.ClientSession._get_target") + target_mock.side_effect = _get_target + + # disable resource updating + target_resource_update_mock = mocker.patch("labgrid.target.Target.update_resources") + return client + + +@pytest.mark.parametrize( + "command", + [ + "bootstrap", + "digital_io", + "power", + "reset", + "sd_mux", + "usb_mux", + "video", + ], +) +def test_command_without_resources(client, command): + with pytest.raises(UserError, match="target has no compatible resource available"): + c = getattr(client, command) + c() + + +@pytest.mark.parametrize( + "command", + [ + "dfu", + "fastboot", + "flashscript", + "audio", + "write_files", + "write_image", + "ssh", + "scp", + "rsync", + "sshfs", + "telnet", + ], +) +def test_command_without_suplierers(client, command): + with pytest.raises(NoSupplierFoundError): + c = getattr(client, command) + c() + + +def network_usb_ressource(target): + network_args = { + "name": "network_usb_ressource", + "host": "None", + "busnum": 0, + "devnum": 0, + "path": "1-12", + "vendor_id": 0, + "model_id": 0, + } + r = NetworkUSBDebugger(target, **network_args) + r.avail = True + + +def pyocd_driver(target): + network_usb_ressource(target) + PyOCDDriver(target, "pyocd") + + +def network_power_port(target): + network_args = { + "name": "network_power_port", + "host": "None", + "model": "rest", + "index": "0", + } + r = NetworkPowerPort(target, **network_args) + r.avail = True + + +def network_power_driver(target): + network_power_port(target) + NetworkPowerDriver(target, "fake_power") + + +def network_LXA_gpio(target): + network_args = { + "name": "network_lxa_io_bus_name", + "node": "network_lxa_io_bus_node", + "host": "None", + "pin": "0", + "invert": False, + } + r = NetworkLXAIOBusPIO(target, **network_args) + r.avail = True + + +def lxaio_bus_driver(target): + network_LXA_gpio(target) + LXAIOBusPIODriver(target, "lxa_pio_driver") + + +def power_and_gpio_ressource(target): + network_LXA_gpio(target) + network_power_port(target) + + +def power_gpio_and_debug_ressource(target): + network_LXA_gpio(target) + network_power_port(target) + network_usb_ressource(target) + + +@pytest.mark.parametrize( + "command,setup,patch", + [ + ("reset", network_usb_ressource, "labgrid.driver.pyocddriver.PyOCDDriver.reset"), + ("reset", pyocd_driver, "labgrid.driver.pyocddriver.PyOCDDriver.reset"), + ("power", network_power_port, "labgrid.driver.power.rest.power_get"), + ("power", network_power_driver, "labgrid.driver.power.rest.power_get"), + ("digital_io", network_LXA_gpio, "labgrid.driver.lxaiobusdriver.LXAIOBusPIODriver.get"), + ("digital_io", lxaio_bus_driver, "labgrid.driver.lxaiobusdriver.LXAIOBusPIODriver.get"), + ("reset", network_power_port, "labgrid.driver.powerdriver.PowerResetMixin.reset"), + ("reset", network_LXA_gpio, "labgrid.driver.resetdriver.DigitalOutputResetDriver.reset"), + ("reset", power_and_gpio_ressource, "labgrid.driver.resetdriver.DigitalOutputResetDriver.reset"), + ("reset", power_gpio_and_debug_ressource, "labgrid.driver.pyocddriver.PyOCDDriver.reset"), + ("bootstrap", network_usb_ressource, "labgrid.driver.pyocddriver.PyOCDDriver.load"), + ("bootstrap", pyocd_driver, "labgrid.driver.pyocddriver.PyOCDDriver.load"), + ], +) +def test_reset_command_with_remote_resource(target, client, mocker, command, setup, patch): + setup(target) + + driver_mock = mocker.patch(patch) + + c = getattr(client, command) + c() + + driver_mock.assert_called_once() diff --git a/tests/test_pyocd.py b/tests/test_pyocd.py new file mode 100644 index 0000000000..8ed25de5ce --- /dev/null +++ b/tests/test_pyocd.py @@ -0,0 +1,245 @@ +import pytest + +from labgrid import Environment, Target +from labgrid.resource.udev import USBDebugger +from labgrid.driver.pyocddriver import PyOCDDriver + +LABGRID_TEST_SERIAL = "LABGRID_TEST_SER" +LABGRID_TARGET = "target_mcu" +LABGRID_TOOL_CMD = "/some/path/to/nowhere" + + +def test_pyocd_driver_activate(target): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None) + target.activate(d) + + +def test_pyocd_driver_reset(target, mocker): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None) + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.reset() + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd == ["pyocd", "reset", "--no-config"] + assert check_output_mock.call_args.kwargs["print_on_silent_log"] + + +def test_pyocd_driver_load(target, mocker): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None) + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.load(__file__) + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd == ["pyocd", "load", "--no-config", "-e", "sector", __file__] + assert check_output_mock.call_args.kwargs["print_on_silent_log"] + + +def test_pyocd_driver_multiple_load_with_load_commands(target, mocker): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None, load_commands=["--frequency", "100"]) + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.load(__file__) + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd == ["pyocd", "load", "--no-config", "--frequency", "100", __file__] + assert check_output_mock.call_args.kwargs["print_on_silent_log"] + + check_output_mock.reset_mock() + + d.load(__file__) + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd == ["pyocd", "load", "--no-config", "--frequency", "100", __file__] + assert check_output_mock.call_args.kwargs["print_on_silent_log"] + + +def test_pyocd_load_error_on_missing_file(target, mocker): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None) + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + with pytest.raises(FileNotFoundError): + d.load() + + check_output_mock.assert_not_called() + + +def test_pyocd_load_error_on_missing_target_env(target, mocker): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None, image="test") + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + with pytest.raises(FileNotFoundError): + d.load() + + check_output_mock.assert_not_called() + + +def test_pyocd_load_image_on_missing_file(tmpdir, mocker): + p = tmpdir.join("config.yaml") + p.write(f""" + images: + test: {__file__} + """) + env = Environment(str(p)) + target = Target("test", env=env) + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None, image="test") + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.load() + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd[0] == "pyocd" + assert cmd[1] == "load" + assert cmd[-1] == __file__ + + +@pytest.mark.parametrize( + "args,params,missing", + [ + ({"serial": LABGRID_TEST_SERIAL}, ["--uid", LABGRID_TEST_SERIAL], []), + ({"target_name": LABGRID_TARGET}, ["--target", LABGRID_TARGET], []), + ({"config": __file__}, ["--config", __file__], ["--no-config"]), + # make sure, that it is not the file loaded argument + ({}, ["--no-config"], [__file__]), + ({"frequency": "400000"}, ["--frequency", "400000"], []), + # load commands should take precedens + ({"frequency": "400000", "load_commands": ["--frequency", "100"]}, ["--frequency", "100"], ["400000"]), + # even as a string instead of array + ({"frequency": "400000", "load_commands": "--frequency 100"}, ["--frequency", "100"], ["400000"]), + # load commands overwrite default load commands + ({"load_commands": ["--frequency", "100"]}, ["--frequency", "100"], ["-e", "sector"]), + ], +) +def test_pyocd_driver_params_load(target, mocker, args, params, missing): + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None, **args) + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.load(__file__) + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd[0] == "pyocd" + assert cmd[1] == "load" + assert cmd[-1] == __file__ + for p in params: + assert p in cmd[2:-1] + for m in missing: + assert m not in cmd[2:-1] + assert check_output_mock.call_args.kwargs["print_on_silent_log"] + + +@pytest.mark.parametrize( + "args,params,missing", + [ + ({"serial": LABGRID_TEST_SERIAL}, ["--uid", LABGRID_TEST_SERIAL], []), + ({"target_name": LABGRID_TARGET}, ["--target", LABGRID_TARGET], []), + ({"config": __file__}, ["--config", __file__], ["--no-config"]), + ({"frequency": "400000"}, ["--frequency", "400000"], []), + # load commands not be used + ({"frequency": "400000", "load_commands": ["--frequency", "100"]}, ["--frequency", "400000"], ["100"]), + ({"load_commands": ["--frequency", "100"]}, [], ["--frequency", "100"]), + ], +) +def test_pyocd_driver_params_reset(tmpdir, mocker, args, params, missing): + p = tmpdir.join("config.yaml") + p.write(""" + dict: {} + """) + env = Environment(str(p)) + target = Target("test", env=env) + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None, **args) + target.activate(d) + + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.reset() + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd[0] == "pyocd" + assert cmd[1] == "reset" + for p in params: + assert p in cmd[2:] + for m in missing: + assert m not in cmd[2:] + assert check_output_mock.call_args.kwargs["print_on_silent_log"] + + +def test_pyocd_respects_env_tool(mocker, tmpdir): + p = tmpdir.join("config.yaml") + p.write(f""" + tools: + pyocd: {LABGRID_TOOL_CMD} + """) + env = Environment(str(p)) + target = Target("test", env=env) + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None) + target.activate(d) + + assert d.tool == LABGRID_TOOL_CMD + + # Make sure, that the tool actually is used in command + check_output_mock = mocker.patch("labgrid.util.helper.processwrapper.check_output") + + d.reset() + + check_output_mock.assert_called_once() + cmd = check_output_mock.call_args.kwargs["command"] + assert cmd[0] == LABGRID_TOOL_CMD + assert cmd[1] == "reset" + + +def test_pyocd_respects_missing_env_tool(tmpdir): + p = tmpdir.join("config.yaml") + p.write(f""" + tools: + none: {LABGRID_TOOL_CMD} + """) + env = Environment(str(p)) + target = Target("test", env=env) + r = USBDebugger(target, name=None, match={"sys_name": "1-12"}) + r.avail = True + d = PyOCDDriver(target, name=None) + target.activate(d) + + assert d.tool == "pyocd"