mirror of
https://github.com/jwinterm/modoboa.git
synced 2026-10-17 05:41:25 +02:00
Do not include extensions code from core.
This commit is contained in:
Antoine Nguyen
@@ -1,4 +1,6 @@
|
||||
"""Management command to check and fix known problems."""
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.management.base import BaseCommand
|
||||
|
||||
from modoboa.lib.permissions import grant_access_to_object
|
||||
@@ -7,20 +9,88 @@ from modoboa.lib.permissions import get_object_owner
|
||||
from modoboa.core.models import User
|
||||
from modoboa.admin import models
|
||||
|
||||
try:
|
||||
from modoboa_postfix_autoreply.models import ARmessage
|
||||
except ImportError:
|
||||
ARmessage = None
|
||||
|
||||
|
||||
known_problems = []
|
||||
|
||||
|
||||
def known_problem(func):
|
||||
known_problems.append(func.__name__)
|
||||
"""Simple decorator to register a known problem."""
|
||||
known_problems.append(func)
|
||||
return func
|
||||
|
||||
|
||||
def log(message, quiet=False, **options):
|
||||
if not quiet:
|
||||
print(message)
|
||||
|
||||
|
||||
def fix_owner(qs, dry_run=False, **options):
|
||||
"""Fix ownership for orphan objects."""
|
||||
model = qs.model
|
||||
for obj in qs:
|
||||
kw = {"cls": model.__name__, "obj": obj}
|
||||
if get_object_owner(obj) is not None:
|
||||
continue
|
||||
if dry_run:
|
||||
log(" {cls} {obj} has no owner".format(**kw), **options)
|
||||
continue
|
||||
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:
|
||||
# Fallback: 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
|
||||
log(" {cls} {obj} is now owned by {admin}".format(**kw),
|
||||
**options)
|
||||
|
||||
|
||||
@known_problem
|
||||
def sometimes_objects_have_no_owner(**options):
|
||||
"""Sometime objects have no owner."""
|
||||
owned_models = (
|
||||
User.objects.all(),
|
||||
models.Domain.objects.all(),
|
||||
models.DomainAlias.objects.all(),
|
||||
models.Alias.objects.filter(domain__isnull=False),
|
||||
models.Mailbox.objects.filter(domain__isnull=False),
|
||||
)
|
||||
for qs in owned_models:
|
||||
fix_owner(qs, **options)
|
||||
|
||||
|
||||
@known_problem
|
||||
def sometimes_mailbox_have_no_alias(**options):
|
||||
"""Sometime mailboxes have no alias."""
|
||||
alias_created = 0
|
||||
recipient_created = 0
|
||||
for instance in models.Mailbox.objects.select_related("domain").all():
|
||||
alias, created = models.Alias.objects.get_or_create(
|
||||
address=instance.full_address,
|
||||
domain=instance.domain,
|
||||
internal=True)
|
||||
if created:
|
||||
alias_created += 1
|
||||
log("Alias {0} created".format(alias))
|
||||
recipient, created = models.AliasRecipient.objects.get_or_create(
|
||||
alias=alias,
|
||||
address=instance.full_address,
|
||||
r_mailbox=instance)
|
||||
if created:
|
||||
recipient_created += 1
|
||||
log("AliasRecipient {0} created".format(recipient))
|
||||
if alias_created or recipient_created:
|
||||
log("{0} alias created. {1} alias recipient created".format(
|
||||
alias_created, recipient_created))
|
||||
|
||||
|
||||
class Repair(BaseCommand):
|
||||
"""Command class."""
|
||||
|
||||
@@ -37,107 +107,14 @@ class Repair(BaseCommand):
|
||||
|
||||
def handle(self, *args, **options):
|
||||
"""Command entry point."""
|
||||
for problem in known_problems:
|
||||
func = getattr(self, problem)
|
||||
# Load known problems from extensions.
|
||||
for ext in settings.MODOBOA_APPS:
|
||||
try:
|
||||
__import__(ext, locals(), globals(), ["known_problems"])
|
||||
except ImportError:
|
||||
pass
|
||||
for func in known_problems:
|
||||
title = func.__doc__.strip()
|
||||
self.log("", **options)
|
||||
self.log("Checking for... {}...".format(title), **options)
|
||||
log("", **options)
|
||||
log("Checking for... {}...".format(title), **options)
|
||||
func(**options)
|
||||
|
||||
def log(self, message, quiet=False, **options):
|
||||
if not quiet:
|
||||
print(message)
|
||||
|
||||
def fix_owner(self, qs, dry_run=False, **options):
|
||||
model = qs.model
|
||||
for obj in qs:
|
||||
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):
|
||||
"""Sometime objects have no owner"""
|
||||
owned_models = (
|
||||
User.objects.all(),
|
||||
models.Domain.objects.all(),
|
||||
models.DomainAlias.objects.all(),
|
||||
models.Alias.objects.filter(domain__isnull=False),
|
||||
models.Mailbox.objects.filter(domain__isnull=False),
|
||||
)
|
||||
for qs in owned_models:
|
||||
self.fix_owner(qs, **options)
|
||||
|
||||
@known_problem
|
||||
def ensure_autoreplies_recipents_are_valids(self, **options):
|
||||
"""Sometime autoreply alias exists when ARmessage is not enabled"""
|
||||
if ARmessage is None:
|
||||
return
|
||||
deleted = 0
|
||||
qs = models.AliasRecipient.objects.filter(
|
||||
address__contains='@autoreply.')
|
||||
for alias in qs:
|
||||
address, domain = alias.alias.address.split('@')
|
||||
arqs = ARmessage.objects.filter(
|
||||
mbox__address=address,
|
||||
mbox__domain__name=domain)
|
||||
if not arqs.count():
|
||||
self.log('Delete {0} (No AR found)'.format(alias))
|
||||
deleted += 1
|
||||
else:
|
||||
for ar in arqs:
|
||||
if not ar.enabled:
|
||||
self.log('Delete {0} (AR disabled)'.format(alias))
|
||||
alias.delete()
|
||||
deleted += 1
|
||||
if deleted:
|
||||
self.log('{0} alias recipient deleted'.format(deleted))
|
||||
|
||||
@known_problem
|
||||
def sometimes_mailbox_have_no_alias(self, **options):
|
||||
"""Sometime mailboxes have no alias"""
|
||||
alias_created = 0
|
||||
recipient_created = 0
|
||||
for instance in models.Mailbox.objects.all():
|
||||
alias, created = models.Alias.objects.get_or_create(
|
||||
address=instance.full_address,
|
||||
domain=instance.domain,
|
||||
internal=True)
|
||||
if created:
|
||||
alias_created += 1
|
||||
self.log('Alias {0} created'.format(alias))
|
||||
recipient, created = models.AliasRecipient.objects.get_or_create(
|
||||
alias=alias,
|
||||
address=instance.full_address,
|
||||
r_mailbox=instance)
|
||||
if created:
|
||||
recipient_created += 1
|
||||
self.log('AliasRecipient {0} created'.format(recipient))
|
||||
if alias_created or recipient_created:
|
||||
self.log('{0} alias created. {1} alias recipient created'.format(
|
||||
alias_created, recipient_created))
|
||||
|
||||
@@ -45,7 +45,7 @@ class RepairTestCase(ModoTestCase):
|
||||
self.assertIs(get_object_owner(mbox), None)
|
||||
|
||||
def test_management_command_with_nul_domain(self):
|
||||
""" Just assume nothing raise when an alias has no domain"""
|
||||
"""Just assume nothing raise when an alias has no domain."""
|
||||
models.Alias.objects.create(address='@modoboa.xxx')
|
||||
ret = management.call_command("modo", "repair", "--quiet")
|
||||
assert ret is None
|
||||
|
||||
Reference in New Issue
Block a user