Implement devhub UI to trigger version rollback

src/olympia/abuse/migrations/0060_auto_20250710_0958.py

@@ -0,0 +1,16 @@
+# Generated by Django 4.2.23 on 2025-07-10 09:58
+
+from django.db import migrations
+
+from olympia.core.db.migrations import CreateWaffleSwitch
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('abuse', '0059_auto_20250526_1436'),
+    ]
+
+    operations = [
+        CreateWaffleSwitch(name='version-rollback')
+    ]

src/olympia/addons/models.py

@@ -27,6 +27,7 @@
 from django.utils.functional import cached_property
 from django.utils.translation import gettext_lazy as _, trans_real
 
+import waffle
 from django_statsd.clients import statsd
 
 import olympia.core.logger
@@ -2021,6 +2022,17 @@ def update_all_due_dates(self):
         ):
             version.reset_due_date(should_have_due_date=False)
 
+    def rollbackable_versions_qs(self, channel):
+        # Needs to be an extension
+        if not self.type == amo.ADDON_EXTENSION or not waffle.switch_is_active(
+            'version-rollback'
+        ):
+            return Version.objects.none()
+        qs = self.versions.filter(channel=channel, file__status=amo.STATUS_APPROVED)
+        # You can't rollback to the latest approved version
+        qs = qs.exclude(id=qs.values('id')[:1])
+        return qs.order_by('-created')
+
 
 dbsignals.pre_save.connect(save_signal, sender=Addon, dispatch_uid='addon_translations')
 

src/olympia/addons/serializers.py

@@ -57,6 +57,10 @@
     Version,
     VersionPreview,
 )
+from olympia.versions.utils import (
+    validate_version_number_does_not_exist,
+    validate_version_number_is_gt_latest_signed_listed_version,
+)
 
 from .fields import (
     CategoriesSerializerField,
@@ -79,11 +83,7 @@
     ReplacementAddon,
 )
 from .tasks import resize_icon, resize_preview
-from .utils import (
-    fetch_translations_from_addon,
-    get_translation_differences,
-    validate_version_number_is_gt_latest_signed_listed_version,
-)
+from .utils import fetch_translations_from_addon, get_translation_differences
 from .validators import (
     AddonDefaultLocaleValidator,
     AddonMetadataValidator,
@@ -545,25 +545,14 @@ def validate_is_disabled(self, disable):
                 raise exceptions.ValidationError(msg)
         return disable
 
-    def _check_for_existing_versions(self, version_string):
-        # Make sure we don't already have this version.
-        existing_versions = Version.unfiltered.filter(
-            addon=self.addon, version=version_string
-        )
-        if existing_versions.exists():
-            if existing_versions[0].deleted:
-                msg = gettext(
-                    'Version {version_string} was uploaded before and deleted.'
-                )
-            else:
-                msg = gettext('Version {version_string} already exists.')
-            raise Conflict({'version': [msg.format(version_string=version_string)]})
-
     def validate(self, data):
         if not self.instance:
             version_string = self.parsed_data.get('version')
             if self.addon:
-                self._check_for_existing_versions(version_string)
+                if error_msg := (
+                    validate_version_number_does_not_exist(self.addon, version_string)
+                ):
+                    raise Conflict({'version': [error_msg]})
 
                 if data['upload'].channel == amo.CHANNEL_LISTED:
                     if error_message := (

src/olympia/addons/tests/test_models.py

@@ -10,6 +10,7 @@
 from django.utils import translation
 
 import pytest
+from waffle.testutils import override_switch
 
 from olympia import amo, core
 from olympia.activity.models import ActivityLog, AddonLog
@@ -1993,6 +1994,65 @@ def test_all_approved_applications_for_group_addition_after_approval(self):
             PromotedGroup.objects.get(group_id=PROMOTED_GROUP_CHOICES.LINE)
         ) == [amo.FIREFOX]
 
+    def test_rollbackable_versions_qs_unavailable(self):
+        def get_rvs(channel):
+            return list(addon.rollbackable_versions_qs(channel=channel))
+
+        addon = addon_factory()
+        version_factory(addon=addon)
+        version_factory(addon=addon, channel=amo.CHANNEL_UNLISTED)
+        version_factory(addon=addon, channel=amo.CHANNEL_UNLISTED)
+
+        with override_switch('version-rollback', active=False):
+            assert get_rvs(amo.CHANNEL_LISTED) == []
+            assert get_rvs(amo.CHANNEL_UNLISTED) == []
+
+        with override_switch('version-rollback', active=True):
+            assert get_rvs(amo.CHANNEL_LISTED) != []
+            assert get_rvs(amo.CHANNEL_UNLISTED) != []
+
+            addon.update(type=amo.ADDON_STATICTHEME)
+            assert get_rvs(amo.CHANNEL_LISTED) == []
+            assert get_rvs(amo.CHANNEL_UNLISTED) == []
+
+    @override_switch('version-rollback', active=True)
+    def test_rollbackable_versions_qs(self):
+        def get_rvs(channel):
+            return list(addon.rollbackable_versions_qs(channel=channel))
+
+        addon = addon_factory()
+        assert get_rvs(amo.CHANNEL_LISTED) == []
+        assert get_rvs(amo.CHANNEL_UNLISTED) == []
+
+        version1 = addon.current_version
+        version2 = version_factory(addon=addon)
+        version3 = version_factory(addon=addon)
+        assert version3 == addon.reload().current_version
+        assert get_rvs(amo.CHANNEL_LISTED) == [version2, version1]
+        assert get_rvs(amo.CHANNEL_UNLISTED) == []
+
+        version2.file.update(status=amo.STATUS_DISABLED)
+        assert get_rvs(amo.CHANNEL_LISTED) == [version1]
+        assert get_rvs(amo.CHANNEL_UNLISTED) == []
+
+        version1.file.update(status=amo.STATUS_DISABLED)
+        assert get_rvs(amo.CHANNEL_LISTED) == []
+        assert get_rvs(amo.CHANNEL_UNLISTED) == []
+
+        self.make_addon_unlisted(addon)
+        File.objects.filter(version__addon=addon).update(status=amo.STATUS_APPROVED)
+        assert addon.reload().current_version is None
+        assert get_rvs(amo.CHANNEL_LISTED) == []
+        assert get_rvs(amo.CHANNEL_UNLISTED) == [version2, version1]
+
+        version2.file.update(status=amo.STATUS_DISABLED)
+        assert get_rvs(amo.CHANNEL_LISTED) == []
+        assert get_rvs(amo.CHANNEL_UNLISTED) == [version1]
+
+        version1.file.update(status=amo.STATUS_DISABLED)
+        assert get_rvs(amo.CHANNEL_LISTED) == []
+        assert get_rvs(amo.CHANNEL_UNLISTED) == []
+
 
 class TestAddonUser(TestCase):
     def test_delete(self):

src/olympia/addons/tests/test_utils_.py

@@ -5,16 +5,14 @@
 import pytest
 from freezegun import freeze_time
 
-from olympia import amo
-from olympia.amo.tests import TestCase, addon_factory, user_factory, version_factory
+from olympia.amo.tests import TestCase, addon_factory, user_factory
 from olympia.users.models import Group, GroupUser
 
 from ..utils import (
     RECOMMENDATIONS,
     DeleteTokenSigner,
     get_addon_recommendations,
     get_filtered_fallbacks,
-    validate_version_number_is_gt_latest_signed_listed_version,
     verify_mozilla_trademark,
 )
 
@@ -106,65 +104,3 @@ def test_delete_token_signer(frozen_time=None):
     # but not after 60 seconds
     frozen_time.tick(timedelta(seconds=2))
     assert not signer.validate(token, addon_id)
-
-
-@pytest.mark.django_db
-def test_validate_version_number_is_gt_latest_signed_listed_version():
-    addon = addon_factory(version_kw={'version': '123.0'}, file_kw={'is_signed': True})
-    # add an unlisted version, which should be ignored.
-    latest_unlisted = version_factory(
-        addon=addon,
-        version='124',
-        channel=amo.CHANNEL_UNLISTED,
-        file_kw={'is_signed': True},
-    )
-    # Version number is greater, but doesn't matter, because the check is listed only.
-    assert latest_unlisted.version > addon.current_version.version
-
-    # version number isn't greater (its the same).
-    assert validate_version_number_is_gt_latest_signed_listed_version(addon, '123') == (
-        'Version 123 must be greater than the previous approved version 123.0.'
-    )
-    # version number is less than the current listed version.
-    assert validate_version_number_is_gt_latest_signed_listed_version(
-        addon, '122.9'
-    ) == ('Version 122.9 must be greater than the previous approved version 123.0.')
-    # version number is greater, so no error message.
-    assert not validate_version_number_is_gt_latest_signed_listed_version(
-        addon, '123.1'
-    )
-
-    addon.current_version.file.update(is_signed=False)
-    # Same as current but check only applies to signed versions, so no error message.
-    assert not validate_version_number_is_gt_latest_signed_listed_version(addon, '123')
-
-    # Set up the scenario when a newer version has been signed, but then disabled
-    addon.current_version.file.update(is_signed=True)
-    disabled = version_factory(
-        addon=addon,
-        version='123.5',
-        file_kw={'is_signed': True, 'status': amo.STATUS_DISABLED},
-    )
-    addon.reload()
-    assert validate_version_number_is_gt_latest_signed_listed_version(
-        addon, '123.1'
-    ) == ('Version 123.1 must be greater than the previous approved version 123.5.')
-
-    disabled.delete()
-    # Shouldn't make a difference even if it's deleted - it was still signed.
-    assert validate_version_number_is_gt_latest_signed_listed_version(
-        addon, '123.1'
-    ) == ('Version 123.1 must be greater than the previous approved version 123.5.')
-
-    # Also check the edge case when addon is None
-    assert not validate_version_number_is_gt_latest_signed_listed_version(None, '123')
-
-
-@pytest.mark.django_db
-def test_validate_version_number_is_gt_latest_signed_listed_version_not_langpack():
-    addon = addon_factory(version_kw={'version': '123.0'}, file_kw={'is_signed': True})
-    assert validate_version_number_is_gt_latest_signed_listed_version(addon, '122') == (
-        'Version 122 must be greater than the previous approved version 123.0.'
-    )
-    addon.update(type=amo.ADDON_LPAPP)
-    assert not validate_version_number_is_gt_latest_signed_listed_version(addon, '122')

src/olympia/addons/utils.py

@@ -127,31 +127,6 @@ def validate(self, token, addon_id):
         return token_payload['addon_id'] == addon_id
 
 
-def validate_version_number_is_gt_latest_signed_listed_version(addon, version_string):
-    """Returns an error string if `version_string` isn't greater than the current
-    approved listed version. Doesn't apply to langpacks."""
-    if (
-        addon
-        and addon.type != amo.ADDON_LPAPP
-        and (
-            latest_version_string := addon.versions(manager='unfiltered_for_relations')
-            .filter(channel=amo.CHANNEL_LISTED, file__is_signed=True)
-            .order_by('created')
-            .values_list('version', flat=True)
-            .last()
-        )
-        and latest_version_string >= version_string
-    ):
-        msg = gettext(
-            'Version {version_string} must be greater than the previous approved '
-            'version {previous_version_string}.'
-        )
-        return msg.format(
-            version_string=version_string,
-            previous_version_string=latest_version_string,
-        )
-
-
 def remove_icons(addon):
     for size in amo.ADDON_ICON_SIZES + ('original',):
         filepath = addon.get_icon_path(size)