fixup! feat(action): migrate steps from yaml to python

.github/workflows/ci.yml

@@ -13,4 +13,4 @@ jobs:
 
       - uses: moneymeets/moneymeets-composite-actions/lint-python@master
 
-      - run: poetry run python -m pytest --cov --cov-fail-under=89
+      - run: poetry run python -m pytest --cov --cov-fail-under=100

action.yml

@@ -55,9 +55,8 @@ runs:
         # ToDo: Re-enable cache when https://github.com/actions/setup-python/issues/361 is fixed
         poetry_cache_enabled: 'false'
 
-    - name: AWS ECS Task Deploy
+    - id: run-ecs-deploy
       shell: bash
-      id: run-ecs-deploy
       working-directory: ${{ github.action_path }}
       run: |
         if [[ "${{ inputs.allow_feature_branch_deployment }}" == "true" ]]; then

actions_helper/commands/create_task_definition.py

@@ -5,9 +5,7 @@
 
 from actions_helper.commands.get_active_task_definition_by_tag import get_active_task_definition_arn_by_tag
 from actions_helper.outputs import CreateTaskDefinitionOutput
-from actions_helper.utils import set_error
-
-PLACEHOLDER_TEXT = "PLACEHOLDER"
+from actions_helper.utils import PLACEHOLDER_TEXT, set_error
 
 KEYS_TO_DELETE_FROM_TASK_DEFINITION = [
     "taskDefinitionArn",

actions_helper/commands/deregister_task_definition.py

@@ -11,9 +11,9 @@ def deregister_task_definition(
     ecs_client: BaseClient,
     cluster: str,
     service: str,
-    production_task_definition_output: Optional[CreateTaskDefinitionOutput] = None,
-    local_task_definition_output: Optional[CreateTaskDefinitionOutput] = None,
-    preflight_task_definition_output: Optional[CreateTaskDefinitionOutput] = None,
+    production_task_definition_output: Optional[CreateTaskDefinitionOutput],
+    local_task_definition_output: Optional[CreateTaskDefinitionOutput],
+    preflight_task_definition_output: Optional[CreateTaskDefinitionOutput],
 ):
     (primary_deployment_definition_arn,) = (
         deployment["taskDefinition"]
@@ -26,35 +26,27 @@ def deregister_task_definition(
 
     click.echo(f"{primary_deployment_definition_arn=}")
 
-    production_task_definition_to_deregister = local_task_definition_to_deregister = (
-        preflight_task_definition_to_deregister
-    ) = None
-
     fail_pipeline = True
     if (
-        production_task_definition_output
+        all((production_task_definition_output, local_task_definition_output, preflight_task_definition_output))
         and primary_deployment_definition_arn == production_task_definition_output.latest_task_definition_arn
     ):
         # Do not deregister task definition for initial deployment
-        if production_task_definition_output.previous_task_definition_arn:
-            production_task_definition_to_deregister = production_task_definition_output.previous_task_definition_arn
-            local_task_definition_to_deregister = local_task_definition_output.previous_task_definition_arn
-            preflight_task_definition_to_deregister = preflight_task_definition_output.previous_task_definition_arn
+        if not production_task_definition_output.previous_task_definition_arn:
+            return
         fail_pipeline = False
-    else:
-        production_task_definition_to_deregister = production_task_definition_output.latest_task_definition_arn
-        local_task_definition_to_deregister = local_task_definition_output.latest_task_definition_arn
-        preflight_task_definition_to_deregister = preflight_task_definition_output.latest_task_definition_arn
 
     for task_definition in (
-        production_task_definition_to_deregister,
-        local_task_definition_to_deregister,
-        preflight_task_definition_to_deregister,
+        production_task_definition_output,
+        local_task_definition_output,
+        preflight_task_definition_output,
     ):
         if task_definition:
             click.echo(f"Deregister {task_definition}")
             ecs_client.deregister_task_definition(
-                taskDefinition=task_definition,
+                taskDefinition=task_definition.latest_task_definition_arn
+                if fail_pipeline
+                else task_definition.previous_task_definition_arn,
             )
 
     if fail_pipeline:

actions_helper/main.py

@@ -10,7 +10,7 @@
 
 @click.group()
 def cli():
-    pass
+    pass  # pragma: no cover
 
 
 @cli.command(
@@ -40,6 +40,7 @@ def cmd_ecs_deploy(
 
     ecs_client = boto3.Session(region_name=aws_region).client("ecs")
     ecr_client = boto3.Session(region_name=aws_region).client("ecr")
+    service = f"{ecr_repository}-{environment}"
     production_task_definition = local_task_definition = preflight_output = None
     try:
         click.echo("Getting docker image URI...")
@@ -56,7 +57,7 @@ def cmd_ecs_deploy(
         click.echo("Creating production task definition...")
         production_task_definition = create_task_definition(
             ecs_client=ecs_client,
-            application_id=f"{ecr_repository}-{environment}",
+            application_id=service,
             deployment_tag=deployment_tag,
             image_uri=image_uri,
         )
@@ -65,7 +66,7 @@ def cmd_ecs_deploy(
             click.echo("Run preflight enabled")
             preflight_output = run_preflight_container(
                 ecs_client=ecs_client,
-                service=f"{ecr_repository}-{environment}",
+                service=service,
                 cluster=environment,
                 application_id=f"{ecr_repository}-preflight-{environment}",
                 image_uri=image_uri,
@@ -77,32 +78,28 @@ def cmd_ecs_deploy(
             taskDefinition=production_task_definition.latest_task_definition_arn,
             desiredCount=desired_count,
             cluster=environment,
-            service=f"{ecr_repository}-{environment}",
+            service=service,
         )
-
         click.echo("Service updated")
 
         click.echo("Waiting for service stability...")
         wait_for_service_stability(
             ecs_client=ecs_client,
             cluster=environment,
-            service=f"{ecr_repository}-{environment}",
+            service=service,
         )
-
-    except Exception as e:
-        click.echo("An exception occurred", err=True)
-        click.echo(e)
+        click.echo("Service stable")
     finally:
         click.echo("De-registering task definition")
         deregister_task_definition(
             ecs_client=ecs_client,
             cluster=environment,
-            service=f"{ecr_repository}-{environment}",
+            service=service,
             production_task_definition_output=production_task_definition,
             local_task_definition_output=local_task_definition,
             preflight_task_definition_output=preflight_output,
         )
 
 
-if __name__ == "__main__":
+if __name__ == "__main__":  # pragma: no cover
     cli()

actions_helper/utils.py

@@ -1,5 +1,7 @@
 from typing import Optional
 
+PLACEHOLDER_TEXT = "PLACEHOLDER"
+
 
 def set_error(message: str, file: Optional[str] = None, line: Optional[str] = None):
     print(f"::error {f'file={file}' if file else ''}{f',line={line}' if line else ''}::{message}")