def run()

in servicecatalog_puppet/workflow/simulate_policies/do_execute_simulate_policy_task.py [0:0]

• 41 lines of code
• 14 McCabe index (conditional complexity)

    def run(self):
        with self.spoke_regional_client("iam") as iam:
            kwargs = dict(ActionNames=self.action_names)
            if len(self.policy_input_list) > 0:
                kwargs["PolicyInputList"] = self.policy_input_list

            if len(self.permissions_boundary_policy_input_list) > 0:
                kwargs[
                    "PermissionsBoundaryPolicyInputList"
                ] = self.permissions_boundary_policy_input_list

            if len(self.resource_arns) > 0:
                kwargs["ResourceArns"] = self.resource_arns

            if self.resource_policy != "":
                kwargs["ResourcePolicy"] = self.resource_policy

            if self.resource_owner != "":
                kwargs["ResourceOwner"] = self.resource_owner

            if self.caller_arn != "":
                kwargs["CallerArn"] = self.caller_arn

            if len(self.context_entries) > 0:
                kwargs["ContextEntries"] = self.context_entries

            if self.resource_handling_option != "":
                kwargs["ResourceHandlingOption"] = self.resource_handling_option

            if self.simulation_type == "principal":
                kwargs["PolicySourceArn"] = self.policy_source_arn.replace(
                    "${AWS::AccountId}", self.account_id
                )
                result = iam.simulate_principal_policy(**kwargs)
            else:
                if len(self.policy_input_list) == 0:
                    raise Exception(
                        "policy_input_list is required when simulation_type is 'custom'"
                    )
                result = iam.simulate_custom_policy(**kwargs)

            failures = list()
            for evaluation_result in result.get("EvaluationResults"):
                if evaluation_result.get("EvalDecision") != self.expected_decision:
                    failures.append(evaluation_result)

            if len(failures) > 0:
                raise Exception(
                    f"{len(failures)} unexpected decision(s) encountered:\n{yaml.safe_dump(failures)}"
                )

        self.write_output(result)