def process_duplicates()

in lemur/certificates/cli.py [0:0]

• 46 lines of code
• 17 McCabe index (conditional complexity)

def process_duplicates(duplicate_candidate_cert, days_since_issuance, skipped_certs, rotation_disabled_certs, processed_unique_cn, commit):
    """
    Process duplicate_candidate_cert to see if there are more certs with exact same details (logic in `is_duplicate()`).
    If Yes, turn off auto


    :param duplicate_candidate_cert: Name of the certificate which has duplicates
    :param days_since_issuance: If not none, include certificates issued in only last days_since_issuance days
    :param skipped_certs: List of certificates which will continue to have rotation on (no change)
    :param rotation_disabled_certs: List of certificates for which rotation got disabled as part of this job
    :param processed_unique_cn: List of unique common names to avoid rework
    :return: Success - True or False; If False, set of duplicates which were not processed
    """
    if duplicate_candidate_cert.cn in processed_unique_cn:
        return True, None

    processed_unique_cn.append(duplicate_candidate_cert.cn)

    certs_with_same_cn = get_certificates_with_same_cn_with_rotate_on(duplicate_candidate_cert.cn,
                                                                      duplicate_candidate_cert.date_created)

    if len(certs_with_same_cn) == 1:
        # this is the only cert with rotation ON, no further action needed
        skipped_certs.append(certs_with_same_cn[0].name)
        metrics.send("disable_rotation_duplicates", "counter", 1,
                     metric_tags={"status": "skipped", "certificate": certs_with_same_cn[0].name}
                     )
        return True, None

    skip_cert = False
    certs_to_stay_on_autorotate = []

    for matching_cert in certs_with_same_cn:
        if matching_cert.name == duplicate_candidate_cert.name:
            # Same cert, no need to compare
            continue

        # Even if one of the certs has different details, skip this set of certs
        # It's safe to do so and this logic can be revisited
        if not is_duplicate(matching_cert, duplicate_candidate_cert):
            skip_cert = True
            break

        # If cert is attached to an endpoint, auto-rotate needs to stay ON
        if matching_cert.endpoints:
            certs_to_stay_on_autorotate.append(matching_cert.name)

    if skip_cert:
        # Not reporting failure for skipping cert since they are not duplicates,
        # comparision is working as intended
        for skipped_cert in certs_with_same_cn:
            skipped_certs.append(skipped_cert.name)
            metrics.send("disable_rotation_duplicates", "counter", 1,
                         metric_tags={"status": "skipped", "certificate": skipped_cert.name}
                         )
        return True, None

    # If no certificate has endpoint, allow autorotaion of only input duplicate_candidate_cert
    if not certs_to_stay_on_autorotate:
        certs_to_stay_on_autorotate.append(duplicate_candidate_cert.name)

    for matching_cert in certs_with_same_cn:
        if matching_cert.name in certs_to_stay_on_autorotate:
            skipped_certs.append(matching_cert.name)
            metrics.send("disable_rotation_duplicates", "counter", 1,
                         metric_tags={"status": "skipped", "certificate": matching_cert.name}
                         )
        else:
            # disable rotation and update DB
            matching_cert.rotation = False
            if commit:
                database.update(matching_cert)
            rotation_disabled_certs.append(matching_cert.name)
            metrics.send("disable_rotation_duplicates", "counter", 1,
                         metric_tags={"status": "success", "certificate": matching_cert.name}
                         )
    return True, None