Add b4 Tool

lisa/tools/__init__.py

@@ -16,6 +16,7 @@
 )
 
 from .aria import Aria
+from .b4 import B4
 from .blkid import Blkid
 from .bzip2 import Bzip2
 from .cargo import Cargo
@@ -129,6 +130,7 @@
 __all__ = [
     "AptAddRepository",
     "Aria",
+    "B4",
     "Blkid",
     "Bzip2",
     "Cargo",

lisa/tools/b4.py

@@ -0,0 +1,54 @@
+import pathlib
+import re
+from typing import List, Type
+
+from lisa.executable import Tool
+from lisa.operating_system import Debian
+from lisa.tools.git import Git
+from lisa.tools.python import Pip
+from lisa.util import LisaException, find_group_in_lines
+
+
+class B4(Tool):
+    # Output log is of the form
+    # git am /mnt/code/linux/v2_20241029_xxx_offers.mbx
+    _output_file_pattern = re.compile(
+        r"^.*git.*/(?P<filename>[\w-]+\.mbx).*$", re.MULTILINE
+    )
+
+    @property
+    def command(self) -> str:
+        return "b4"
+
+    @property
+    def dependencies(self) -> List[Type[Tool]]:
+        return [Git]
+
+    @property
+    def can_install(self) -> bool:
+        return True
+
+    def _install(self) -> bool:
+        if isinstance(self.node.os, Debian):
+            self.node.os.install_packages("b4")
+        installed = self._check_exists()
+        if not installed:
+            pip = self.node.tools[Pip]
+            pip.install_packages("b4", install_to_user=True)
+        return self._check_exists()
+
+    def am(
+        self, message_id: str, output_dir: pathlib.PurePath, sudo: bool = False
+    ) -> pathlib.PurePath:
+        result = self.run(
+            f"am -o '{output_dir}' '{message_id}'",
+            force_run=True,
+            expected_exit_code=0,
+            sudo=sudo,
+        )
+        filename = find_group_in_lines(
+            lines=result.stdout, pattern=self._output_file_pattern, single_line=False
+        ).get("filename")
+        if not filename:
+            raise LisaException("Failed to get filename from b4 am output")
+        return pathlib.PurePath(output_dir, filename)

lisa/tools/git.py

@@ -191,14 +191,25 @@ def apply(
         cwd: pathlib.PurePath,
         patches: pathlib.PurePath,
     ) -> None:
-        result = self.run(
-            f"apply {patches}",
-            shell=True,
-            cwd=cwd,
-            force_run=True,
-            no_info_log=True,
-            no_error_log=True,
-        )
+        file_extension = patches.suffix
+        if file_extension == ".mbx":
+            result = self.run(
+                f"am {patches}",
+                shell=True,
+                cwd=cwd,
+                force_run=True,
+                no_info_log=True,
+                no_error_log=True,
+            )
+        else:
+            result = self.run(
+                f"apply {patches}",
+                shell=True,
+                cwd=cwd,
+                force_run=True,
+                no_info_log=True,
+                no_error_log=True,
+            )
         result.assert_exit_code(message=f"failed on applying patches. {result.stdout}")
 
     def list_tags(self, cwd: pathlib.PurePath) -> List[str]:

lisa/transformers/kernel_source_installer.py

@@ -11,7 +11,7 @@
 from lisa.base_tools import Mv
 from lisa.node import Node
 from lisa.operating_system import CBLMariner, Redhat, Ubuntu
-from lisa.tools import Cp, Echo, Git, Make, Sed, Uname
+from lisa.tools import B4, Cp, Echo, Git, Make, Sed, Uname
 from lisa.tools.gcc import Gcc
 from lisa.tools.lscpu import Lscpu
 from lisa.util import LisaException, field_metadata, subclasses
@@ -75,6 +75,12 @@ class PatchModifierSchema(BaseModifierSchema):
     file_pattern: str = "*.patch"
 
 
+@dataclass_json()
+@dataclass
+class B4PatchModifierSchema(BaseModifierSchema):
+    message_id: str = field(default="", metadata=field_metadata(required=True))
+
+
 @dataclass_json()
 @dataclass
 class SourceInstallerSchema(BaseInstallerSchema):
@@ -489,3 +495,24 @@ def _get_code_path(path: str, node: Node, default_name: str) -> PurePath:
         code_path = node.working_path / default_name
 
     return code_path
+
+
+class B4PatchModifier(BaseModifier):
+    @classmethod
+    def type_name(cls) -> str:
+        return "b4_patch"
+
+    @classmethod
+    def type_schema(cls) -> Type[schema.TypedSchema]:
+        return B4PatchModifierSchema
+
+    def modify(self) -> None:
+        runbook: B4PatchModifierSchema = self.runbook
+
+        git = self._node.tools[Git]
+        b4 = self._node.tools[B4]
+
+        message_id = runbook.message_id
+        patch_file = b4.am(message_id=message_id, output_dir=self._code_path)
+
+        git.apply(cwd=self._code_path, patches=patch_file)