def bump_addon_version()

in src/olympia/lib/crypto/tasks.py [0:0]

• 87 lines of code
• 9 McCabe index (conditional complexity)

def bump_addon_version(old_version):
    mail_subject, mail_message = MAIL_COSE_SUBJECT, MAIL_COSE_MESSAGE
    task_user = get_task_user()
    addon = old_version.addon
    old_file_obj = old_version.file
    carryover_groups = [
        promotion
        for promotion in addon.promoted_groups()
        if promotion.listed_pre_review
    ]
    # We only sign files that have been reviewed
    if old_file_obj.status not in amo.APPROVED_STATUSES:
        log.info(
            'Not signing addon {}, version {} (no files)'.format(
                old_version.addon, old_version
            )
        )
        return

    log.info(f'Bumping addon {old_version.addon}, version {old_version}')
    bumped_version_number = get_new_version_number(old_version.version)

    if not old_file_obj.file or not os.path.isfile(old_file_obj.file.path):
        log.info(f'File {old_file_obj.pk} does not exist, skip')
        return

    old_validation = (
        old_file_obj.validation.validation if old_file_obj.has_been_validated else None
    )

    try:
        # Copy the original file to a new FileUpload.
        task_user = get_task_user()
        # last login ip should already be set in the database even on the
        # task user, but in case it's not, like in tests/local envs, set it
        # on the instance, forcing it to be localhost, that should be
        # enough for our needs.
        task_user.last_login_ip = '127.0.0.1'
        original_author = addon.authors.first()
        upload = FileUpload.objects.create(
            addon=addon,
            version=bumped_version_number,
            user=task_user,
            channel=amo.CHANNEL_LISTED,
            source=amo.UPLOAD_SOURCE_GENERATED,
            ip_address=task_user.last_login_ip,
            validation=old_validation,
        )
        upload.name = f'{upload.uuid.hex}_{bumped_version_number}.zip'
        upload.path = upload.generate_path('.zip')
        upload.valid = True
        upload.save()

        # Create the xpi with the bumped version number.
        copy_bumping_version_number(
            old_file_obj.file.path, upload.file_path, bumped_version_number
        )

        # Parse the add-on. We use the original author of the add-on, not
        # the task user, in case they have special permissions allowing
        # the original version to be submitted.
        parsed_data = parse_addon(
            upload, addon=addon, user=original_author, bypass_trademark_checks=True
        )
        parsed_data['approval_notes'] = old_version.approval_notes

        with transaction.atomic():
            # Create a version object out of the FileUpload + parsed data.
            new_version = Version.from_upload(
                upload,
                old_version.addon,
                compatibility=old_version.compatible_apps,
                channel=amo.CHANNEL_LISTED,
                parsed_data=parsed_data,
            )

            # Sign it (may raise SigningError).
            sign_file(new_version.file)

            # Approve it.
            new_version.file.update(
                approval_date=new_version.file.created,
                datestatuschanged=new_version.file.created,
                status=amo.STATUS_APPROVED,
            )

            # Carry over promotion if necessary.
            if carryover_groups:
                addon.approve_for_version(new_version, carryover_groups)

    except Exception:
        log.exception(f'Failed re-signing file {old_file_obj.pk}', exc_info=True)
        # Next loop iteration will clear the task queue.
        return

    # Now notify the developers of that add-on. Any exception should have
    # caused an early return before reaching this point.
    ActivityLog.objects.create(
        amo.LOG.VERSION_RESIGNED,
        addon,
        new_version,
        str(old_version.version),
        user=task_user,
    )
    # Send a mail to the owners warning them we've automatically created and
    # signed a new version of their addon.
    qs = AddonUser.objects.filter(role=amo.AUTHOR_ROLE_OWNER, addon=addon).exclude(
        user__email__isnull=True
    )
    subject = mail_subject.format(addon=addon.name)
    message = mail_message.format(addon=addon.name)
    for email in qs.values_list('user__email', flat=True).order_by('user__pk'):
        amo.utils.send_mail(
            subject,
            message,
            reply_to=['mozilla-add-ons-community@mozilla.com'],
            recipient_list=[email],
        )