scheduler: add support for SUT reboot tests

libkirk/channels/ltx_chan.py

@@ -76,6 +76,10 @@ def setup(self, **kwargs: Dict[str, Any]) -> None:
     def parallel_execution(self) -> bool:
         return True
 
+    @property
+    def supports_reboot(self) -> bool:
+        return True
+
     async def active(self) -> bool:
         if not self._ltx:
             return False

libkirk/channels/qemu.py

@@ -220,6 +220,10 @@ def config_help(self) -> Dict[str, str]:
     def parallel_execution(self) -> bool:
         return False
 
+    @property
+    def supports_reboot(self) -> bool:
+        return True
+
     async def active(self) -> bool:
         if self._proc is None:
             return False

libkirk/channels/ssh.py

@@ -246,6 +246,10 @@ def setup(self, **kwargs: Any) -> None:
     def parallel_execution(self) -> bool:
         return True
 
+    @property
+    def supports_reboot(self) -> bool:
+        return True
+
     async def active(self) -> bool:
         return self._conn is not None
 

libkirk/com.py

@@ -6,6 +6,7 @@
 .. moduleauthor:: Andrea Cervesato <andrea.cervesato@suse.com>
 """
 
+import asyncio
 from typing import (
     Any,
     Dict,
@@ -54,6 +55,14 @@ def parallel_execution(self) -> bool:
         """
         raise NotImplementedError()
 
+    @property
+    def supports_reboot(self) -> bool:
+        """
+        :return: If True, communication supports SUT reboot.
+        :rtype: bool
+        """
+        return False
+
     async def active(self) -> bool:
         """
         :return: Return True if communication is active. False otherwise.
@@ -134,7 +143,10 @@ async def fetch_file(self, target_path: str) -> bytes:
         raise NotImplementedError()
 
     async def ensure_communicate(
-        self, iobuffer: Optional[IOBuffer] = None, retries: int = 10
+        self,
+        iobuffer: Optional[IOBuffer] = None,
+        retries: int = 10,
+        delay: float = 0.0,
     ) -> None:
         """
         Ensure that communicate is completed, retrying as many times we
@@ -145,6 +157,8 @@ async def ensure_communicate(
         :type iobuffer: IOBuffer
         :param retries: Number of times we retry to communicate.
         :type retries: int
+        :param delay: Delay between retries in seconds.
+        :type delay: float
         """
         retries = max(retries, 1)
 
@@ -157,6 +171,8 @@ async def ensure_communicate(
                     raise err
 
                 await self.stop(iobuffer=iobuffer)
+                if delay > 0:
+                    await asyncio.sleep(delay)
 
 
 def discover(path: str, extend: bool = True) -> None:

libkirk/data.py

@@ -26,6 +26,7 @@ def __init__(
         env: Optional[Dict[str, str]] = None,
         args: Optional[List[str]] = None,
         parallelizable: bool = False,
+        reboots_sut: bool = False,
     ) -> None:
         """
         :param name: Name of the test.
@@ -40,6 +41,8 @@ def __init__(
         :type args: list(str)
         :param parallelizable: If True, test can be run in parallel.
         :type parallelizable: bool
+        :param reboots_sut: If True, test reboots the SUT.
+        :type reboots_sut: bool
         """
         if not name:
             raise ValueError("Test must have a name")
@@ -52,7 +55,8 @@ def __init__(
         self._cwd = cwd
         self._args = args if args else []
         self._env = env if env else {}
-        self._parallelizable = parallelizable
+        self._reboots_sut = reboots_sut
+        self._parallelizable = False if reboots_sut else parallelizable
 
     def __repr__(self) -> str:
         return (
@@ -61,7 +65,8 @@ def __repr__(self) -> str:
             f"arguments: {self._args}, "
             f"cwd: '{self._cwd}', "
             f"environ: '{self._env}', "
-            f"parallelizable: {self._parallelizable}"
+            f"parallelizable: {self._parallelizable}, "
+            f"reboots_sut: {self._reboots_sut}"
         )
 
     @property
@@ -96,6 +101,14 @@ def parallelizable(self) -> bool:
         """
         return self._parallelizable
 
+    @property
+    def reboots_sut(self) -> bool:
+        """
+        :return: If True, test reboots the SUT.
+        :rtype: bool
+        """
+        return self._reboots_sut
+
     @property
     def cwd(self) -> Optional[str]:
         """
@@ -131,7 +144,8 @@ def force_parallel(self) -> None:
         """
         :return: Force test to be parallelizable.
         """
-        self._parallelizable = True
+        if not self._reboots_sut:
+            self._parallelizable = True
 
 
 class Suite:

libkirk/ltp.py

@@ -66,6 +66,7 @@ class LTPFramework(Framework):
             "format_device",
             "save_restore",
             "max_runtime",
+            "reboots_sut",
         }
     )
 
@@ -143,6 +144,7 @@ def __init__(
         self._root = os.environ.get("LTPROOT", "/opt/ltp")
         self._tc_folder = os.path.join(self._root, "testcases", "bin")
         self._env: Dict[str, str] = {}
+        self._metadata: Optional[dict] = None
 
         self._update_env_vars(timeout)
 
@@ -255,6 +257,7 @@ async def _read_runtest(
 
             test_name, test_cmd, *test_args = parts
             parallelizable = False
+            reboots_sut = False
 
             if metadata_tests is not None:
                 test_params = metadata_tests.get(test_name)
@@ -269,6 +272,11 @@ async def _read_runtest(
                         continue
 
                     parallelizable = not (self.PARALLEL_BLACKLIST & test_params.keys())
+                    reboots_sut = bool(
+                        test_params.get("reboots_sut")
+                        and test_params.get("reboots_sut")
+                        not in ("0", 0, "false", "False")
+                    )
 
             self._logger.info(
                 "Test '%s' is%s parallelizable",
@@ -283,6 +291,7 @@ async def _read_runtest(
                 cwd=self._tc_folder,
                 env=env,
                 parallelizable=parallelizable,
+                reboots_sut=reboots_sut,
             )
             tests.append(test)
 
@@ -319,6 +328,23 @@ async def get_suites(self, channel: ComChannel) -> List[str]:
 
         return [line for line in stdout.split("\n") if line]
 
+    async def _get_metadata(self, channel: ComChannel) -> Optional[dict]:
+        """
+        Fetch and parse metadata/ltp.json from SUT, caching the result.
+        """
+        if self._metadata is None:
+            metadata_path = os.path.join(self._root, "metadata", "ltp.json")
+            ret = await channel.run_command(f"test -f {metadata_path}")
+            if ret and ret["returncode"] == 0:
+                try:
+                    self._metadata = json.loads(await channel.fetch_file(metadata_path))
+                except Exception:
+                    self._metadata = {}
+            else:
+                self._metadata = {}
+
+        return self._metadata or None
+
     async def find_command(self, channel: ComChannel, command: str) -> Test:
         if not channel:
             raise ValueError("SUT is None")
@@ -328,19 +354,31 @@ async def find_command(self, channel: ComChannel, command: str) -> Test:
         cmd_args = self._get_cmd_args(command)
         cwd = None
         env = None
+        reboots_sut = False
 
         ret = await channel.run_command(f"test -d {self._tc_folder}")
         if ret and ret["returncode"] == 0:
             cwd = self._tc_folder
             env = await self._read_path(channel)
 
+        metadata_dict = await self._get_metadata(channel)
+        if metadata_dict:
+            tests = metadata_dict.get("tests", {})
+            test_params = tests.get(cmd_args[0])
+            if test_params:
+                reboots_sut = bool(
+                    test_params.get("reboots_sut")
+                    and test_params.get("reboots_sut") not in ("0", 0, "false", "False")
+                )
+
         return Test(
             name=cmd_args[0],
             cmd=cmd_args[0],
             args=cmd_args[1:] or None,
             cwd=cwd,
             env=env,
             parallelizable=False,
+            reboots_sut=reboots_sut,
         )
 
     async def find_suite(self, channel: ComChannel, name: str) -> Suite:
@@ -362,11 +400,7 @@ async def find_suite(self, channel: ComChannel, name: str) -> Suite:
             encoding="utf-8", errors="ignore"
         )
 
-        metadata_dict = None
-        metadata_path = os.path.join(self._root, "metadata", "ltp.json")
-        ret = await channel.run_command(f"test -f {metadata_path}")
-        if ret and ret["returncode"] == 0:
-            metadata_dict = json.loads(await channel.fetch_file(metadata_path))
+        metadata_dict = await self._get_metadata(channel)
 
         return await self._read_runtest(channel, name, runtest_str, metadata_dict)