def run()

in src/alert-manager/config/alert_manager.py [0:0]

• 44 lines of code
• 24 McCabe index (conditional complexity)

    def run(self):
        result = update_nested_dict(self.default_service_conf, self.service_conf)

        # check if email_configs is properly configured
        if result.get("alert-handler") is not None and \
            result["alert-handler"].get("email-configs") is not None and \
                result["alert-handler"]["email-configs"].get("admin-receiver") is not None and \
                result["alert-handler"]["email-configs"].get("smtp-host") is not None and \
                result["alert-handler"]["email-configs"].get("smtp-port") is not None and \
                result["alert-handler"]["email-configs"].get("smtp-from") is not None and \
                result["alert-handler"]["email-configs"].get("smtp-auth-username") is not None and \
                result["alert-handler"]["email-configs"].get("smtp-auth-password") is not None:
            email_configured = True
        else:
            email_configured = False
        
        if email_configured:
            result["alert-handler"]["email-configs"]["templates"] = self.get_email_templates()

        # check if `pai-bearer-token` is properly configured
        if result.get("pai-bearer-token") is not None:
            token_configured = True
        # legacy: to be compatible with pai version <= v1.5
        elif result.get("alert-handler") is not None and \
            result["alert-handler"].get("pai-bearer-token") is not None:
            result["pai-bearer-token"] = result["alert-handler"]["pai-bearer-token"]
            token_configured = True
        else:
            token_configured = False

        result["alert-handler"]["configured"] = True
        if email_configured and token_configured:
            result["actions-available"].extend(["email-admin", "email-user", "stop-jobs", "tag-jobs"])
        elif email_configured:
            result["actions-available"].append("email-admin")
        elif token_configured:
            result["actions-available"].extend(["stop-jobs", "tag-jobs"])

        if result.get("cluster-utilization") is not None and \
            result["cluster-utilization"].get("schedule") is not None and \
            token_configured:
            result["cluster-utilization"]["configured"] = True
        else:
            result["cluster-utilization"]["configured"] = False

        if result.get("job-status-change-notification") is not None and \
            result["job-status-change-notification"].get("enable"):
            result["job-status-change-notification"]["configured"] = True
        else:
            result["job-status-change-notification"]["configured"] = False

        result["host"] = self.get_master_ip()
        result["url"] = "http://{0}:{1}".format(self.get_master_ip(), result["port"])

        return result