fix: do not override last deployment in database if it is not in stat…

tdp/cli/commands/plan/dag.py

@@ -83,10 +83,10 @@ def dag(
 ):
     """Deploy from the DAG."""
 
-    from tdp.cli.utils import print_deployment, validate_plan_creation
+    from tdp.cli.utils import print_deployment, validate_clean_last_deployment_state
     from tdp.core.dag import Dag
     from tdp.core.models import DeploymentModel
-    from tdp.core.models.enums import FilterTypeEnum
+    from tdp.core.models.enums import DeploymentStateEnum, FilterTypeEnum
     from tdp.dao import Dao
 
     filter_type = None
@@ -135,7 +135,13 @@ def dag(
         return
     with Dao(db_engine, commit_on_exit=True) as dao:
         if last_deployment := dao.get_last_deployment():
-            validate_plan_creation(last_deployment.state, force)
-            deployment.id = last_deployment.id
+            validated_plan = validate_clean_last_deployment_state(
+                last_deployment.state, force
+            )
+            deployment.id = (
+                last_deployment.id
+                if validated_plan is DeploymentStateEnum.PLANNED
+                else last_deployment.id + 1
+            )
         dao.session.merge(deployment)
     click.echo("Deployment plan successfully created.")

tdp/cli/commands/plan/import_file.py

@@ -31,8 +31,9 @@ def import_file(
 ):
     """Import a deployment from a file."""
 
-    from tdp.cli.utils import parse_file, validate_plan_creation
+    from tdp.cli.utils import parse_file, validate_clean_last_deployment_state
     from tdp.core.models.deployment_model import DeploymentModel
+    from tdp.core.models.enums import DeploymentStateEnum
     from tdp.dao import Dao
 
     with Dao(db_engine, commit_on_exit=True) as dao:
@@ -46,9 +47,15 @@ def import_file(
                 collections, new_operations_hosts_vars
             )
             if last_deployment := dao.get_last_deployment():
-                validate_plan_creation(last_deployment.state, force)
+                validated_plan = validate_clean_last_deployment_state(
+                    last_deployment.state, force
+                )
                 # if a planned deployment is present, update it
-                deployment.id = last_deployment.id
+                deployment.id = (
+                    last_deployment.id
+                    if validated_plan is DeploymentStateEnum.PLANNED
+                    else last_deployment.id + 1
+                )
             dao.session.merge(deployment)
             dao.session.commit()
             click.echo("Deployment plan successfully imported.")

tdp/cli/commands/plan/ops.py

@@ -50,8 +50,9 @@ def ops(
 ):
     """Run a list of operations."""
 
-    from tdp.cli.utils import print_deployment, validate_plan_creation
+    from tdp.cli.utils import print_deployment, validate_clean_last_deployment_state
     from tdp.core.models import DeploymentModel
+    from tdp.core.models.enums import DeploymentStateEnum
     from tdp.dao import Dao
 
     click.echo(
@@ -65,7 +66,13 @@ def ops(
         return
     with Dao(db_engine, commit_on_exit=True) as dao:
         if last_deployment := dao.get_last_deployment():
-            validate_plan_creation(last_deployment.state, force)
-            deployment.id = last_deployment.id
+            validated_plan = validate_clean_last_deployment_state(
+                last_deployment.state, force
+            )
+            deployment.id = (
+                last_deployment.id
+                if validated_plan is DeploymentStateEnum.PLANNED
+                else last_deployment.id + 1
+            )
         dao.session.merge(deployment)
     click.echo("Deployment plan successfully created.")

tdp/cli/commands/plan/reconfigure.py

@@ -36,8 +36,9 @@ def reconfigure(
 ):
     """Reconfigure required TDP services."""
 
-    from tdp.cli.utils import print_deployment, validate_plan_creation
+    from tdp.cli.utils import print_deployment, validate_clean_last_deployment_state
     from tdp.core.models import DeploymentModel
+    from tdp.core.models.enums import DeploymentStateEnum
     from tdp.dao import Dao
 
     click.echo("Creating a deployment plan to reconfigure services.")
@@ -53,7 +54,13 @@ def reconfigure(
             print_deployment(deployment)
             return
         if last_deployment := dao.get_last_deployment():
-            validate_plan_creation(last_deployment.state, force)
-            deployment.id = last_deployment.id
+            validated_plan = validate_clean_last_deployment_state(
+                last_deployment.state, force
+            )
+            deployment.id = (
+                last_deployment.id
+                if validated_plan is DeploymentStateEnum.PLANNED
+                else last_deployment.id + 1
+            )
         dao.session.merge(deployment)
     click.echo("Deployment plan successfully created.")

tdp/cli/commands/plan/resume.py

@@ -28,8 +28,9 @@ def resume(
 ):
     """Resume a failed deployment."""
 
-    from tdp.cli.utils import print_deployment, validate_plan_creation
+    from tdp.cli.utils import print_deployment, validate_clean_last_deployment_state
     from tdp.core.models import DeploymentModel
+    from tdp.core.models.enums import DeploymentStateEnum
     from tdp.dao import Dao
 
     with Dao(db_engine, commit_on_exit=True) as dao:
@@ -50,7 +51,11 @@ def resume(
             print_deployment(deployment)
             return
         if last_deployment := dao.get_last_deployment():
-            validate_plan_creation(last_deployment.state)
-            deployment.id = last_deployment.id
+            validated_plan = validate_clean_last_deployment_state(last_deployment.state)
+            deployment.id = (
+                last_deployment.id
+                if validated_plan is DeploymentStateEnum.PLANNED
+                else last_deployment.id + 1
+            )
         dao.session.merge(deployment)
     click.echo("Deployment plan successfully created.")

tdp/cli/utils.py

@@ -6,7 +6,7 @@
 import re
 from collections.abc import Iterable
 from logging import getLogger
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Literal, Optional
 
 import click
 from tabulate import tabulate
@@ -177,9 +177,13 @@ def parse_file(file_name) -> list[tuple[str, Optional[str], Optional[list[str]]]
     ]
 
 
-def validate_plan_creation(
+def validate_clean_last_deployment_state(
     last_deployment_state: Optional[DeploymentStateEnum], force: Optional[bool] = None
-) -> None:
+) -> Literal[
+    DeploymentStateEnum.SUCCESS,
+    DeploymentStateEnum.FAILURE,
+    DeploymentStateEnum.PLANNED,
+]:
     """Validates that a new deployment plan can be created.
 
     Args:
@@ -199,20 +203,20 @@ def validate_plan_creation(
         DeploymentStateEnum.SUCCESS,
         DeploymentStateEnum.FAILURE,
     ):
-        return
+        return last_deployment_state
 
     # OK if forced
     if force:
         logger.debug("Force option enabled, overriding existing deployment plan.")
-        return
+        return DeploymentStateEnum.PLANNED
 
     # Ask to confirm overriding if last deployment is PLANNED
     if last_deployment_state == DeploymentStateEnum.PLANNED:
         click.confirm(
             "A deployment plan already exists, do you want to override it?",
             abort=True,
         )
-        return
+        return DeploymentStateEnum.PLANNED
 
     # Display an error is last deployment is RUNNING
     if last_deployment_state == DeploymentStateEnum.RUNNING: