Validation for active infractions

pydis_site/apps/api/migrations/0047_active_infractions_migration.py

@@ -0,0 +1,105 @@
+# Generated by Django 2.2.6 on 2019-10-07 15:59
+
+from django.db import migrations
+from django.db.models import Count, Prefetch, QuerySet
+
+
+class ExpirationWrapper:
+    """Wraps an expiration date to properly compare permanent and temporary infractions."""
+
+    def __init__(self, infraction):
+        self.expiration_date = infraction.expires_at
+
+    def __lt__(self, other):
+        """An `expiration_date` is considered smaller when it comes earlier than the `other`."""
+        if self.expiration_date is None:
+            # A permanent infraction can never end sooner than another infraction
+            return False
+        elif other.expiration_date is None:
+            # If `self` is temporary, but `other` is permanent, `self` is smaller
+            return True
+        else:
+            return self.expiration_date < other.expiration_date
+
+    def __eq__(self, other):
+        """If both expiration dates are permanent they're equal, otherwise compare dates."""
+        if self.expiration_date is None and other.expiration_date is None:
+            return True
+        elif self.expiration_date is None or other.expiration_date is None:
+            return False
+        else:
+            return self.expiration_date == other.expiration_date
+
+
+def migrate_inactive_types_to_inactive(apps, schema_editor):
+    """Migrates infractions of non-active types to inactive."""
+    infraction_model = apps.get_model('api', 'Infraction')
+    infraction_model.objects.filter(type__in=('note', 'warning', 'kick')).update(active=False)
+
+
+def get_query(user_model, infraction_model, infr_type: str) -> QuerySet:
+    """
+    Creates QuerySet to fetch users with multiple active infractions of the given `type`.
+
+    The QuerySet will prefetch the infractions and attach them as an `.infractions` attribute to the
+    `User` instances.
+    """
+    active_infractions = infraction_model.objects.filter(type=infr_type, active=True)
+
+    # Build an SQL query by chaining methods together
+
+    # Get users with active infraction(s) of the provided `infr_type`
+    query = user_model.objects.filter(
+        infractions_received__type=infr_type, infractions_received__active=True
+    )
+
+    # Prefetch their active received infractions of `infr_type` and attach `.infractions` attribute
+    query = query.prefetch_related(
+        Prefetch('infractions_received', queryset=active_infractions, to_attr='infractions')
+    )
+
+    # Count and only include them if they have at least 2 active infractions of the `type`
+    query = query.annotate(num_infractions=Count('infractions_received'))
+    query = query.filter(num_infractions__gte=2)
+
+    # Make sure we return each individual only once
+    query = query.distinct()
+
+    return query
+
+
+def migrate_multiple_active_infractions_per_user_to_one(apps, schema_editor):
+    """
+    Make sure a user only has one active infraction of a given "active" infraction type.
+
+    If a user has multiple active infraction, we keep the one with longest expiration date active
+    and migrate the others to inactive.
+    """
+    infraction_model = apps.get_model('api', 'Infraction')
+    user_model = apps.get_model('api', 'User')
+
+    for infraction_type in ('ban', 'mute', 'superstar', 'watch'):
+        query = get_query(user_model, infraction_model, infraction_type)
+        for user in query:
+            infractions = sorted(user.infractions, key=ExpirationWrapper, reverse=True)
+            for infraction in infractions[1:]:
+                infraction.active = False
+                infraction.save()
+
+
+def reverse_migration(apps, schema_editor):
+    """There's no need to do anything special to reverse these migrations."""
+    return
+
+
+class Migration(migrations.Migration):
+    """Data migration to get the database consistent with the new infraction validation rules."""
+
+    dependencies = [
+        ('api', '0046_reminder_jump_url'),
+    ]
+
+    operations = [
+        migrations.RunPython(migrate_inactive_types_to_inactive, reverse_migration),
+        migrations.RunPython(migrate_multiple_active_infractions_per_user_to_one, reverse_migration)
+    ]

pydis_site/apps/api/migrations/0048_add_infractions_unique_constraints_active.py

@@ -0,0 +1,17 @@
+# Generated by Django 2.2.6 on 2019-10-07 18:27
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api', '0047_active_infractions_migration'),
+    ]
+
+    operations = [
+        migrations.AddConstraint(
+            model_name='infraction',
+            constraint=models.UniqueConstraint(condition=models.Q(active=True), fields=('user', 'type'), name='unique_active_infraction_per_type_per_user'),
+        ),
+    ]

pydis_site/apps/api/models/bot/infraction.py

@@ -71,3 +71,10 @@ class Meta:
         """Defines the meta options for the infraction model."""
 
         ordering = ['-inserted_at']
+        constraints = (
+            models.UniqueConstraint(
+                fields=["user", "type"],
+                condition=models.Q(active=True),
+                name="unique_active_infraction_per_type_per_user"
+            ),
+        )

pydis_site/apps/api/models/bot/user.py

@@ -50,7 +50,7 @@ class User(ModelReprMixin, models.Model):
 
     def __str__(self):
         """Returns the name and discriminator for the current user, for display purposes."""
-        return f"{self.name}#{self.discriminator}"
+        return f"{self.name}#{self.discriminator:0>4}"
 
     @property
     def top_role(self) -> Role:

pydis_site/apps/api/serializers.py

@@ -1,6 +1,6 @@
 """Converters from Django models to data interchange formats and back."""
-
 from rest_framework.serializers import ModelSerializer, PrimaryKeyRelatedField, ValidationError
+from rest_framework.validators import UniqueTogetherValidator
 from rest_framework_bulk import BulkSerializerMixin
 
 from .models import (
@@ -105,11 +105,22 @@ class Meta:
         fields = (
             'id', 'inserted_at', 'expires_at', 'active', 'user', 'actor', 'type', 'reason', 'hidden'
         )
+        validators = [
+            UniqueTogetherValidator(
+                queryset=Infraction.objects.filter(active=True),
+                fields=['user', 'type'],
+                message='This user already has an active infraction of this type.',
+            )
+        ]
 
     def validate(self, attrs: dict) -> dict:
         """Validate data constraints for the given data and abort if it is invalid."""
         infr_type = attrs.get('type')
 
+        active = attrs.get('active')
+        if active and infr_type in ('note', 'warning', 'kick'):
+            raise ValidationError({'active': [f'{infr_type} infractions cannot be active.']})
+
         expires_at = attrs.get('expires_at')
         if expires_at and infr_type in ('kick', 'warning'):
             raise ValidationError({'expires_at': [f'{infr_type} infractions cannot expire.']})

pydis_site/apps/api/tests/migrations/__init__.py

@@ -0,0 +1 @@
+"""This submodule contains tests for functions used in data migrations."""

pydis_site/apps/api/tests/migrations/base.py

@@ -0,0 +1,102 @@
+"""Includes utilities for testing migrations."""
+from django.db import connection
+from django.db.migrations.executor import MigrationExecutor
+from django.test import TestCase
+
+
+class MigrationsTestCase(TestCase):
+    """
+    A `TestCase` subclass to test migration files.
+
+    To be able to properly test a migration, we will need to inject data into the test database
+    before the migrations we want to test are applied, but after the older migrations have been
+    applied. This makes sure that we are testing "as if" we were actually applying this migration
+    to a database in the state it was in before introducing the new migration.
+
+    To set up a MigrationsTestCase, create a subclass of this class and set the following
+    class-level attributes:
+
+    - app: The name of the app that contains the migrations (e.g., `'api'`)
+    - migration_prior: The name* of the last migration file before the migrations you want to test
+    - migration_target: The name* of the last migration file we want to test
+
+    *) Specify the file names without a path or the `.py` file extension.
+
+    Additionally, overwrite the `setUpMigrationData` in the subclass to inject data into the
+    database before the migrations we want to test are applied. Please read the docstring of the
+    method for more information. An optional hook, `setUpPostMigrationData` is also provided.
+    """
+
+    # These class-level attributes should be set in classes that inherit from this base class.
+    app = None
+    migration_prior = None
+    migration_target = None
+
+    @classmethod
+    def setUpTestData(cls):
+        """
+        Injects data into the test database prior to the migration we're trying to test.
+
+        This class methods reverts the test database back to the state of the last migration file
+        prior to the migrations we want to test. It will then allow the user to inject data into the
+        test database by calling the `setUpMigrationData` hook. After the data has been injected, it
+        will apply the migrations we want to test and call the `setUpPostMigrationData` hook. The
+        user can now test if the migration correctly migrated the injected test data.
+        """
+        if not cls.app:
+            raise ValueError("The `app` attribute was not set.")
+
+        if not cls.migration_prior or not cls.migration_target:
+            raise ValueError("Both ` migration_prior` and `migration_target` need to be set.")
+
+        cls.migrate_from = [(cls.app, cls.migration_prior)]
+        cls.migrate_to = [(cls.app, cls.migration_target)]
+
+        # Reverse to database state prior to the migrations we want to test
+        executor = MigrationExecutor(connection)
+        executor.migrate(cls.migrate_from)
+
+        # Call the data injection hook with the current state of the project
+        old_apps = executor.loader.project_state(cls.migrate_from).apps
+        cls.setUpMigrationData(old_apps)
+
+        # Run the migrations we want to test
+        executor = MigrationExecutor(connection)
+        executor.loader.build_graph()
+        executor.migrate(cls.migrate_to)
+
+        # Save the project state so we're able to work with the correct model states
+        cls.apps = executor.loader.project_state(cls.migrate_to).apps
+
+        # Call `setUpPostMigrationData` to potentially set up post migration data used in testing
+        cls.setUpPostMigrationData(cls.apps)
+
+    @classmethod
+    def setUpMigrationData(cls, apps):
+        """
+        Override this method to inject data into the test database before the migration is applied.
+
+        This method will be called after setting up the database according to the migrations that
+        come before the migration(s) we are trying to test, but before the to-be-tested migration(s)
+        are applied. This allows us to simulate a database state just prior to the migrations we are
+        trying to test.
+
+        To make sure we're creating objects according to the state the models were in at this point
+        in the migration history, use `apps.get_model(app_name: str, model_name: str)` to get the
+        appropriate model, e.g.:
+
+        >>> Infraction = apps.get_model('api', 'Infraction')
+        """
+        pass
+
+    @classmethod
+    def setUpPostMigrationData(cls, apps):
+        """
+        Set up additional test data after the target migration has been applied.
+
+        Use `apps.get_model(app_name: str, model_name: str)` to get the correct instances of the
+        model classes:
+
+        >>> Infraction = apps.get_model('api', 'Infraction')
+        """
+        pass