Add tests

.gitignore

@@ -1,2 +1,6 @@
 .env
-*/__pycache__
+*/__pycache__
+venv
+*/venv
+.vscode
+*/.vscode

Makefile

@@ -42,4 +42,8 @@ worker-deploy:
 ecr-scan:
 	@echo "make ecr-scan"
 	docker-compose -f docker-compose.yml run --rm ecr-scan
-
+
+test:
+	@echo "Running tests"
+	python -m unittest src/*.py
+

src/codedeploy_client.py

@@ -0,0 +1,68 @@
+import os
+import boto3
+import json
+
+from utils import json_template
+
+
+class CodeDeployClient(object):
+    def __init__(self):
+        self.boto = boto3.client("codedeploy")
+
+    def create_app_spec(
+        self, file_name: str, task_arn: str, capacity_provider_strategy=None
+    ):
+        env_vars = dict(os.environ)
+        env_vars["TASK_ARN"] = task_arn
+        env_vars["CAPACITY_PROVIDER_STRATEGY"] = ""
+        if capacity_provider_strategy:
+            env_vars["CAPACITY_PROVIDER_STRATEGY"] = (
+                ',\\"CapacityProviderStrategy\\":[\\"%s\\"]'
+                % capacity_provider_strategy
+            )
+        try:
+            app_spec_tpl = json_template(file_name, env_vars)
+        except Exception as err:
+            print("Error: Templating app spec :", err)
+            exit(1)
+        print("App spec file content: \n%s" % app_spec_tpl)
+        return json.loads(app_spec_tpl)
+
+    def list_deployments(
+        self, application_name, deployment_group, statuses=["InProgress", "Ready"]
+    ):
+        result = self.boto.list_deployments(
+            applicationName=application_name,
+            deploymentGroupName=deployment_group,
+            includeOnlyStatuses=statuses,
+        )
+        self.deployments = result["deployments"]
+        return result
+
+    def create_deployment(
+        self, application_name, deployment_config_name, deployment_group, revision
+    ):
+        result = self.boto.create_deployment(
+            applicationName=application_name,
+            deploymentGroupName=deployment_group,
+            deploymentConfigName=deployment_config_name,
+            description="Deployment",
+            revision=revision,
+        )
+        self.deploymentId = result["deploymentId"]
+        return result
+
+    def continue_deployment(self, deployment_id):
+        return self.boto.continue_deployment(
+            deploymentId=deployment_id, deploymentWaitType="READY_WAIT"
+        )
+
+    def get_deployment(self, deployment_id):
+        result = self.boto.get_deployment(deploymentId=deployment_id)
+        self.status = result["deploymentInfo"]["status"]
+        return result
+
+    def stop_deployment(self, deployment_id, auto_rollback=True):
+        return self.boto.stop_deployment(
+            deploymentId=deployment_id, autoRollbackEnabled=auto_rollback
+        )

src/codedeploy_client_test.py

@@ -0,0 +1,38 @@
+import unittest
+import json
+import os
+
+from codedeploy_client import CodeDeployClient
+
+
+class CodeDeployClientTest(unittest.TestCase):
+    def setUp(self):
+        os.environ["APP_NAME"] = "test-app"
+        os.environ["CONTAINER_PORT"] = "8080"
+        self.codedeploy_client = CodeDeployClient()
+        # self.deploy = Deploy()
+
+    def test_create_app_spec_should_success_with_capacity_provider_strategy(self):
+        task_def_arn = "arn:aws:1234567890:asdf"
+        capacity_provider_strategy = "asdfasdf"
+        result = self.codedeploy_client.create_app_spec(
+            "../templates/app-spec.tpl.json", task_def_arn, capacity_provider_strategy
+        )
+        with open("./tests/app-spec-with-capacity-provider-strategy.json", "r") as f:
+            expected_json = f.read()
+        expected = json.loads(expected_json)
+        self.assertEqual(result, expected)
+
+    def test_create_app_spec_should_success_without_capacity_provider_strategy(self):
+        task_def_arn = "arn:aws:1234567890:asdf"
+        result = self.codedeploy_client.create_app_spec(
+            "../templates/app-spec.tpl.json", task_def_arn
+        )
+        with open("./tests/app-spec-without-capacity-provider-strategy.json", "r") as f:
+            expected_json = f.read()
+        expected = json.loads(expected_json)
+        self.assertEqual(result, expected)
+
+
+if __name__ == "__main__":
+    unittest.main()