adding unit tests

servicecatalog_puppet/workflow/stack/create_template_for_stack_task.py

@@ -9,7 +9,7 @@
 from servicecatalog_puppet.workflow import tasks
 
 
-class CreateTemplateForStackTask(tasks.PuppetTask):
+class CreateTemplateForStackTask(tasks.PuppetTask): #TODO make task with reference
     def params_for_results_display(self):
         return {
             "cache_invalidator": self.cache_invalidator,

servicecatalog_puppet/workflow/stack/get_cloud_formation_template_from_s3_test.py

@@ -0,0 +1,48 @@
+from unittest import skip
+from servicecatalog_puppet.workflow import tasks_unit_tests_helper
+
+
+class GetCloudFormationTemplateFromS3Test(tasks_unit_tests_helper.PuppetTaskUnitTest):
+    account_id = "account_id"
+    bucket = "bucket"
+    key = "key"
+    region = "region"
+    version_id = "version_id"
+
+    def setUp(self) -> None:
+        from servicecatalog_puppet.workflow.stack import get_cloud_formation_template_from_s3
+        self.module = get_cloud_formation_template_from_s3
+
+        self.sut = self.module.GetCloudFormationTemplateFromS3(
+            **self.get_common_args(),
+            account_id=self.account_id, bucket=self.bucket, key=self.key, region=self.region, version_id=self.version_id        
+        )
+
+        self.wire_up_mocks()    
+
+    def test_params_for_results_display(self):
+        # setup
+        expected_result = {
+            "task_reference": self.task_reference,
+            "bucket": self.bucket,
+            "key": self.key.replace("-${AWS::Region}", f"-{self.region}"),
+            "region": self.region,
+            "version_id": self.version_id,
+            "cache_invalidator": self.cache_invalidator,
+        }        
+
+        # exercise
+        actual_result = self.sut.params_for_results_display()
+
+        # verify
+        self.assertEqual(expected_result, actual_result)
+
+    @skip
+    def test_run(self):
+        # setup
+        # exercise
+        actual_result = self.sut.run()
+
+        # verify
+        raise NotImplementedError()
+

servicecatalog_puppet/workflow/stack/prepare_account_for_stack_task.py

@@ -8,7 +8,7 @@
 from servicecatalog_puppet.workflow.stack import create_template_for_stack_task
 
 
-class PrepareAccountForWorkspaceTask(tasks.PuppetTask):
+class PrepareAccountForWorkspaceTask(tasks.PuppetTask): #TODO make task with reference
     account_id = luigi.Parameter()
 
     def params_for_results_display(self):

servicecatalog_puppet/workflow/stack/provision_stack_dry_run_task_test.py

@@ -0,0 +1,63 @@
+from unittest import skip
+from servicecatalog_puppet.workflow import tasks_unit_tests_helper
+
+
+class ProvisionStackDryRunTaskTest(tasks_unit_tests_helper.PuppetTaskUnitTest):
+    stack_name = "stack_name"
+    region = "region"
+    account_id = "account_id"
+    bucket = "bucket"
+    key = "key"
+    version_id = "version_id"
+    launch_name = "launch_name"
+    stack_set_name = "stack_set_name"
+    get_s3_template_ref = "get_s3_template_ref"
+    capabilities = []
+    ssm_param_inputs = []
+    launch_parameters = {}
+    manifest_parameters = {}
+    account_parameters = {}
+    ssm_param_outputs = []
+    execution = "execution"
+    manifest_file_path = "manifest_file_path"
+
+    def setUp(self) -> None:
+        from servicecatalog_puppet.workflow.stack import provision_stack_dry_run_task
+        self.module = provision_stack_dry_run_task
+
+        self.sut = self.module.ProvisionStackDryRunTask(
+            **self.get_common_args(),
+            stack_name=self.stack_name, region=self.region, account_id=self.account_id, bucket=self.bucket,
+            key=self.key, version_id=self.version_id, launch_name=self.launch_name, stack_set_name=self.stack_set_name,
+            get_s3_template_ref=self.get_s3_template_ref, capabilities=self.capabilities,
+            ssm_param_inputs=self.ssm_param_inputs, launch_parameters=self.launch_parameters,
+            manifest_parameters=self.manifest_parameters, account_parameters=self.account_parameters,
+            ssm_param_outputs=self.ssm_param_outputs, execution=self.execution,
+            manifest_file_path=self.manifest_file_path
+
+        )
+
+        self.wire_up_mocks()    
+
+    def test_api_calls_used(self):
+        # setup
+        expected_result = [
+            f"cloudformation.get_template_summary_{self.account_id}_{self.region}",
+            f"cloudformation.describe_stacks_{self.account_id}_{self.region}",
+        ]        
+
+        # exercise
+        actual_result = self.sut.api_calls_used()
+
+        # verify
+        self.assertEqual(expected_result, actual_result)
+
+    @skip
+    def test_run(self):
+        # setup
+        # exercise
+        actual_result = self.sut.run()
+
+        # verify
+        raise NotImplementedError()
+

servicecatalog_puppet/workflow/stack/provision_stack_task.py

@@ -16,9 +16,7 @@
 
 
 class ProvisionStackTask(tasks.TaskWithParameters):
-
     stack_name = luigi.Parameter()
-    puppet_account_id = luigi.Parameter()
 
     region = luigi.Parameter()
     account_id = luigi.Parameter()

servicecatalog_puppet/workflow/stack/provision_stack_task_test.py

@@ -0,0 +1,59 @@
+from unittest import skip
+from servicecatalog_puppet.workflow import tasks_unit_tests_helper
+
+
+class ProvisionStackTaskTest(tasks_unit_tests_helper.PuppetTaskUnitTest):
+    stack_name = "stack_name"
+    region = "region"
+    account_id = "account_id"
+    bucket = "bucket"
+    key = "key"
+    version_id = "version_id"
+    launch_name = "launch_name"
+    stack_set_name = "stack_set_name"
+    get_s3_template_ref = "get_s3_template_ref"
+    capabilities = []
+    ssm_param_inputs = []
+    launch_parameters = {}
+    manifest_parameters = {}
+    account_parameters = {}
+    ssm_param_outputs = []
+    execution = "execution"
+    manifest_file_path = "manifest_file_path"
+
+    def setUp(self) -> None:
+        from servicecatalog_puppet.workflow.stack import provision_stack_task
+        self.module = provision_stack_task
+
+        self.sut = self.module.ProvisionStackTask(
+            **self.get_common_args(),
+            stack_name=self.stack_name, region=self.region, account_id=self.account_id, bucket=self.bucket, key=self.key, version_id=self.version_id, launch_name=self.launch_name, stack_set_name=self.stack_set_name, get_s3_template_ref=self.get_s3_template_ref, capabilities=self.capabilities, ssm_param_inputs=self.ssm_param_inputs, launch_parameters=self.launch_parameters, manifest_parameters=self.manifest_parameters, account_parameters=self.account_parameters, ssm_param_outputs=self.ssm_param_outputs, execution=self.execution, manifest_file_path=self.manifest_file_path        
+        )
+
+        self.wire_up_mocks()    
+
+    def test_params_for_results_display(self):
+        # setup
+        expected_result = {
+            "puppet_account_id": self.puppet_account_id,
+            "stack_name": self.stack_name,
+            "account_id": self.account_id,
+            "region": self.region,
+            "cache_invalidator": self.cache_invalidator,
+        }        
+
+        # exercise
+        actual_result = self.sut.params_for_results_display()
+
+        # verify
+        self.assertEqual(expected_result, actual_result)
+
+    @skip
+    def test_run(self):
+        # setup
+        # exercise
+        actual_result = self.sut.run()
+
+        # verify
+        raise NotImplementedError()
+

servicecatalog_puppet/workflow/stack/terminate_stack_task.py

@@ -11,9 +11,7 @@
 
 
 class TerminateStackTask(tasks.TaskWithReference):
-
     stack_name = luigi.Parameter()
-    puppet_account_id = luigi.Parameter()
 
     region = luigi.Parameter()
     account_id = luigi.Parameter()

servicecatalog_puppet/workflow/stack/terminate_stack_task_test.py

@@ -0,0 +1,57 @@
+from unittest import skip
+from servicecatalog_puppet.workflow import tasks_unit_tests_helper
+
+
+class TerminateStackTaskTest(tasks_unit_tests_helper.PuppetTaskUnitTest):
+    stack_name = "stack_name"
+    region = "region"
+    account_id = "account_id"
+    bucket = "bucket"
+    key = "key"
+    version_id = "version_id"
+    launch_name = "launch_name"
+    stack_set_name = "stack_set_name"
+    capabilities = []
+    ssm_param_inputs = []
+    launch_parameters = {}
+    manifest_parameters = {}
+    account_parameters = {}
+    ssm_param_outputs = []
+    execution = "execution"
+
+    def setUp(self) -> None:
+        from servicecatalog_puppet.workflow.stack import terminate_stack_task
+        self.module = terminate_stack_task
+
+        self.sut = self.module.TerminateStackTask(
+            **self.get_common_args(),
+            stack_name=self.stack_name, region=self.region, account_id=self.account_id, bucket=self.bucket, key=self.key, version_id=self.version_id, launch_name=self.launch_name, stack_set_name=self.stack_set_name, capabilities=self.capabilities, ssm_param_inputs=self.ssm_param_inputs, launch_parameters=self.launch_parameters, manifest_parameters=self.manifest_parameters, account_parameters=self.account_parameters, ssm_param_outputs=self.ssm_param_outputs, execution=self.execution        
+        )
+
+        self.wire_up_mocks()    
+
+    def test_params_for_results_display(self):
+        # setup
+        expected_result = {
+            "puppet_account_id": self.puppet_account_id,
+            "stack_name": self.stack_name,
+            "account_id": self.account_id,
+            "region": self.region,
+            "cache_invalidator": self.cache_invalidator,
+        }        
+
+        # exercise
+        actual_result = self.sut.params_for_results_display()
+
+        # verify
+        self.assertEqual(expected_result, actual_result)
+
+    @skip
+    def test_run(self):
+        # setup
+        # exercise
+        actual_result = self.sut.run()
+
+        # verify
+        raise NotImplementedError()
+

servicecatalog_puppet/workflow/tasks_unit_tests_helper.py

@@ -26,6 +26,7 @@ def set_value(self, name, value):
 
 
 class PuppetTaskUnitTest(unittest.TestCase):
+    task_reference = "task_reference"
     puppet_account_id = "puppet_account_id"
 
     def get_common_args(self):

test_writer.py

@@ -265,7 +265,7 @@ def get_initial_args_for(c):
     return dict()
 
 
-package = "tag_policies"
+package = "stack"
 
 for input in glob.glob("servicecatalog_puppet/workflow/**/*.py", recursive=True):
     print(input)