Rework architecture of the plugins and add tasks functionality

Added copy and call_command action

Author: stephnxp

socon_embedded/action/__init__.py

@@ -0,0 +1,30 @@
+from typing import Union
+
+from socon.core.manager import Hook
+from pydantic import BaseModel, ConfigDict
+
+from socon_embedded.schema.task import Task
+
+
+class ActionSchema(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+
+class ActionBase(Hook, abstract=True):
+
+    manager = "ActionManager"
+    schema = None
+
+    def __init__(self, task: Task) -> None:
+        self._task = task
+        self.args = self.schema(**self._task.args)
+
+    def run(self, tasks_vars=None):
+        """Run the task"""
+        return {"failed": False}
+
+    def cleanup(self, force: bool = False):
+        """Method to perform a clean up at the end of an action plugin execution
+
+        Action plugins may override this if they deem necessary
+        """

socon_embedded/action/call.py

@@ -0,0 +1,24 @@
+from typing import Optional
+
+from socon.core.management import call_command
+from socon.core.management.base import CommandError
+from socon_embedded.action import ActionBase, ActionSchema
+
+
+class Schema(ActionSchema):
+    cmd: str
+    argv: Optional[list[str]] = []
+
+
+class CallCommandAction(ActionBase):
+    name = "call_command"
+    schema = Schema
+
+    def run(self, tasks_vars=None):
+        result = super().run(tasks_vars)
+        try:
+            call_command(self.args.cmd, *self.args.argv)
+        except CommandError as e:
+            result["failed"] = True
+            result["msg"] = f"Command error: {e}"
+        return result

socon_embedded/action/copy.py

@@ -0,0 +1,86 @@
+import os
+import tempfile
+import json
+import shutil
+
+from pathlib import Path
+from typing import Optional
+
+from socon_embedded.action import ActionBase, ActionSchema
+
+from pydantic import model_validator
+
+from socon_embedded.utils.converter import to_bytes
+
+
+class CopySchema(ActionSchema):
+    dest: str
+    src: Optional[str] = None
+    content: Optional[str] = None
+
+    @model_validator(mode="after")
+    def basic_validation(self):
+        """If src is a directory, dest must be a directory too."""
+        if self.src is None and self.content is None:
+            raise ValueError("'src' or 'content' field must be supply")
+        if self.src is not None and self.content is not None:
+            raise ValueError("Only 'src' or 'content' must be supply at the same time")
+        if self.content is not None and Path(self.dest).is_dir():
+            raise ValueError("Can not use 'content' with a dir as 'dest'")
+        return self
+
+
+class CopyAction(ActionBase):
+    name = "copy"
+    schema = CopySchema
+
+    def run(self, tasks_vars=None):
+        if tasks_vars is None:
+            tasks_vars = dict()
+
+        # Result of the task stored in a dictionary
+        result = super().run(tasks_vars)
+
+        source = self.args.src
+        content = self.args.content
+
+        # Define content_tempfile in case we set it after finding content populated.
+        content_tempfile = None
+
+        if content is not None:
+            try:
+                # If content comes to us as a dict it should be decoded json.
+                # We need to encode it back into a string to write it out.
+                if isinstance(content, dict) or isinstance(content, list):
+                    content_tempfile = self._create_content_tempfile(
+                        json.dumps(content)
+                    )
+                else:
+                    content_tempfile = self._create_content_tempfile(content)
+                source = content_tempfile
+            except Exception as err:
+                result["failed"] = True
+                result["msg"] = f"could not write content temp file: {err}"
+                return result
+
+        source = Path(source).expanduser().absolute()
+        if source.is_dir():
+            shutil.copytree(source, self.args.dest, dirs_exist_ok=True)
+        else:
+            shutil.copy2(source, self.args.dest)
+
+        return result
+
+    def _create_content_tempfile(self, content):
+        """Create a tempfile containing defined content"""
+        fd, content_tempfile = tempfile.mkstemp(prefix=".")
+        f = os.fdopen(fd, "wb")
+        content = to_bytes(content)
+        try:
+            f.write(content)
+        except Exception as err:
+            os.remove(content_tempfile)
+            raise Exception(err)
+        finally:
+            f.close()
+        return content_tempfile

socon_embedded/action/move.py

@@ -0,0 +1,13 @@
+from socon_embedded.action import ActionBase, ActionSchema
+
+
+class MoveActionShema(ActionSchema):
+    pass
+
+
+class MoveAction(ActionBase):
+    name = "move"
+    action_schema = MoveActionShema
+
+    def run(self, tasks_vars=None):
+        pass

socon_embedded/builder/__init__.py

@@ -40,6 +40,7 @@ def __str__(self) -> str:
 @dataclass
 class BuildInfo:
     """Store building information"""
+
     app: str
     builder: str
     project_file: str
@@ -50,7 +51,7 @@ def __str__(self) -> str:
         output = [
             f"Application: {self.app}",
             f"Input file: {self.project_file}",
-            f"Builder: {self.builder}"
+            f"Builder: {self.builder}",
         ]
         variant_args = self.variant_args
         if variant_args:
@@ -60,11 +61,7 @@ def __str__(self) -> str:
 
     def get_case_name(self) -> str:
         """Get the testcase name that will be used in the junit report"""
-        return " - ".join([
-            self.app,
-            self.builder,
-            *self.variant_args.values()
-        ])
+        return " - ".join([self.app, self.builder, *self.variant_args.values()])
 
 
 class Builder(Hook, abstract=True):
@@ -81,7 +78,7 @@ class Builder(Hook, abstract=True):
     def __init__(
         self,
         name: Optional[str] = None,
-        executable: Optional[Union[str, os.PathLike]] = None
+        executable: Optional[Union[str, os.PathLike]] = None,
     ) -> None:
         self.name = name or getattr(self, "name", self.__class__.__name__)
         self.executable = executable or self.get_executable()
@@ -107,7 +104,7 @@ def build(
         warning_as_error: bool = False,
         output_file: Union[str, os.PathLike] = None,
         clean: bool = False,
-        vars: dict = {},
+        variables: dict = {},
         **kwargs: Any,
     ) -> BuildResult:
         """
@@ -149,12 +146,12 @@ def build(
             builder=self.name,
             project_file=project_file,
             cmdline=cmdline,
-            variant_args=variant_args
+            variant_args=variant_args,
         )
         self.display_build_info(buildinfo)
 
         # Run pre_build method if required
-        self.pre_build(buildinfo, **vars)
+        self.pre_build(buildinfo, **variables)
 
         # Get an approximation build time execution
         time_started = time.time()
@@ -164,9 +161,7 @@ def build(
         try:
             status_code, output = self.execute(cmdline, **kwargs)
         except BuildCommandNotFound as e:
-            build_result = BuildResult(
-                Result(Status.FAILURE, str(e)), str(e)
-            )
+            build_result = BuildResult(Result(Status.FAILURE, str(e)), str(e))
         else:
             # Get the execution time of the build
             execution_time = time.time() - time_started
@@ -184,12 +179,12 @@ def build(
 
         # Save the log into the artifact path
         if output_file:
-            with open(output_file, 'w+') as f:
+            with open(output_file, "w+") as f:
                 f.write(build_result.output)
 
         # call the post_build method
         output_dir = Path(output_file).parent
-        self.post_build(build_result, output_dir, **vars)
+        self.post_build(build_result, output_dir, **variables)
 
         return build_result
 
@@ -201,8 +196,7 @@ def get_main_args(
         pass
 
     def execute(
-        self, commands: list[str],
-        **subprocess_args: Any
+        self, commands: list[str], **subprocess_args: Any
     ) -> Tuple[int, Union[str, bytes], str]:
         return self._execute(commands, **subprocess_args)
 
@@ -216,11 +210,11 @@ def display_result(self, build_result: BuildResult) -> None:
             terminal.line(build_result.result.message)
         terminal.line(f"Result: {status_msg}\n", fg=color)
 
-    def pre_build(self, build_info: BuildInfo, **vars):
+    def pre_build(self, build_info: BuildInfo, **variables):
         """Method executed just before the build execution"""
         pass
 
-    def post_build(self, result: BuildResult, **vars):
+    def post_build(self, result: BuildResult, output_dir: str, **variables):
         """Method executed after the build execution"""
         pass
 
@@ -270,7 +264,7 @@ def _execute(
                 stdout=subprocess.PIPE,
                 stderr=subprocess.STDOUT,
                 universal_newlines=True,
-                **subprocess_kwargs
+                **subprocess_kwargs,
             )
 
             if silent is True: