|
| 1 | +.. SPDX-License-Identifier: GPL-2.0-or-later |
| 2 | +
|
| 3 | +Plugin System |
| 4 | +============= |
| 5 | + |
| 6 | +Sometimes, we need to cover complex testing scenarios where our System Under |
| 7 | +Test utilizes specific protocols and infrastructures to communicate with our |
| 8 | +host machine and execute LTP tests. |
| 9 | + |
| 10 | +For this reason, Kirk provides a plugin system to recognize custom ``SUT`` |
| 11 | +and ``ComChannel`` class implementations inside any external folder. These |
| 12 | +classes are used to implement complex scenarios, and in the next sections, we |
| 13 | +will see how to communicate with the plugin system. |
| 14 | + |
| 15 | +To verify supported ``SUT``, please run: |
| 16 | + |
| 17 | +.. code-block:: bash |
| 18 | +
|
| 19 | + kirk --sut help |
| 20 | +
|
| 21 | +.. note:: |
| 22 | + |
| 23 | + If you want to implement a new ``ComChannel`` communication handler, please |
| 24 | + refer to the natively supported implementations such as ``shell.py``. |
| 25 | + |
| 26 | +Custom System Under Test |
| 27 | +------------------------ |
| 28 | + |
| 29 | +All the channels implementations provided by kirk can be used or duplicated to |
| 30 | +use them inside our ``SUT``. The way we create these instances is as follows: |
| 31 | + |
| 32 | +.. code-block:: bash |
| 33 | +
|
| 34 | + kirk --plugins my/plugins/folder \ |
| 35 | + --com ssh:host=192.168.0.1:id=ssh_host0 \ |
| 36 | + --sut mysut \ |
| 37 | + --run-suite syscalls |
| 38 | +
|
| 39 | +We just created a SSH channel ``ssh_host0`` that can be used by ``mysut`` |
| 40 | +implementation in order to setup testing. |
| 41 | + |
| 42 | +Our ``SUT`` can now use the ``libkirk.com.get_channels()`` utility to read |
| 43 | +available channels and get the one we need, as follows: |
| 44 | + |
| 45 | +.. code-block:: python |
| 46 | +
|
| 47 | + def setup(self, **kwargs: Dict[str, Any]) -> None: |
| 48 | + self._ssh = next( |
| 49 | + (c for in libkirk.com.get_channels() if c.name == "ssh_host0"), |
| 50 | + None, |
| 51 | + ) |
| 52 | +
|
| 53 | +But remember that **only one** channel must be given back to kirk in order to |
| 54 | +communicate with the System Under Test via ``get_channel()`` API. |
| 55 | + |
| 56 | +.. code-block:: python |
| 57 | +
|
| 58 | + def get_channel() -> ComChannel: |
| 59 | + return self._ssh |
| 60 | +
|
| 61 | +Practical example |
| 62 | +----------------- |
| 63 | + |
| 64 | +We might want to test LTP inside an embedded system on our desk via SSH. |
| 65 | +We have two scripts to run before communicating with the SUT: |
| 66 | + |
| 67 | +- ``install_firmware.sh`` to install a new firmware |
| 68 | +- ``reboot_board.sh`` to reboot board if it's not responding anymore |
| 69 | + |
| 70 | +The idea is that we install a new firmware before running tests, run tests and |
| 71 | +if system breaks/panic/timeout, we reboot it, continuing testing suite from |
| 72 | +where we left. |
| 73 | + |
| 74 | +We can easily achieve this scenario with the following implementation: |
| 75 | + |
| 76 | +.. code-block:: python |
| 77 | +
|
| 78 | + import os |
| 79 | + from typing import Dict, Optional |
| 80 | +
|
| 81 | + import libkirk.com |
| 82 | + from libkirk.com import ComChannel, IOBuffer |
| 83 | + from libkirk.errors import SUTError |
| 84 | + from libkirk.sut import SUT |
| 85 | +
|
| 86 | +
|
| 87 | + class EmbeddedSUT(SUT): |
| 88 | + # This is needed by kirk to know what is the name of the SUT |
| 89 | + # we are implementing |
| 90 | + _name = "embedded" |
| 91 | +
|
| 92 | + def __init__(self) -> None: |
| 93 | + self._ssh = None |
| 94 | + self._shell = None |
| 95 | +
|
| 96 | + currdir = os.path.dirname(os.path.realpath(__file__)) |
| 97 | + self._install_sh = os.path.join(currdir, "install_firmware.sh") |
| 98 | + self._reboot_sh = os.path.join(currdir, "reboot_board.sh") |
| 99 | +
|
| 100 | + def setup(self, **kwargs: Dict[str, str]) -> None: |
| 101 | + # Here we fetch all data we need. At this point we know that kirk |
| 102 | + # already initialized all communication channels |
| 103 | + chan_name = kwargs.get("com", "ssh") |
| 104 | +
|
| 105 | + self._ssh = next( |
| 106 | + (c for c in libkirk.com.get_channels() if c.name == chan_name), None |
| 107 | + ) |
| 108 | + self._shell = next( |
| 109 | + (c for c in libkirk.com.get_channels() if c.name == "shell"), None |
| 110 | + ) |
| 111 | +
|
| 112 | + if not self._ssh: |
| 113 | + raise SUTError(f"Can't find channel '{chan_name}'") |
| 114 | +
|
| 115 | + @property |
| 116 | + def config_help(self) -> Dict[str, str]: |
| 117 | + # Parameters to setup our SUT |
| 118 | + return { |
| 119 | + "com": "Communication channel (default: ssh)", |
| 120 | + } |
| 121 | +
|
| 122 | + def get_channel(self) -> ComChannel: |
| 123 | + # Here we return our main communication channel |
| 124 | + return self._ssh |
| 125 | +
|
| 126 | + async def start(self, iobuffer: Optional[IOBuffer] = None) -> None: |
| 127 | + # Initialize the SUT by running commands, scripts and everything |
| 128 | + # that can be done via our communication channels |
| 129 | + if await self.is_running: |
| 130 | + return |
| 131 | +
|
| 132 | + await self._shell.ensure_communicate(iobuffer=iobuffer) |
| 133 | +
|
| 134 | + ret = await self._shell.run_command(self._install_sh, iobuffer=iobuffer) |
| 135 | + if ret["returncode"] != 0: |
| 136 | + raise SUTError(f"{self._install_sh} failed") |
| 137 | +
|
| 138 | + await self._ssh.ensure_communicate(iobuffer=iobuffer) |
| 139 | +
|
| 140 | + async def stop(self, iobuffer: Optional[IOBuffer] = None) -> None: |
| 141 | + # Stop any operation in our SUT. This can be requires in any moment |
| 142 | + # during tests run |
| 143 | + if not await self.is_running: |
| 144 | + return |
| 145 | +
|
| 146 | + await self._ssh.stop(iobuffer=iobuffer) |
| 147 | +
|
| 148 | + async def restart(self, iobuffer: Optional[IOBuffer] = None) -> None: |
| 149 | + # Stop any operation in our SUT and restart the system |
| 150 | + await self.stop(iobuffer=iobuffer) |
| 151 | +
|
| 152 | + ret = await self._shell.run_command(self._reboot_sh, iobuffer=iobuffer) |
| 153 | + if ret["returncode"] != 0: |
| 154 | + raise SUTError(f"{self._reboot_sh} failed") |
| 155 | +
|
| 156 | + await self._shell.stop(iobuffer=iobuffer) |
| 157 | + await self.start(iobuffer=iobuffer) |
| 158 | +
|
| 159 | + @property |
| 160 | + async def is_running(self) -> bool: |
| 161 | + # Tell kirk when SUT is operating or not |
| 162 | + return await self._ssh.active |
| 163 | +
|
| 164 | +
|
| 165 | +Let's suppose we have a ``$HOME/plugins`` folder where we placed our |
| 166 | +``EmbeddedSUT`` implementation and its scripts. Then we can run ``syscalls`` |
| 167 | +testing suite with kirk as following: |
| 168 | + |
| 169 | +.. code-block:: python |
| 170 | +
|
| 171 | + kirk --plugins $HOME/plugins \ |
| 172 | + --sut embedded \ |
| 173 | + --com ssh:host=192.168.0.1:user=root:key_file=/home/user/.ssh/id_rsa \ |
| 174 | + --run-suite syscalls |
0 commit comments