Merge pull request readthedocs#5998 from stsewd/add-move-method

Add move method to automation rule

readthedocs/builds/models.py

@@ -8,6 +8,7 @@
 
 from django.conf import settings
 from django.db import models
+from django.db.models import F
 from django.urls import reverse
 from django.utils import timezone
 from django.utils.translation import ugettext
@@ -1035,6 +1036,87 @@ def apply_action(self, version, match_result):
             raise NotImplementedError
         action(version, match_result, self.action_arg)
 
+    def move(self, steps):
+        """
+        Change the priority of this Automation Rule.
+
+        This is done by moving it ``n`` steps,
+        relative to the other priority rules.
+        The priority from the other rules are updated too.
+
+        :param steps: Number of steps to be moved
+                      (it can be negative)
+        :returns: True if the priority was changed
+        """
+        total = self.project.automation_rules.count()
+        current_priority = self.priority
+        new_priority = (current_priority + steps) % total
+
+        if current_priority == new_priority:
+            return False
+
+        # Move other's priority
+        if new_priority > current_priority:
+            # It was moved down
+            rules = (
+                self.project.automation_rules
+                .filter(priority__gt=current_priority, priority__lte=new_priority)
+                # We sort the queryset in asc order
+                # to be updated in that order
+                # to avoid hitting the unique constraint (project, priority).
+                .order_by('priority')
+            )
+            expression = F('priority') - 1
+        else:
+            # It was moved up
+            rules = (
+                self.project.automation_rules
+                .filter(priority__lt=current_priority, priority__gte=new_priority)
+                .exclude(pk=self.pk)
+                # We sort the queryset in desc order
+                # to be updated in that order
+                # to avoid hitting the unique constraint (project, priority).
+                .order_by('-priority')
+            )
+            expression = F('priority') + 1
+
+        # Put an imposible priority to avoid
+        # the unique constraint (project, priority)
+        # while updating.
+        self.priority = total + 99
+        self.save()
+
+        # We update each object one by one to
+        # avoid hitting the unique constraint (project, priority).
+        for rule in rules:
+            rule.priority = expression
+            rule.save()
+
+        # Put back new priority
+        self.priority = new_priority
+        self.save()
+        return True
+
+    def delete(self, *args, **kwargs):  # pylint: disable=arguments-differ
+        """Override method to update the other priorities after delete."""
+        current_priority = self.priority
+        project = self.project
+        super().delete(*args, **kwargs)
+
+        rules = (
+            project.automation_rules
+            .filter(priority__gte=current_priority)
+            # We sort the queryset in asc order
+            # to be updated in that order
+            # to avoid hitting the unique constraint (project, priority).
+            .order_by('priority')
+        )
+        # We update each object one by one to
+        # avoid hitting the unique constraint (project, priority).
+        for rule in rules:
+            rule.priority = F('priority') - 1
+            rule.save()
+
     def get_description(self):
         if self.description:
             return self.description

readthedocs/rtd_tests/tests/test_automation_rules.py

@@ -183,3 +183,259 @@ def test_add_rule_regex(self):
         )
         assert self.project.automation_rules.count() == 4
         assert rule.priority == 10
+
+
+@pytest.mark.django_db
+class TestAutomationRuleMove:
+
+    @pytest.fixture(autouse=True)
+    def setup_method(self):
+        self.project = get(Project)
+        self.rule_0 = self._add_rule('Zero')
+        self.rule_1 = self._add_rule('One')
+        self.rule_2 = self._add_rule('Two')
+        self.rule_3 = self._add_rule('Three')
+        self.rule_4 = self._add_rule('Four')
+        self.rule_5 = self._add_rule('Five')
+        assert self.project.automation_rules.count() == 6
+
+    def _add_rule(self, description):
+        rule = RegexAutomationRule.objects.add_rule(
+            project=self.project,
+            description=description,
+            match_arg='.*',
+            version_type=BRANCH,
+            action=VersionAutomationRule.ACTIVATE_VERSION_ACTION,
+        )
+        return rule
+
+    def test_move_rule_one_step(self):
+        self.rule_0.move(1)
+        new_order = [
+            self.rule_1,
+            self.rule_0,
+            self.rule_2,
+            self.rule_3,
+            self.rule_4,
+            self.rule_5,
+        ]
+
+        for priority, rule in enumerate(self.project.automation_rules.all()):
+            assert rule == new_order[priority]
+            assert rule.priority == priority
+
+    def test_move_rule_positive_steps(self):
+        self.rule_1.move(1)
+        self.rule_1.move(2)
+
+        new_order = [
+            self.rule_0,
+            self.rule_2,
+            self.rule_3,
+            self.rule_4,
+            self.rule_1,
+            self.rule_5,
+        ]
+
+        for priority, rule in enumerate(self.project.automation_rules.all()):
+            assert rule == new_order[priority]
+            assert rule.priority == priority
+
+    def test_move_rule_positive_steps_overflow(self):
+        self.rule_2.move(3)
+        self.rule_2.move(2)
+
+        new_order = [
+            self.rule_0,
+            self.rule_2,
+            self.rule_1,
+            self.rule_3,
+            self.rule_4,
+            self.rule_5,
+        ]
+
+        for priority, rule in enumerate(self.project.automation_rules.all()):
+            assert rule == new_order[priority]
+            assert rule.priority == priority
+
+    def test_move_rules_positive_steps(self):
+        self.rule_2.move(2)
+        self.rule_0.refresh_from_db()
+        self.rule_0.move(7)
+        self.rule_4.refresh_from_db()
+        self.rule_4.move(4)
+        self.rule_1.refresh_from_db()
+        self.rule_1.move(1)
+
+        new_order = [
+            self.rule_4,
+            self.rule_1,
+            self.rule_0,
+            self.rule_3,
+            self.rule_2,
+            self.rule_5,
+        ]
+
+        for priority, rule in enumerate(self.project.automation_rules.all()):
+            assert rule == new_order[priority]
+            assert rule.priority == priority
+
+    def test_move_rule_one_negative_step(self):
+        self.rule_3.move(-1)
+        new_order = [
+            self.rule_0,
+            self.rule_1,
+            self.rule_3,
+            self.rule_2,
+            self.rule_4,
+            self.rule_5,
+        ]
+
+        for priority, rule in enumerate(self.project.automation_rules.all()):
+            assert rule == new_order[priority]
+            assert rule.priority == priority
+
+    def test_move_rule_negative_steps(self):
+        self.rule_4.move(-1)
+        self.rule_4.move(-2)
+
+        new_order = [
+            self.rule_0,
+            self.rule_4,
+            self.rule_1,
+            self.rule_2,
+            self.rule_3,
+            self.rule_5,
+        ]
+
+        for priority, rule in enumerate(self.project.automation_rules.all()):
+            assert rule == new_order[priority]
+            assert rule.priority == priority
+
+    def test_move_rule_negative_steps_overflow(self):
+        self.rule_2.move(-3)
+        self.rule_2.move(-2)
+
+        new_order = [
+            self.rule_0,
+            self.rule_1,
+            self.rule_3,
+            self.rule_2,
+            self.rule_4,
+            self.rule_5,
+        ]
+
+        for priority, rule in enumerate(self.project.automation_rules.all()):
+            assert rule == new_order[priority]
+            assert rule.priority == priority
+
+    def test_move_rules_negative_steps(self):
+        self.rule_2.move(-2)
+        self.rule_5.refresh_from_db()
+        self.rule_5.move(-7)
+        self.rule_3.refresh_from_db()
+        self.rule_3.move(-2)
+        self.rule_1.refresh_from_db()
+        self.rule_1.move(-1)
+
+        new_order = [
+            self.rule_2,
+            self.rule_3,
+            self.rule_1,
+            self.rule_0,
+            self.rule_5,
+            self.rule_4,
+        ]
+
+        for priority, rule in enumerate(self.project.automation_rules.all()):
+            assert rule == new_order[priority]
+            assert rule.priority == priority
+
+    def test_move_rules_up_and_down(self):
+        self.rule_2.move(2)
+        self.rule_5.refresh_from_db()
+        self.rule_5.move(-3)
+        self.rule_3.refresh_from_db()
+        self.rule_3.move(4)
+        self.rule_1.refresh_from_db()
+        self.rule_1.move(-1)
+
+        new_order = [
+            self.rule_0,
+            self.rule_1,
+            self.rule_3,
+            self.rule_5,
+            self.rule_4,
+            self.rule_2,
+        ]
+
+        for priority, rule in enumerate(self.project.automation_rules.all()):
+            assert rule == new_order[priority]
+            assert rule.priority == priority
+
+    def test_delete_fist_rule(self):
+        self.rule_0.delete()
+        assert self.project.automation_rules.all().count() == 5
+
+        new_order = [
+            self.rule_1,
+            self.rule_2,
+            self.rule_3,
+            self.rule_4,
+            self.rule_5,
+        ]
+
+        for priority, rule in enumerate(self.project.automation_rules.all()):
+            assert rule == new_order[priority]
+            assert rule.priority == priority
+
+    def test_delete_last_rule(self):
+        self.rule_5.delete()
+        assert self.project.automation_rules.all().count() == 5
+
+        new_order = [
+            self.rule_0,
+            self.rule_1,
+            self.rule_2,
+            self.rule_3,
+            self.rule_4,
+        ]
+
+        for priority, rule in enumerate(self.project.automation_rules.all()):
+            assert rule == new_order[priority]
+            assert rule.priority == priority
+
+    def test_delete_some_rule(self):
+        self.rule_2.delete()
+        assert self.project.automation_rules.all().count() == 5
+
+        new_order = [
+            self.rule_0,
+            self.rule_1,
+            self.rule_3,
+            self.rule_4,
+            self.rule_5,
+        ]
+
+        for priority, rule in enumerate(self.project.automation_rules.all()):
+            assert rule == new_order[priority]
+            assert rule.priority == priority
+
+    def test_delete_some_rules(self):
+        self.rule_2.delete()
+        self.rule_0.refresh_from_db()
+        self.rule_0.delete()
+        self.rule_5.refresh_from_db()
+        self.rule_5.delete()
+
+        assert self.project.automation_rules.all().count() == 3
+
+        new_order = [
+            self.rule_1,
+            self.rule_3,
+            self.rule_4,
+        ]
+
+        for priority, rule in enumerate(self.project.automation_rules.all()):
+            assert rule == new_order[priority]
+            assert rule.priority == priority