def process_outputs()

in critter/stack.py [0:0]
51 lines of code
16 McCabe index (conditional complexity)

    def process_outputs(self):
        # Save stack outputs in an easy access dict
        self.stack_outputs = self.OUTPUTS_DEFAULTS.copy()
        for o in self.stack.outputs:
            self.stack_outputs[o["OutputKey"]] = o["OutputValue"]

        # Sleep for DelayAfterDeploy immediately after loading stack outputs
        self.delay_after_deploy = int(self.stack_outputs[self.OUTPUT_KEYS["DELAY_AFTER_DEPLOY"]])
        if self.delay_after_deploy and self.deploy_action_performed:
            logger.info(f"Sleeping {self.delay_after_deploy} seconds")
            time.sleep(self.delay_after_deploy)

        self.config_rule_name = self.stack_outputs[self.OUTPUT_KEYS["CONFIG_RULE_NAME"]].strip()
        if self.config_rule_name == self.OUTPUTS_DEFAULTS[self.OUTPUT_KEYS["CONFIG_RULE_NAME"]]:
            raise Exception(
                f"Error - Missing required output '{self.OUTPUT_KEYS['CONFIG_RULE_NAME']}' "
                f"on CloudFormation stack '{self.stack_name}'"
            )

        try:
            self.config_rule = self.config.describe_config_rules(ConfigRuleNames=[self.config_rule_name])[
                "ConfigRules"
            ][0]
        except botocore.exceptions.ClientError as e:
            if e.response["Error"]["Code"] == "NoSuchConfigRuleException":
                raise Exception(f"Error - Config rule '{self.config_rule_name}' not found!")
            else:
                raise e

        self.resources = {}
        resource_ids_output_keys = self.EXPECTED_COMPLIANCE_TYPE_LOOKUP.keys()
        for output_key in resource_ids_output_keys:
            resource_ids_csv = self.stack_outputs[output_key]
            if not resource_ids_csv:
                # Empty string means cfn stack did not declare the output
                continue
            for r_id in [i.strip() for i in resource_ids_csv.split(",")]:
                expected_compliance_type = self.EXPECTED_COMPLIANCE_TYPE_LOOKUP[output_key]
                if expected_compliance_type == "NOT_APPLICABLE":
                    logger.warning(
                        "Warning - Testing for NOT_APPLICABLE compliance type is experimental and "
                        "may yield unexpected results."
                    )
                if r_id in self.resources.keys():
                    raise Exception(f"Error - Resource id '{r_id}' declared in multiple outputs")
                self.resources[r_id] = {
                    "expected_compliance_type": expected_compliance_type,
                    "evaluation_result": {},
                }

        if not self.resources:
            raise Exception(
                "Error - Did not find any resource ids outputs. Specify one or more of the following "
                f"CloudFormation stack outputs: {list(resource_ids_output_keys)}"
            )

        # TODO: load resource types from test stack output if not provided in rule scope attribute
        self.resource_types = self.config_rule["Scope"]["ComplianceResourceTypes"]

        self.skip_wait_for_resource_recording = (
            self.stack_outputs[self.OUTPUT_KEYS["SKIP_WAIT_FOR_RESOURCE_RECORDING"]].lower() == "true"
        )