Provide ltp_skip_test_file var for LTP, set can_install to False for Swap tools (#4287)

Author: paxue

lisa/microsoft/testsuites/ltp/ltp.py

@@ -2,6 +2,7 @@
 # Licensed under the MIT license.
 import os
 import re
+import tempfile
 from dataclasses import dataclass
 from pathlib import PurePath, PurePosixPath
 from typing import Any, Dict, List, Optional, Type
@@ -104,6 +105,7 @@ def run_test(
         ltp_tests: List[str],
         skip_tests: List[str],
         log_path: str,
+        user_input_skip_file: str = "",
         block_device: Optional[str] = None,
         temp_dir: str = "/tmp/",
         ltp_run_timeout: int = 12000,
@@ -147,13 +149,32 @@ def run_test(
         parameters += f"-d {temp_dir} "
 
         # add the list of skip tests to run
-        if len(skip_tests) > 0:
-            # write skip test to skipfile with newline separator
-            skip_file_value = "\n".join(skip_tests)
-            self.node.tools[Echo].write_to_file(
-                skip_file_value, PurePosixPath(skip_file), sudo=True
+        if user_input_skip_file:
+            # copy user provided skip file to remote skip_file location
+            if not os.path.exists(user_input_skip_file):
+                raise LisaException(
+                    f"User provided skip file does not exist: {user_input_skip_file}"
+                )
+            self._log.debug(f"user_input_skip_file: {user_input_skip_file}")
+            self.node.shell.copy(
+                PurePath(user_input_skip_file),
+                PurePosixPath(skip_file),
             )
             parameters += f"-S {skip_file} "
+        elif len(skip_tests) > 0:
+            # Write skip tests to a local temp file and copy to remote,
+            # avoiding shell quoting issues entirely.
+            skip_file_value = "\n".join(skip_tests)
+            with tempfile.NamedTemporaryFile(
+                mode="w", suffix=".skipfile", delete=False
+            ) as tmp:
+                tmp.write(skip_file_value)
+                local_tmp_path = tmp.name
+            try:
+                self.node.shell.copy(PurePath(local_tmp_path), PurePosixPath(skip_file))
+            finally:
+                os.unlink(local_tmp_path)
+            parameters += f"-S {skip_file} "
 
         # Minimum 4M swap space is needed by some mmp test
         if self.node.tools[Free].get_swap_size() < 4:

lisa/microsoft/testsuites/ltp/ltpsuite.py

@@ -118,6 +118,10 @@ def _verify_ltp(
     ) -> None:
         test_list = test_str.split(",") if test_str else []
         skip_test_str = variables.get("ltp_skip_test", "")
+        skip_test_file = variables.get(
+            "ltp_skip_test_file",
+            "",
+        )
         source_file = variables.get("ltp_source_file", "")
         git_tag = variables.get("ltp_tests_git_tag", "")
         run_timeout = int(variables.get("ltp_run_timeout", 12000))
@@ -152,9 +156,10 @@ def _verify_ltp(
         )
         ltp.run_test(
             result,
-            test_list,
-            skip_test_list,
-            log_path,
+            ltp_tests=test_list,
+            skip_tests=skip_test_list,
+            log_path=log_path,
+            user_input_skip_file=skip_test_file,
             block_device=block_device,
             ltp_run_timeout=run_timeout,
         )

lisa/tools/swap.py

@@ -28,18 +28,30 @@ class SwapOn(Tool):
     def command(self) -> str:
         return "swapon"
 
+    @property
+    def can_install(self) -> bool:
+        return False
+
 
 class SwapOff(Tool):
     @property
     def command(self) -> str:
         return "swapoff"
 
+    @property
+    def can_install(self) -> bool:
+        return False
+
 
 class MkSwap(Tool):
     @property
     def command(self) -> str:
         return "mkswap"
 
+    @property
+    def can_install(self) -> bool:
+        return False
+
 
 class Swap(Tool):
     # Filename       Type         Size      Used      Priority
@@ -137,6 +149,10 @@ class SwapInfoBSD(Swap):
     def command(self) -> str:
         return "swapinfo"
 
+    @property
+    def can_install(self) -> bool:
+        return False
+
     def is_swap_enabled(self) -> bool:
         entries = find_patterns_in_lines(self.run("-k").stdout, [self._SWAP_ENTRIES])
         if len(entries[0]) > 0: