diff options
Diffstat (limited to 'pydis_site/apps')
4 files changed, 754 insertions, 0 deletions
diff --git a/pydis_site/apps/api/migrations/0044_active_infractions_migration.py b/pydis_site/apps/api/migrations/0044_active_infractions_migration.py new file mode 100644 index 00000000..14746712 --- /dev/null +++ b/pydis_site/apps/api/migrations/0044_active_infractions_migration.py @@ -0,0 +1,106 @@ +# Generated by Django 2.2.6 on 2019-10-07 15:59 + +from django.db import migrations +from django.db.models import Count, Prefetch, Q, 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.""" + inactive_infraction_types = Q(type="note") | Q(type="warning") | Q(type="kick") + infraction_model = apps.get_model('api', 'Infraction') + infraction_model.objects.filter(inactive_infraction_types).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( + Q(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', '0043_infraction_hidden_warnings_to_notes'), + ] + + operations = [ + migrations.RunPython(migrate_inactive_types_to_inactive, reverse_migration), + migrations.RunPython(migrate_multiple_active_infractions_per_user_to_one, reverse_migration) + ] diff --git a/pydis_site/apps/api/tests/migrations/__init__.py b/pydis_site/apps/api/tests/migrations/__init__.py new file mode 100644 index 00000000..38e42ffc --- /dev/null +++ b/pydis_site/apps/api/tests/migrations/__init__.py @@ -0,0 +1 @@ +"""This submodule contains tests for functions used in data migrations.""" diff --git a/pydis_site/apps/api/tests/migrations/base.py b/pydis_site/apps/api/tests/migrations/base.py new file mode 100644 index 00000000..0c0a5bd0 --- /dev/null +++ b/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 diff --git a/pydis_site/apps/api/tests/migrations/test_active_infraction_migration.py b/pydis_site/apps/api/tests/migrations/test_active_infraction_migration.py new file mode 100644 index 00000000..cfae4219 --- /dev/null +++ b/pydis_site/apps/api/tests/migrations/test_active_infraction_migration.py @@ -0,0 +1,545 @@ +"""Tests for the data migration in `filename`.""" +import logging +from collections import ChainMap, namedtuple +from datetime import timedelta +from itertools import count +from typing import Dict, Iterable, Type, Union + +from django.db.models import Q +from django.forms.models import model_to_dict +from django.utils import timezone + +from pydis_site.apps.api.models import Infraction, User +from .base import MigrationsTestCase + +log = logging.getLogger(__name__) +log.setLevel(logging.DEBUG) + + +InfractionHistory = namedtuple('InfractionHistory', ("user_id", "infraction_history")) + + +class InfractionFactory: + """Factory that creates infractions for a User instance.""" + + infraction_id = count(1) + user_id = count(1) + default_values = { + 'active': True, + 'expires_at': None, + 'hidden': False, + } + + @classmethod + def create( + cls, + actor: User, + infractions: Iterable[Dict[str, Union[str, int, bool]]], + infraction_model: Type[Infraction] = Infraction, + user_model: Type[User] = User, + ) -> InfractionHistory: + """ + Creates `infractions` for the `user` with the given `actor`. + + The `infractions` dictionary can contain the following fields: + - `type` (required) + - `active` (default: True) + - `expires_at` (default: None; i.e, permanent) + - `hidden` (default: False). + + The parameters `infraction_model` and `user_model` can be used to pass in an instance of + both model classes from a different migration/project state. + """ + user_id = next(cls.user_id) + user = user_model.objects.create( + id=user_id, + name=f"Infracted user {user_id}", + discriminator=user_id, + avatar_hash=None, + ) + infraction_history = [] + + for infraction in infractions: + infraction = dict(infraction) + infraction["id"] = next(cls.infraction_id) + infraction = ChainMap(infraction, cls.default_values) + new_infraction = infraction_model.objects.create( + user=user, + actor=actor, + type=infraction["type"], + reason=f"`{infraction['type']}` infraction (ID: {infraction['id']} of {user}", + active=infraction['active'], + hidden=infraction['hidden'], + expires_at=infraction['expires_at'], + ) + infraction_history.append(new_infraction) + + return InfractionHistory(user_id=user_id, infraction_history=infraction_history) + + +class InfractionFactoryTests(MigrationsTestCase): + """Tests for the InfractionFactory.""" + + app = "api" + migration_prior = "0043_infraction_hidden_warnings_to_notes" + migration_target = "0043_infraction_hidden_warnings_to_notes" + + @classmethod + def setUpPostMigrationData(cls, apps): + """Create a default actor for all infractions.""" + cls.infraction_model = apps.get_model('api', 'Infraction') + cls.user_model = apps.get_model('api', 'User') + + cls.actor = cls.user_model.objects.create( + id=9999, + name="Unknown Moderator", + discriminator=1040, + avatar_hash=None, + ) + + def test_infraction_factory_total_count(self): + """Does the test database hold as many infractions as we tried to create?""" + InfractionFactory.create( + actor=self.actor, + infractions=( + {'type': 'kick', 'active': False, 'hidden': False}, + {'type': 'ban', 'active': True, 'hidden': False}, + {'type': 'note', 'active': False, 'hidden': True}, + ), + infraction_model=self.infraction_model, + user_model=self.user_model, + ) + database_count = Infraction.objects.all().count() + self.assertEqual(3, database_count) + + def test_infraction_factory_multiple_users(self): + """Does the test database hold as many infractions as we tried to create?""" + created_infractions = {} + for user in range(5): + created_infractions.update( + { + user: InfractionFactory.create( + actor=self.actor, + infractions=( + {'type': 'kick', 'active': False, 'hidden': True}, + {'type': 'ban', 'active': True, 'hidden': False}, + ), + infraction_model=self.infraction_model, + user_model=self.user_model, + ) + } + ) + + # Check if infractions and users are recorded properly in the database + database_count = Infraction.objects.all().count() + self.assertEqual(database_count, 10) + + user_count = User.objects.all().count() + self.assertEqual(user_count, 5 + 1) + + def test_infraction_factory_sets_correct_fields(self): + """Does the InfractionFactory set the correct attributes?""" + infractions = ( + { + 'type': 'note', + 'active': False, + 'hidden': True, + 'expires_at': timezone.now() + }, + {'type': 'warning', 'active': False, 'hidden': False, 'expires_at': None}, + {'type': 'watch', 'active': False, 'hidden': True, 'expires_at': None}, + {'type': 'mute', 'active': True, 'hidden': False, 'expires_at': None}, + {'type': 'kick', 'active': True, 'hidden': True, 'expires_at': None}, + {'type': 'ban', 'active': True, 'hidden': False, 'expires_at': None}, + { + 'type': 'superstar', + 'active': True, + 'hidden': True, + 'expires_at': timezone.now() + }, + ) + + InfractionFactory.create( + actor=self.actor, + infractions=infractions, + infraction_model=self.infraction_model, + user_model=self.user_model, + ) + + for infraction in infractions: + with self.subTest(**infraction): + self.assertTrue(Infraction.objects.filter(**infraction).exists()) + + +class ActiveInfractionMigrationTests(MigrationsTestCase): + """ + Tests the active infraction data migration. + + The active infraction data migration should do the following things: + + 1. migrates all active notes, warnings, and kicks to an inactive status; + 2. migrates all users with multiple active infractions of a single type to have only one active + infraction of that type. The infraction with the longest duration stays active. + """ + + app = "api" + migration_prior = "0043_infraction_hidden_warnings_to_notes" + migration_target = "0044_active_infractions_migration" + + @classmethod + def setUpMigrationData(cls, apps): + """Sets up an initial database state that contains the relevant test cases.""" + # Fetch the Infraction and User model in the current migration state + cls.infraction_model = apps.get_model('api', 'Infraction') + cls.user_model = apps.get_model('api', 'User') + + cls.created_infractions = {} + + # Moderator that serves as actor for all infractions + cls.user_moderator = cls.user_model.objects.create( + id=9999, + name="Olivier de Vienne", + discriminator=1040, + avatar_hash=None, + ) + + # User #1: clean user with no infractions + cls.created_infractions.update( + { + 1: InfractionFactory.create( + actor=cls.user_moderator, + infractions=[], + infraction_model=cls.infraction_model, + user_model=cls.user_model, + ) + } + ) + + # User #2: One inactive note infraction + cls.created_infractions.update( + { + 2: InfractionFactory.create( + actor=cls.user_moderator, + infractions=( + {'type': 'note', 'active': False, 'hidden': True}, + ), + infraction_model=cls.infraction_model, + user_model=cls.user_model, + ) + } + ) + + # User #3: One active note infraction + cls.created_infractions.update( + { + 3: InfractionFactory.create( + actor=cls.user_moderator, + infractions=( + {'type': 'note', 'active': True, 'hidden': True}, + ), + infraction_model=cls.infraction_model, + user_model=cls.user_model, + ) + } + ) + + # User #4: One active and one inactive note infraction + cls.created_infractions.update( + { + 4: InfractionFactory.create( + actor=cls.user_moderator, + infractions=( + {'type': 'note', 'active': False, 'hidden': True}, + {'type': 'note', 'active': True, 'hidden': True}, + ), + infraction_model=cls.infraction_model, + user_model=cls.user_model, + ) + } + ) + + # User #5: Once active note, one active kick, once active warning + cls.created_infractions.update( + { + 5: InfractionFactory.create( + actor=cls.user_moderator, + infractions=( + {'type': 'note', 'active': True, 'hidden': True}, + {'type': 'kick', 'active': True, 'hidden': True}, + {'type': 'warning', 'active': True, 'hidden': True}, + ), + infraction_model=cls.infraction_model, + user_model=cls.user_model, + ) + } + ) + + # User #6: One inactive ban and one active ban + cls.created_infractions.update( + { + 6: InfractionFactory.create( + actor=cls.user_moderator, + infractions=( + {'type': 'ban', 'active': False, 'hidden': True}, + {'type': 'ban', 'active': True, 'hidden': True}, + ), + infraction_model=cls.infraction_model, + user_model=cls.user_model, + ) + } + ) + + # User #7: Two active permanent bans + cls.created_infractions.update( + { + 7: InfractionFactory.create( + actor=cls.user_moderator, + infractions=( + {'type': 'ban', 'active': True, 'hidden': True}, + {'type': 'ban', 'active': True, 'hidden': True}, + ), + infraction_model=cls.infraction_model, + user_model=cls.user_model, + ) + } + ) + + # User #8: Multiple active temporary bans + cls.created_infractions.update( + { + 8: InfractionFactory.create( + actor=cls.user_moderator, + infractions=( + { + 'type': 'ban', + 'active': True, + 'hidden': True, + 'expires_at': timezone.now() + timedelta(days=1) + }, + { + 'type': 'ban', + 'active': True, + 'hidden': True, + 'expires_at': timezone.now() + timedelta(days=10) + }, + { + 'type': 'ban', + 'active': True, + 'hidden': True, + 'expires_at': timezone.now() + timedelta(days=20) + }, + { + 'type': 'ban', + 'active': True, + 'hidden': True, + 'expires_at': timezone.now() + timedelta(days=5) + }, + ), + infraction_model=cls.infraction_model, + user_model=cls.user_model, + ) + } + ) + + # User #9: One active permanent ban, two active temporary bans + cls.created_infractions.update( + { + 9: InfractionFactory.create( + actor=cls.user_moderator, + infractions=( + { + 'type': 'ban', + 'active': True, + 'hidden': True, + 'expires_at': timezone.now() + timedelta(days=10) + }, + { + 'type': 'ban', + 'active': True, + 'hidden': True, + 'expires_at': None, + }, + { + 'type': 'ban', + 'active': True, + 'hidden': True, + 'expires_at': timezone.now() + timedelta(days=7) + }, + ), + infraction_model=cls.infraction_model, + user_model=cls.user_model, + ) + } + ) + + # User #10: One inactive permanent ban, two active temporary bans + cls.created_infractions.update( + { + 10: InfractionFactory.create( + actor=cls.user_moderator, + infractions=( + { + 'type': 'ban', + 'active': True, + 'hidden': True, + 'expires_at': timezone.now() + timedelta(days=10) + }, + { + 'type': 'ban', + 'active': False, + 'hidden': True, + 'expires_at': None, + }, + { + 'type': 'ban', + 'active': True, + 'hidden': True, + 'expires_at': timezone.now() + timedelta(days=7) + }, + ), + infraction_model=cls.infraction_model, + user_model=cls.user_model, + ) + } + ) + + # User #11: Active ban, active mute, active superstar + cls.created_infractions.update( + { + 11: InfractionFactory.create( + actor=cls.user_moderator, + infractions=( + {'type': 'ban', 'active': True, 'hidden': True}, + {'type': 'mute', 'active': True, 'hidden': True}, + {'type': 'superstar', 'active': True, 'hidden': True}, + {'type': 'watch', 'active': True, 'hidden': True}, + ), + infraction_model=cls.infraction_model, + user_model=cls.user_model, + ) + } + ) + + # User #12: Multiple active bans, active mutes, active superstars + cls.created_infractions.update( + { + 12: InfractionFactory.create( + actor=cls.user_moderator, + infractions=( + {'type': 'ban', 'active': True, 'hidden': True}, + {'type': 'ban', 'active': True, 'hidden': True}, + {'type': 'ban', 'active': True, 'hidden': True}, + {'type': 'mute', 'active': True, 'hidden': True}, + {'type': 'mute', 'active': True, 'hidden': True}, + {'type': 'mute', 'active': True, 'hidden': True}, + {'type': 'superstar', 'active': True, 'hidden': True}, + {'type': 'superstar', 'active': True, 'hidden': True}, + {'type': 'superstar', 'active': True, 'hidden': True}, + {'type': 'watch', 'active': True, 'hidden': True}, + {'type': 'watch', 'active': True, 'hidden': True}, + {'type': 'watch', 'active': True, 'hidden': True}, + ), + infraction_model=cls.infraction_model, + user_model=cls.user_model, + ) + } + ) + + def test_all_never_active_types_became_inactive(self): + """Are all infractions of a non-active type inactive after the migration?""" + inactive_type_query = Q(type="note") | Q(type="warning") | Q(type="kick") + self.assertFalse( + self.infraction_model.objects.filter(inactive_type_query, active=True).exists() + ) + + def test_migration_left_clean_user_without_infractions(self): + """Do users without infractions have no infractions after the migration?""" + user_id, infraction_history = self.created_infractions[1] + self.assertFalse( + self.infraction_model.objects.filter(user__id=user_id).exists() + ) + + def test_migration_left_user_with_inactive_note_untouched(self): + """Did the migration leave users with only an inactive note untouched?""" + user_id, infraction_history = self.created_infractions[2] + inactive_note = infraction_history[0] + self.assertTrue( + self.infraction_model.objects.filter(**model_to_dict(inactive_note)).exists() + ) + + def test_migration_only_touched_active_field_of_active_note(self): + """Does the migration only change the `active` field?""" + user_id, infraction_history = self.created_infractions[3] + note = model_to_dict(infraction_history[0]) + note['active'] = False + self.assertTrue( + self.infraction_model.objects.filter(**note).exists() + ) + + def test_migration_only_touched_active_field_of_active_note_left_inactive_untouched(self): + """Does the migration only change the `active` field of active notes?""" + user_id, infraction_history = self.created_infractions[4] + for note in infraction_history: + with self.subTest(active=note.active): + note = model_to_dict(note) + note['active'] = False + self.assertTrue( + self.infraction_model.objects.filter(**note).exists() + ) + + def test_migration_migrates_all_nonactive_types_to_inactive(self): + """Do we set the `active` field of all non-active infractions to `False`?""" + user_id, infraction_history = self.created_infractions[5] + self.assertFalse( + self.infraction_model.objects.filter(user__id=user_id, active=True).exists() + ) + + def test_migration_leaves_user_with_one_active_ban_untouched(self): + """Do we leave a user with one active and one inactive ban untouched?""" + user_id, infraction_history = self.created_infractions[6] + for infraction in infraction_history: + with self.subTest(active=infraction.active): + self.assertTrue( + self.infraction_model.objects.filter(**model_to_dict(infraction)).exists() + ) + + def test_migration_turns_double_active_perm_ban_into_single_active_perm_ban(self): + """Does the migration turn two active permanent bans into one active permanent ban?""" + user_id, infraction_history = self.created_infractions[7] + active_count = self.infraction_model.objects.filter(user__id=user_id, active=True).count() + self.assertEqual(active_count, 1) + + def test_migration_leaves_temporary_ban_with_longest_duration_active(self): + """Does the migration turn two active permanent bans into one active permanent ban?""" + user_id, infraction_history = self.created_infractions[8] + active_ban = self.infraction_model.objects.get(user__id=user_id, active=True) + self.assertEqual(active_ban.expires_at, infraction_history[2].expires_at) + + def test_migration_leaves_permanent_ban_active(self): + """Does the migration leave the permanent ban active?""" + user_id, infraction_history = self.created_infractions[9] + active_ban = self.infraction_model.objects.get(user__id=user_id, active=True) + self.assertIsNone(active_ban.expires_at) + + def test_migration_leaves_longest_temp_ban_active_with_inactive_permanent_ban(self): + """Does the longest temp ban stay active, even with an inactive perm ban present?""" + user_id, infraction_history = self.created_infractions[10] + active_ban = self.infraction_model.objects.get(user__id=user_id, active=True) + self.assertEqual(active_ban.expires_at, infraction_history[0].expires_at) + + def test_migration_leaves_all_active_types_active_if_one_of_each_exists(self): + """Do all active infractions stay active if only one of each is present?""" + user_id, infraction_history = self.created_infractions[11] + active_count = self.infraction_model.objects.filter(user__id=user_id, active=True).count() + self.assertEqual(active_count, 4) + + def test_migration_reduces_all_active_types_to_a_single_active_infraction(self): + """Do we reduce all of the infraction types to one active infraction?""" + user_id, infraction_history = self.created_infractions[12] + active_infractions = self.infraction_model.objects.filter(user__id=user_id, active=True) + self.assertEqual(len(active_infractions), 4) + types_observed = [infraction.type for infraction in active_infractions] + + for infraction_type in ('ban', 'mute', 'superstar', 'watch'): + with self.subTest(type=infraction_type): + self.assertIn(infraction_type, types_observed) |