🎯 feat: add dryrun mode to stage exec.

Author: korawica

src/ddeutil/workflow/stages.py

@@ -610,12 +610,31 @@ def dryrun(
         *,
         parent_run_id: Optional[str] = None,
         event: Optional[Event] = None,
-    ) -> Optional[Result]:  # pragma: no cov
+    ) -> Optional[Result]:
         """Pre-process method that will use to run with dry-run mode, and it
         should be used before process method.
+
+        Args:
+            params (DictData): A parameter data that want to use in this
+                execution.
+            run_id (str): A running stage ID.
+            context (DictData): A context data.
+            parent_run_id: A parent running ID. (Default is None)
+            event: An event manager that use to track parent process
+                was not force stopped.
+
+        Returns:
+            Result: The execution result with status and context data.
         """
+        trace: Trace = get_trace(
+            run_id, parent_run_id=parent_run_id, extras=self.extras
+        )
+        trace.debug("[STAGE]: Start Dryrun ...")
+        return self.to_empty().process(
+            params, run_id, context, parent_run_id=parent_run_id, event=event
+        )
 
-    def to_empty(self, sleep: int = 0.35) -> EmptyStage:  # pragma: no cov
+    def to_empty(self, sleep: int = 0.35) -> EmptyStage:
         """Convert the current Stage model to the EmptyStage model for dry-run
         mode if the `action_stage` class attribute has set.
 
@@ -627,14 +646,14 @@ def to_empty(self, sleep: int = 0.35) -> EmptyStage:  # pragma: no cov
                 message.
         """
         if isinstance(self, EmptyStage):
-            return self
+            return self.model_copy(update={"sleep": sleep})
         return EmptyStage.model_validate(
             {
                 "name": self.name,
                 "id": self.id,
                 "desc": self.desc,
                 "if": self.condition,
-                "echo": f"Convert from {self.__class__.__name__}",
+                "echo": f"Convert from {self.__class__.__name__} to EmptyStage",
                 "sleep": sleep,
             }
         )
@@ -852,18 +871,20 @@ def _execute(
         trace: Trace = get_trace(
             run_id, parent_run_id=parent_run_id, extras=self.extras
         )
-        model: Union[Self, EmptyStage] = (
-            self.to_empty()
+        # NOTE: First execution for not pass to retry step if it passes.
+        try:
             if (
                 self.extras.get("__sys_release_dryrun_mode", False)
                 and self.action_stage
-            )
-            else self
-        )
-
-        # NOTE: First execution for not pass to retry step if it passes.
-        try:
-            return model.process(
+            ):
+                return self.dryrun(
+                    params | {"retry": current_retry},
+                    run_id=run_id,
+                    context=context,
+                    parent_run_id=parent_run_id,
+                    event=event,
+                )
+            return self.process(
                 params | {"retry": current_retry},
                 run_id=run_id,
                 context=context,
@@ -889,7 +910,18 @@ def _execute(
                     status=WAIT,
                     updated={"retry": current_retry},
                 )
-                return model.process(
+                if (
+                    self.extras.get("__sys_release_dryrun_mode", False)
+                    and self.action_stage
+                ):
+                    return self.dryrun(
+                        params | {"retry": current_retry},
+                        run_id=run_id,
+                        context=context,
+                        parent_run_id=parent_run_id,
+                        event=event,
+                    )
+                return self.process(
                     params | {"retry": current_retry},
                     run_id=run_id,
                     context=context,
@@ -953,6 +985,17 @@ async def _axecute(
 
         # NOTE: First execution for not pass to retry step if it passes.
         try:
+            if (
+                self.extras.get("__sys_release_dryrun_mode", False)
+                and self.action_stage
+            ):
+                return self.dryrun(
+                    params | {"retry": current_retry},
+                    run_id=run_id,
+                    context=context,
+                    parent_run_id=parent_run_id,
+                    event=event,
+                )
             return await model.async_process(
                 params | {"retry": current_retry},
                 run_id=run_id,
@@ -979,6 +1022,17 @@ async def _axecute(
                     status=WAIT,
                     updated={"retry": current_retry},
                 )
+                if (
+                    self.extras.get("__sys_release_dryrun_mode", False)
+                    and self.action_stage
+                ):
+                    return self.dryrun(
+                        params | {"retry": current_retry},
+                        run_id=run_id,
+                        context=context,
+                        parent_run_id=parent_run_id,
+                        event=event,
+                    )
                 return await model.async_process(
                     params | {"retry": current_retry},
                     run_id=run_id,

tests/stages/test_stage_call.py

@@ -80,6 +80,19 @@ def test_call_stage_exec_necessary_args():
     }
 
 
+def test_call_stage():
+    stage: Stage = CallStage.model_validate(
+        obj={
+            "name": "Extract & Load Local System",
+            "id": "second-job",
+            "uses": "tasks/simple-task@demo",
+            "with": {"source": "src", "sink": "sink"},
+        }
+    )
+    empty_stage = stage.to_empty()
+    assert empty_stage.sleep == 0.35
+
+
 def test_call_stage_exec():
     stage: Stage = CallStage.model_validate(
         obj={

tests/stages/test_stage_empty.py

@@ -33,6 +33,9 @@ def test_empty_stage():
     )
     assert stage.desc == "This is a test stage\n\tnewline"
 
+    empty_stage = stage.to_empty(sleep=10)
+    assert empty_stage.sleep == 10
+
 
 def test_empty_stage_execute():
     stage: EmptyStage = EmptyStage(name="Empty Stage", echo="hello world")

tests/test_workflow_release.py

@@ -257,8 +257,25 @@ def test_workflow_release_dryrun():
         }
     )
     rs: Result = workflow.release(
-        release=datetime.now(),
+        release=datetime(2024, 10, 1),
         params={},
         release_type=DRYRUN,
     )
-    print(rs)
+    assert rs.status == SUCCESS
+    assert rs.context == {
+        "status": "SUCCESS",
+        "params": {},
+        "release": {
+            "type": DRYRUN,
+            "logical_date": datetime(2024, 10, 1, tzinfo=ZoneInfo(key="UTC")),
+        },
+        "jobs": {
+            "first-job": {
+                "status": SUCCESS,
+                "stages": {
+                    "7782830343": {"outputs": {}, "status": SUCCESS},
+                    "second-stage": {"outputs": {}, "status": SUCCESS},
+                },
+            },
+        },
+    }