Merge pull request #957 from bearstech/repair_command

add repair command

modoboa/admin/management/commands/modo.py

@@ -5,6 +5,7 @@ from subcommand.base import SubcommandCommand
 from .subcommands._mx import CheckMXRecords
 from .subcommands._export import ExportCommand
 from .subcommands._import import ImportCommand
+from .subcommands._repair import Repair
 
 
 class Command(SubcommandCommand):
@@ -19,4 +20,5 @@ class Command(SubcommandCommand):
         "export": ExportCommand,
         "import": ImportCommand,
         "check_mx": CheckMXRecords,
+        "repair": Repair,
     }

modoboa/admin/management/commands/subcommands/_repair.py

@@ -0,0 +1,86 @@
+"""Management command to check and fix known problems."""
+from django.core.management.base import BaseCommand
+
+from modoboa.lib.permissions import grant_access_to_object
+from modoboa.lib.permissions import get_object_owner
+
+from modoboa.core.models import User
+from modoboa.admin import models
+
+known_problems = []
+
+
+def known_problem(func):
+    known_problems.append(func.__name__)
+    return func
+
+
+class Repair(BaseCommand):
+    """Command class."""
+
+    help = "Check and fix known problems."
+
+    def add_arguments(self, parser):
+        """Add extra arguments to command."""
+        parser.add_argument(
+            "--dry-run", action="store_true", default=False,
+            help="List known problems without fixing them.")
+        parser.add_argument(
+            "--quiet", action="store_true", default=False,
+            help="Quiet mode.")
+
+    def handle(self, *args, **options):
+        """Command entry point."""
+        for problem in known_problems:
+            func = getattr(self, problem)
+            title = func.__name__.capitalize().replace("_", " ")
+            self.log("", **options)
+            self.log("Checking for... {}...".format(title), **options)
+            func(**options)
+
+    def log(self, message, quiet=False, **options):
+        if not quiet:
+            print(message)
+
+    def fix_owner(self, model, dry_run=False, **options):
+        for obj in model.objects.all():
+            kw = dict(
+                cls=model.__name__,
+                obj=obj
+            )
+            if get_object_owner(obj) is None:
+                if dry_run:
+                    self.log(
+                        "  {cls} {obj} has no owner".format(**kw),
+                        **options)
+                else:
+                    if isinstance(obj, User):
+                        admin = User.objects.filter(is_superuser=True,
+                                                    is_active=True).first()
+                    elif isinstance(obj, models.Domain):
+                        admin = obj.admins.first()
+                    elif isinstance(obj, models.DomainAlias):
+                        admin = obj.target.admins.first()
+                    else:
+                        admin = obj.domain.admins.first()
+                    if not admin:
+                        # domain has no admin. use the first superuser found
+                        admin = User.objects.filter(is_superuser=True,
+                                                    is_active=True).first()
+                    grant_access_to_object(admin, obj, is_owner=True)
+                    kw['admin'] = admin
+                    self.log(
+                        "  {cls} {obj} is now owned by {admin}".format(**kw),
+                        **options)
+
+    @known_problem
+    def sometimes_objects_have_no_owner(self, **options):
+        owned_models = (
+            User,
+            models.Domain,
+            models.Mailbox,
+            models.Alias,
+            models.DomainAlias
+        )
+        for model in owned_models:
+            self.fix_owner(model, **options)

modoboa/admin/tests/test_repair.py

@@ -0,0 +1,41 @@
+"""Repair command tests"""
+from django.core import management
+
+from modoboa.lib.permissions import get_object_owner
+from modoboa.lib.permissions import ObjectAccess
+from modoboa.lib.tests import ModoTestCase
+
+from .. import factories
+from .. import models
+
+
+class RepairTestCase(ModoTestCase):
+    """TestCase for repair command."""
+
+    @classmethod
+    def setUpTestData(cls):
+        """Create some data."""
+        super(RepairTestCase, cls).setUpTestData()
+        factories.populate_database()
+
+    def test_management_command(self):
+        """Check that command works fine."""
+        ObjectAccess.objects.all().delete()
+        mbox = models.Mailbox.objects.all()[0]
+        # assert mbox has no owner
+        self.assertIs(get_object_owner(mbox), None)
+        # fix it. run in quiet mode because we dont want output in tests
+        management.call_command("modo", "repair", "--quiet")
+        # assert its fixed
+        self.assertIsNot(get_object_owner(mbox), None)
+
+    def test_management_command_with_dry_run(self):
+        """Check that command works fine."""
+        ObjectAccess.objects.all().delete()
+        mbox = models.Mailbox.objects.all()[0]
+        # assert mbox has no owner
+        self.assertIs(get_object_owner(mbox), None)
+        # show problems. run in quiet mode because we dont want output in tests
+        management.call_command("modo", "repair", "--quiet", "--dry-run")
+        # assert its not fixed
+        self.assertIs(get_object_owner(mbox), None)