github-jwinterm/modoboa
1
0
Fork 0
mirror of https://github.com/jwinterm/modoboa.git synced 2026-10-17 05:41:25 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #862 from tonioo/feature/per_domain_limits_799

Browse Source 
Per domain limits
This commit is contained in:
Antoine Nguyen
2016-04-27 11:50:45 +02:00
parent e85a0c8f77 f1863e61a6
commit d6f0f4fcda
34 changed files with 1145 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modoboa/admin/callbacks.py
+6 -1
View File
@@ -3,6 +3,7 @@
from django.core.urlresolvers import reverse
from django.utils.translation import ugettext as _, ugettext_lazy
from modoboa.core import signals as core_signals
from modoboa.lib import parameters, events
from modoboa.lib.email_utils import split_mailbox
from modoboa.lib.exceptions import PermDeniedException, BadRequest, Conflict
@@ -103,7 +104,6 @@ def import_account_mailbox(user, account, row):
"""
account.email = row[0].strip()
if account.email:
account.save()
mailbox, domname = split_mailbox(account.email)
try:
domain = Domain.objects.get(name=domname)
@@ -114,6 +114,11 @@ def import_account_mailbox(user, account, row):
)
if not user.can_access(domain):
raise PermDeniedException
core_signals.can_create_object.send(
sender="import", context=user, object_type="mailboxes")
core_signals.can_create_object.send(
sender="import", context=domain, object_type="mailboxes")
account.save()
if Mailbox.objects.filter(address=mailbox, domain=domain).exists():
raise Conflict(
_("Mailbox {} already exists").format(account.email))
modoboa/admin/forms/account.py
+17 -10
View File
@@ -13,7 +13,7 @@ from modoboa.core.models import User
from modoboa.core import signals as core_signals
from modoboa.lib import events, parameters
from modoboa.lib.email_utils import split_mailbox
from modoboa.lib.exceptions import PermDeniedException
from modoboa.lib import exceptions as lib_exceptions
from modoboa.lib import fields as lib_fields
from modoboa.lib.form_utils import (
DynamicForm, TabForms, WizardForm, WizardStep
@@ -161,7 +161,8 @@ class AccountFormGeneral(forms.ModelForm):
def save(self, commit=True):
account = super(AccountFormGeneral, self).save(commit=False)
if self.user == account and not self.cleaned_data["is_active"]:
raise PermDeniedException(_("You can't disable your own account"))
raise lib_exceptions.PermDeniedException(
_("You can't disable your own account"))
if commit:
if "password1" in self.cleaned_data \
and self.cleaned_data["password1"] != "":
@@ -198,11 +199,7 @@ class AccountFormMail(forms.Form, DynamicForm):
)
def __init__(self, *args, **kwargs):
if "instance" in kwargs:
self.mb = kwargs["instance"]
del kwargs["instance"]
else:
self.mb = None
self.mb = kwargs.pop("instance", None)
super(AccountFormMail, self).__init__(*args, **kwargs)
self.field_widths = {
"quota": 3
@@ -245,6 +242,13 @@ class AccountFormMail(forms.Form, DynamicForm):
self.domain = Domain.objects.get(name=domname)
except Domain.DoesNotExist:
raise forms.ValidationError(_("Domain does not exist"))
if not self.mb:
try:
core_signals.can_create_object.send(
sender=self.__class__, context=self.domain,
object_type="mailboxes")
except lib_exceptions.ModoboaException as inst:
raise forms.ValidationError(inst)
return email
def clean(self):
@@ -262,9 +266,9 @@ class AccountFormMail(forms.Form, DynamicForm):
def create_mailbox(self, user, account):
"""Create a mailbox associated to :kw:`account`."""
if not user.can_access(self.domain):
raise PermDeniedException
raise lib_exceptions.PermDeniedException
core_signals.can_create_object.send(
self.__class__, user=user, object_type="mailboxes")
self.__class__, context=user, object_type="mailboxes")
self.mb = Mailbox(
address=self.locpart, domain=self.domain, user=account,
use_domain_quota=self.cleaned_data["quota_act"])
@@ -296,8 +300,11 @@ class AccountFormMail(forms.Form, DynamicForm):
if not aliases:
return
core_signals.can_create_object.send(
self.__class__, user=user, object_type="mailbox_aliases",
self.__class__, context=user, object_type="mailbox_aliases",
count=len(aliases))
core_signals.can_create_object.send(
self.__class__, context=self.mb.domain,
object_type="mailbox_aliases", count=len(aliases))
for alias in aliases:
if self.mb.aliasrecipient_set.select_related("alias").filter(
alias__address=alias).exists():
modoboa/admin/forms/alias.py
+9 -1
View File
@@ -6,6 +6,8 @@ from django import forms
from django.http import QueryDict
from django.utils.translation import ugettext as _, ugettext_lazy
from modoboa.core import signals as core_signals
from modoboa.lib import exceptions as lib_exceptions
from modoboa.lib.email_utils import split_mailbox
from modoboa.lib import fields as lib_fields
from modoboa.lib.form_utils import (
@@ -16,7 +18,6 @@ from ..models import Domain, Alias
class AliasForm(forms.ModelForm, DynamicForm):
"""A form to create/modify an alias."""
address = lib_fields.UTF8AndEmptyUserEmailField(
@@ -74,6 +75,13 @@ class AliasForm(forms.ModelForm, DynamicForm):
raise forms.ValidationError(
_("You don't have access to this domain")
)
if not self.instance.pk:
try:
core_signals.can_create_object.send(
sender=self.__class__, context=domain,
object_type="mailbox_aliases")
except lib_exceptions.ModoboaException as inst:
raise forms.ValidationError(inst)
return self.cleaned_data["address"].lower()
def clean(self):
modoboa/admin/forms/domain.py
+16 -21
View File
@@ -101,6 +101,7 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm):
"name", _("A %s with this name already exists")
% unicode(label)
)
self.aliases = []
for k in cleaned_data.keys():
if not k.startswith("aliases"):
continue
@@ -120,6 +121,8 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm):
k, _("A %s with this name already exists")
% unicode(label)
)
else:
self.aliases.append(cleaned_data[k])
return cleaned_data
def save(self, user, commit=True, domalias_post_create=False):
@@ -134,28 +137,20 @@ class DomainFormGeneral(forms.ModelForm, DynamicForm):
if not commit:
return d
d.save()
aliases = []
for k, v in self.cleaned_data.iteritems():
if not k.startswith("aliases"):
continue
if v in ["", None]:
continue
aliases.append(v)
for dalias in d.domainalias_set.all():
if dalias.name not in aliases:
if dalias.name not in self.aliases:
dalias.delete()
else:
aliases.remove(dalias.name)
if aliases:
self.aliases.remove(dalias.name)
if self.aliases:
core_signals.can_create_object.send(
self.__class__, user=user, object_type="domain_aliases",
count=len(aliases))
for alias in aliases:
try:
d.domainalias_set.get(name=alias)
except DomainAlias.DoesNotExist:
pass
else:
self.__class__, context=user, object_type="domain_aliases",
count=len(self.aliases))
core_signals.can_create_object.send(
self.__class__, context=d, object_type="domain_aliases",
count=len(self.aliases))
for alias in self.aliases:
if d.domainalias_set.filter(name=alias).exists():
continue
options = {"creator": user} if domalias_post_create else {}
DomainAlias(name=alias, target=d, enabled=d.enabled).save(
@@ -225,9 +220,9 @@ class DomainFormOptions(forms.Form):
except User.DoesNotExist:
pass
else:
raise Conflict(_("User '%s' already exists" % username))
raise Conflict(_("User '%s' already exists") % username)
core_signals.can_create_object.send(
self.__class__, user=user, object_type="mailboxes")
self.__class__, context=user, object_type="mailboxes")
da = User(username=username, email=username, is_active=True)
da.set_password("password")
da.save()
@@ -245,7 +240,7 @@ class DomainFormOptions(forms.Form):
if domain.type == "domain" and \
self.cleaned_data["create_aliases"] == "yes":
core_signals.can_create_object.send(
self.__class__, user=user, object_type="mailbox_aliases")
self.__class__, context=user, object_type="mailbox_aliases")
alias = Alias(
address=u"postmaster@{}".format(domain.name),
domain=domain, enabled=True
modoboa/admin/lib.py
+5
View File
@@ -10,6 +10,7 @@ from django.db.models import Q
from django.utils.translation import ugettext as _
from modoboa.core.models import User
from modoboa.core import signals as core_signals
from modoboa.lib import events
from modoboa.lib.exceptions import PermDeniedException
@@ -118,6 +119,8 @@ def import_domain(user, row, formopts):
"""Specific code for domains import"""
if not user.has_perm("admin.add_domain"):
raise PermDeniedException(_("You are not allowed to import domains"))
core_signals.can_create_object.send(
sender="import", context=user, object_type="domains")
dom = Domain()
dom.from_csv(user, row)
@@ -127,6 +130,8 @@ def import_domainalias(user, row, formopts):
if not user.has_perm("admin.add_domainalias"):
raise PermDeniedException(
_("You are not allowed to import domain aliases."))
core_signals.can_create_object.send(
sender="import", context=user, object_type="domain_aliases")
domalias = DomainAlias()
domalias.from_csv(user, row)
modoboa/admin/models/alias.py
+5
View File
@@ -6,6 +6,7 @@ from django.utils.translation import ugettext as _, ugettext_lazy
import reversion
from modoboa.core import signals as core_signals
from modoboa.lib.email_utils import split_mailbox
from modoboa.lib.exceptions import (
PermDeniedException, BadRequest, Conflict, NotFound
@@ -171,6 +172,10 @@ class Alias(AdminObject):
raise BadRequest(_("Domain '%s' does not exist" % domname))
if not user.can_access(domain):
raise PermDeniedException
core_signals.can_create_object.send(
sender="import", context=user, object_type="mailbox_aliases")
core_signals.can_create_object.send(
sender="import", context=domain, object_type="mailbox_aliases")
if Alias.objects.filter(address=address).exists():
raise Conflict
self.address = address
modoboa/admin/models/domain.py
+13 -5
View File
@@ -8,11 +8,13 @@ from django.utils.translation import ugettext as _, ugettext_lazy
import reversion
from .base import AdminObject
from modoboa.core.models import User, ObjectAccess
from modoboa.core import signals as core_signals
from modoboa.lib import events, parameters
from modoboa.lib.exceptions import BadRequest, Conflict
from .base import AdminObject
class DomainManager(Manager):
@@ -85,22 +87,28 @@ class Domain(AdminObject):
@property
def admins(self):
"""Return the domain administrators of this domain
"""Return the domain administrators of this domain.
:return: a list of User objects
"""
return [oa.user for oa in self.owners.filter(user__is_superuser=False)]
return User.objects.filter(
is_superuser=False,
objectaccess__content_type__model="domain",
objectaccess__object_id=self.pk)
@property
def aliases(self):
return self.domainalias_set
def add_admin(self, account):
"""Add a new administrator for this domain
"""Add a new administrator to this domain.
:param User account: the administrotor to add
:param User account: the administrator
"""
from modoboa.lib.permissions import grant_access_to_object
core_signals.can_create_object.send(
sender=self.__class__, context=self, object_type="domain_admins")
grant_access_to_object(account, self)
for mb in self.mailbox_set.all():
if mb.user.has_perm("admin.add_domain"):
modoboa/admin/models/domain_alias.py
+6 -2
View File
@@ -8,10 +8,12 @@ from django.contrib.contenttypes import generic
import reversion
from modoboa.core import models as core_models
from modoboa.core import signals as core_signals
from modoboa.lib.exceptions import BadRequest, Conflict
from .base import AdminObject
from .domain import Domain
from modoboa.core import models as core_models
from modoboa.lib.exceptions import BadRequest, Conflict
class DomainAliasManager(models.Manager):
@@ -76,6 +78,8 @@ class DomainAlias(AdminObject):
self.target = Domain.objects.get(name=domname)
except Domain.DoesNotExist:
raise BadRequest(_("Unknown domain %s") % domname)
core_signals.can_create_object.send(
sender="import", context=self.target, object_type="domain_aliases")
self.enabled = row[3].strip() in ["True", "1", "yes", "y"]
self.save(creator=user)
modoboa/admin/serializers.py
+38
View File
@@ -9,6 +9,8 @@ from passwords import validators
from modoboa.admin import models as admin_models
from modoboa.core import models as core_models
from modoboa.core import signals as core_signals
from modoboa.lib import exceptions as lib_exceptions
from modoboa.lib import permissions, email_utils, fields as lib_fields
from . import models
@@ -42,6 +44,23 @@ class DomainAliasSerializer(serializers.ModelSerializer):
raise serializers.ValidationError(_("Permission denied."))
return value
def create(self, validated_data):
"""Custom creation."""
domain_alias = models.DomainAlias(**validated_data)
creator = self.context["request"].user
try:
core_signals.can_create_object.send(
sender=self.__class__, context=creator,
object_type="domain_aliases")
core_signals.can_create_object.send(
sender=self.__class__, context=domain_alias.target,
object_type="domain_aliases")
except lib_exceptions.ModoboaException as inst:
raise serializers.ValidationError({
"domain": unicode(inst)})
domain_alias.save(creator=creator)
return domain_alias
class MailboxSerializer(serializers.ModelSerializer):
"""Base mailbox serializer."""
@@ -175,6 +194,16 @@ class WritableAccountSerializer(AccountSerializer):
if not creator.can_access(domain):
raise serializers.ValidationError({
"domain": _("Permission denied.")})
try:
core_signals.can_create_object.send(
sender=self.__class__, context=creator,
object_type="mailboxes")
core_signals.can_create_object.send(
sender=self.__class__, context=domain,
object_type="mailboxes")
except lib_exceptions.ModoboaException as inst:
raise serializers.ValidationError({
"domain": unicode(inst)})
mb = admin_models.Mailbox(
user=account, address=address, domain=domain, **data)
mb.set_quota(
@@ -241,6 +270,15 @@ class AliasSerializer(serializers.ModelSerializer):
def create(self, validated_data):
"""Create appropriate objects."""
creator = self.context["request"].user
try:
core_signals.can_create_object.send(
sender=self.__class__, context=creator,
object_type="mailbox_aliases")
core_signals.can_create_object.send(
sender=self.__class__, context=self.domain,
object_type="mailbox_aliases")
except lib_exceptions.ModoboaException as inst:
raise serializers.ValidationError(unicode(inst))
recipients = validated_data.pop("recipients", None)
alias = admin_models.Alias(domain=self.domain, **validated_data)
alias.save(creator=creator)
modoboa/admin/tests/test_account.py
+2 -2
View File
@@ -161,9 +161,9 @@ class PermissionsTestCase(ModoTestCase):
from modoboa.limits import utils as limits_utils
super(PermissionsTestCase, cls).setUpTestData()
for name, tpl in limits_utils.get_limit_templates():
for name, tpl in limits_utils.get_user_limit_templates():
parameters.save_admin(
"DEFLT_{0}_LIMIT".format(name.upper()), 2,
"DEFLT_USER_{0}_LIMIT".format(name.upper()), 2,
app="limits"
)
factories.populate_database()
modoboa/admin/tests/test_api.py
+2
View File
@@ -9,6 +9,7 @@ from django.core.urlresolvers import reverse
from rest_framework.authtoken.models import Token
from modoboa.core import models as core_models
from modoboa.lib import parameters
from modoboa.lib.tests import ModoAPITestCase
from .. import factories
@@ -401,6 +402,7 @@ class AliasAPITestCase(ModoAPITestCase):
def setUpTestData(cls):
"""Create test data."""
super(AliasAPITestCase, cls).setUpTestData()
parameters.save_admin("ENABLE_ADMIN_LIMITS", "no", app="limits")
factories.populate_database()
cls.da_token = Token.objects.create(
user=core_models.User.objects.get(username="admin@test.com"))
modoboa/admin/tests/test_import_.py
+2
View File
@@ -3,6 +3,7 @@ from django.core.files.base import ContentFile
from django.core.urlresolvers import reverse
from modoboa.core.models import User
from modoboa.lib import parameters
from modoboa.lib.tests import ModoTestCase
from .. import factories
@@ -15,6 +16,7 @@ class ImportTestCase(ModoTestCase):
def setUpTestData(cls):
"""Create test data."""
super(ImportTestCase, cls).setUpTestData()
parameters.save_admin("ENABLE_ADMIN_LIMITS", "no", app="limits")
factories.populate_database()
def test_domains_import(self):
modoboa/admin/views/alias.py
+1 -1
View File
@@ -37,7 +37,7 @@ def _validate_alias(request, form, successmsg, callback=None):
def _new_alias(request, title, action, successmsg,
tplname="admin/aliasform.html"):
core_signals.can_create_object.send(
"new_alias", user=request.user, object_type="mailbox_aliases")
"new_alias", context=request.user, object_type="mailbox_aliases")
if request.method == "POST":
def callback(user, alias):
alias.post_create(user)
modoboa/admin/views/domain.py
+1 -1
View File
@@ -100,7 +100,7 @@ def domains_list(request):
@reversion.create_revision()
def newdomain(request):
core_signals.can_create_object.send(
"newdomain", user=request.user, object_type="domains")
"newdomain", context=request.user, object_type="domains")
return DomainWizard(request).process()
modoboa/core/__init__.py
+4 -1
View File
@@ -39,7 +39,10 @@ PERMISSIONS = {
["admin", "domain", "view_domains"],
["admin", "domain", "add_domain"],
["admin", "domain", "change_domain"],
["admin", "domain", "delete_domain"]
["admin", "domain", "delete_domain"],
["admin", "domainalias", "add_domainalias"],
["admin", "domainalias", "change_domainalias"],
["admin", "domainalias", "delete_domainalias"],
]
}
modoboa/core/serializers.py
-41
View File
@@ -1,41 +0,0 @@
"""Modoboa core serializers."""
from django.core.exceptions import ValidationError
from rest_framework import serializers
from passwords import validators
from . import models
class UserPasswordSerializer(serializers.ModelSerializer):
"""A serializer used to change a user password."""
new_password = serializers.CharField()
class Meta:
model = models.User
fields = (
"password", "new_password", )
def validate_password(self, value):
"""Check password."""
if not self.instance.check_password(value):
raise serializers.ValidationError("Password not correct")
return value
def validate_new_password(self, value):
"""Check new password."""
try:
validators.validate_length(value)
validators.complexity(value)
except ValidationError as exc:
raise serializers.ValidationError(exc.messages[0])
return value
def update(self, instance, validated_data):
"""Set new password."""
instance.set_password(validated_data["new_password"])
instance.save()
return instance
modoboa/core/signals.py
+1 -1
View File
@@ -3,4 +3,4 @@
import django.dispatch
can_create_object = django.dispatch.Signal(
providing_args=["user", "object_type", "count"])
providing_args=["context", "object_type", "count"])
modoboa/lib/migrations/0005_auto_20160416_1449.py
+23
View File
@@ -0,0 +1,23 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import migrations
def rename_limits_parameters(apps, schema_editor):
"""Rename parameters."""
Parameter = apps.get_model("lib", "Parameter")
for param in Parameter.objects.filter(name__startswith="DEFLT_"):
param.name = param.name.replace("DEFLT_", "DEFLT_USER_")
param.save()
class Migration(migrations.Migration):
dependencies = [
('lib', '0004_auto_20151114_1409'),
]
operations = [
migrations.RunPython(rename_limits_parameters)
]
modoboa/limits/app_settings.py
+78 -7
View File
@@ -3,16 +3,24 @@
from django import forms
from django.utils.translation import ugettext_lazy as _
from modoboa.lib.form_utils import SeparatorField
from modoboa.lib.form_utils import SeparatorField, YesNoField
from modoboa.lib.parameters import AdminParametersForm
class ParametersForm(AdminParametersForm):
"""Available admin parameters."""
app = "limits"
defv_sep = SeparatorField(label=_("Default limits"))
defv_sep = SeparatorField(label=_("Default per-admin limits"))
deflt_domain_admins_limit = forms.IntegerField(
enable_admin_limits = YesNoField(
label=_("Enable per-admin limits"),
initial="yes",
help_text=_("Enable or disable per-admin limits")
)
deflt_user_domain_admins_limit = forms.IntegerField(
label=_("Domain admins"),
initial=0,
help_text=_(
@@ -23,7 +31,7 @@ class ParametersForm(AdminParametersForm):
widget=forms.widgets.TextInput(
attrs={"class": "col-md-1 form-control"})
)
deflt_domains_limit = forms.IntegerField(
deflt_user_domains_limit = forms.IntegerField(
label=_("Domains"),
initial=0,
help_text=_(
@@ -33,7 +41,7 @@ class ParametersForm(AdminParametersForm):
widget=forms.widgets.TextInput(
attrs={"class": "col-md-1 form-control"})
)
deflt_domain_aliases_limit = forms.IntegerField(
deflt_user_domain_aliases_limit = forms.IntegerField(
label=_("Domain aliases"),
initial=0,
help_text=_(
@@ -44,7 +52,7 @@ class ParametersForm(AdminParametersForm):
widget=forms.widgets.TextInput(
attrs={"class": "col-md-1 form-control"})
)
deflt_mailboxes_limit = forms.IntegerField(
deflt_user_mailboxes_limit = forms.IntegerField(
label=_("Mailboxes"),
initial=0,
help_text=_(
@@ -54,7 +62,7 @@ class ParametersForm(AdminParametersForm):
widget=forms.widgets.TextInput(
attrs={"class": "col-md-1 form-control"})
)
deflt_mailbox_aliases_limit = forms.IntegerField(
deflt_user_mailbox_aliases_limit = forms.IntegerField(
label=_("Mailbox aliases"),
initial=0,
help_text=_(
@@ -65,6 +73,69 @@ class ParametersForm(AdminParametersForm):
attrs={"class": "col-md-1 form-control"})
)
domain_limits_sep = SeparatorField(label=_("Default per-domain limits"))
enable_domain_limits = YesNoField(
label=_("Enable per-domain limits"),
initial="no",
help_text=_("Enable or disable per-domain limits")
)
deflt_domain_domain_admins_limit = forms.IntegerField(
label=_("Domain admins"),
initial=0,
help_text=_(
"Maximum number of allowed domain administrators for a new "
"domain. (0 to deny any creation, -1 to allow unlimited "
"creations)"
),
widget=forms.widgets.TextInput(
attrs={"class": "col-md-1 form-control"})
)
deflt_domain_domain_aliases_limit = forms.IntegerField(
label=_("Domain aliases"),
initial=0,
help_text=_(
"Maximum number of allowed domain aliases for a new "
"domain. (0 to deny any creation, -1 to allow "
"unlimited creations)"
),
widget=forms.widgets.TextInput(
attrs={"class": "col-md-1 form-control"})
)
deflt_domain_mailboxes_limit = forms.IntegerField(
label=_("Mailboxes"),
initial=0,
help_text=_(
"Maximum number of allowed mailboxes for a new domain. "
"(0 to deny any creation, -1 to allow unlimited creations)"
),
widget=forms.widgets.TextInput(
attrs={"class": "col-md-1 form-control"})
)
deflt_domain_mailbox_aliases_limit = forms.IntegerField(
label=_("Mailbox aliases"),
initial=0,
help_text=_(
"Maximum number of allowed aliases for a new domain. "
"(0 to deny any creation, -1 to allow unlimited creations)"
),
widget=forms.widgets.TextInput(
attrs={"class": "col-md-1 form-control"})
)
visibility_rules = {
"deflt_user_domains_limit": "enable_admin_limits=yes",
"deflt_user_domain_aliases_limit": "enable_admin_limits=yes",
"deflt_user_mailboxes_limit": "enable_admin_limits=yes",
"deflt_user_mailbox_aliases_limit": "enable_admin_limits=yes",
"deflt_user_domain_admins_limit": "enable_admin_limits=yes",
"deflt_domain_mailboxes_limit": "enable_domain_limits=yes",
"deflt_domain_mailbox_aliases_limit": "enable_domain_limits=yes",
"deflt_domain_domain_aliases_limit": "enable_domain_limits=yes",
"deflt_domain_domain_admins_limit": "enable_domain_limits=yes",
}
def __init__(self, *args, **kwargs):
super(AdminParametersForm, self).__init__(*args, **kwargs)
self._load_extra_parameters('A')
modoboa/limits/constants.py
+22 -1
View File
@@ -5,7 +5,7 @@ import collections
from django.utils.translation import ugettext_lazy as _
DEFAULT_LIMITS = collections.OrderedDict((
DEFAULT_USER_LIMITS = collections.OrderedDict((
("domains", {
"content_type": "admin.domain", "label": _("Domains"),
"help": _("Maximum number of domains this user can create"),
@@ -28,3 +28,24 @@ DEFAULT_LIMITS = collections.OrderedDict((
"required_role": "Resellers",
"extra_filters": {"groups__name": "DomainAdmins"}})
))
DEFAULT_DOMAIN_LIMITS = collections.OrderedDict((
("domain_aliases", {
"relation": "domainalias_set", "label": _("Domain aliases"),
"help": _("Maximum number of domain aliases allowed for this domain.")
}),
("mailboxes", {
"relation": "mailbox_set", "label": _("Mailboxes"),
"help": _("Maximum number of mailboxes allowed for this domain.")}),
("mailbox_aliases", {
"relation": "alias_set", "label": _("Mailbox aliases"),
"help": _(
"Maximum number of mailbox aliases allowed for this domain."),
"extra_filters": {"internal": False}
}),
("domain_admins", {
"relation": "admins", "label": _("Domain admins"),
"help": _("Maximum number of domain admins allowed for this domain."),
})
))
modoboa/limits/controls.py
+9 -6
View File
@@ -5,7 +5,7 @@
"""
from modoboa.lib import events
from modoboa.lib import events, parameters
from modoboa.lib.permissions import get_object_owner
from . import utils
@@ -18,11 +18,11 @@ def move_pool_resource(owner, user):
associated resource to the pool of its owner.
"""
if not owner.is_superuser:
for name, ltpl in utils.get_limit_templates():
l = user.objectlimit_set.get(name=name)
for name, ltpl in utils.get_user_limit_templates():
l = user.userobjectlimit_set.get(name=name)
if l.max_value < 0:
continue
ol = owner.objectlimit_set.get(name=name)
ol = owner.userobjectlimit_set.get(name=name)
ol.max_value += l.max_value
ol.save()
@@ -39,12 +39,15 @@ def user_can_set_role(user, role, account=None):
:param ``User`` account: account modified (None on creation)
:param str newrole: role to check
"""
if role != "DomainAdmins":
condition = (
parameters.get_admin("ENABLE_ADMIN_LIMITS") == "no" or
role != "DomainAdmins")
if condition:
return [True]
lname = "domain_admins"
condition = (
user.is_superuser or
not user.objectlimit_set.get(name=lname).is_exceeded()
not user.userobjectlimit_set.get(name=lname).is_exceeded()
)
if condition:
return [True]
modoboa/limits/forms.py
+46 -11
View File
@@ -5,7 +5,6 @@
from django import forms
from django.utils.translation import ugettext as _
from modoboa.lib import parameters
from modoboa.lib.permissions import get_object_owner
from .lib import BadLimitValue, UnsufficientResource
@@ -16,12 +15,9 @@ class ResourcePoolForm(forms.Form):
"""Dynamic form to display user limits."""
def __init__(self, *args, **kwargs):
self.account = None
if "instance" in kwargs:
self.account = kwargs["instance"]
del kwargs["instance"]
self.account = kwargs.pop("instance", None)
super(ResourcePoolForm, self).__init__(*args, **kwargs)
for name, tpl in utils.get_limit_templates():
for name, tpl in utils.get_user_limit_templates():
if "required_role" in tpl:
condition = (
self.account is not None and
@@ -32,7 +28,7 @@ class ResourcePoolForm(forms.Form):
self.fields["{}_limit".format(name)] = forms.IntegerField(
label=tpl["label"], help_text=tpl["help"]
)
if hasattr(self, "account"):
if self.account:
self.load_from_user(self.account)
def clean(self):
@@ -50,7 +46,7 @@ class ResourcePoolForm(forms.Form):
for fieldname in self.fields.keys():
lname = fieldname.replace("_limit", "")
self.fields[fieldname].initial = (
user.objectlimit_set.get(name=lname).max_value)
user.userobjectlimit_set.get(name=lname).max_value)
def allocate_from_user(self, limit, user):
"""Allocate resource using an existing user.
@@ -60,7 +56,7 @@ class ResourcePoolForm(forms.Form):
also be limited, the resource he gives is taken from its own
pool.
"""
ol = user.objectlimit_set.get(name=limit.name)
ol = user.userobjectlimit_set.get(name=limit.name)
fieldname = "{}_limit".format(limit.name)
newvalue = self.cleaned_data[fieldname]
if newvalue == -1 and ol.max_value != -1:
@@ -80,12 +76,51 @@ class ResourcePoolForm(forms.Form):
def save(self):
owner = get_object_owner(self.account)
for name, ltpl in utils.get_limit_templates():
for name, ltpl in utils.get_user_limit_templates():
fieldname = "{}_limit".format(name)
if fieldname not in self.cleaned_data:
continue
l = self.account.objectlimit_set.get(name=name)
l = self.account.userobjectlimit_set.get(name=name)
if not owner.is_superuser:
self.allocate_from_user(l, owner)
l.max_value = self.cleaned_data[fieldname]
l.save()
class DomainLimitsForm(forms.Form):
"""Per-domain limits form."""
def __init__(self, *args, **kwargs):
"""Define limits as fields."""
self.domain = kwargs.pop("instance", None)
super(DomainLimitsForm, self).__init__(*args, **kwargs)
for name, tpl in utils.get_domain_limit_templates():
self.fields["{}_limit".format(name)] = forms.IntegerField(
label=tpl["label"], help_text=tpl["help"]
)
if not self.domain:
return
for fieldname in self.fields.keys():
lname = fieldname.replace("_limit", "")
self.fields[fieldname].initial = (
self.domain.domainobjectlimit_set.get(name=lname).max_value)
def clean(self):
"""Ensure limit values are correct."""
cleaned_data = super(DomainLimitsForm, self).clean()
if self.errors:
return cleaned_data
for lname in self.fields.keys():
if cleaned_data[lname] < -1:
self.add_error(lname, _("Invalid limit"))
return cleaned_data
def save(self, user):
"""Set limits."""
for name, ltpl in utils.get_domain_limit_templates():
fieldname = "{}_limit".format(name)
if fieldname not in self.cleaned_data:
continue
l = self.domain.domainobjectlimit_set.get(name=name)
l.max_value = self.cleaned_data[fieldname]
l.save()
modoboa/limits/general_callbacks.py
+45 -13
View File
@@ -2,9 +2,9 @@
from django.utils.translation import ugettext as _
from modoboa.lib import events
from modoboa.lib import events, parameters
from .forms import ResourcePoolForm
from . import forms
from . import utils
@@ -12,7 +12,10 @@ from . import utils
def display_pool_usage(user, target, currentpage):
from django.template.loader import render_to_string
if target != "leftcol" or user.is_superuser:
condition = (
parameters.get_admin("ENABLE_ADMIN_LIMITS") == "no" or
target != "leftcol" or user.is_superuser)
if condition:
return []
if currentpage == "identities":
names = ["mailboxes", "mailbox_aliases"]
@@ -21,13 +24,13 @@ def display_pool_usage(user, target, currentpage):
else:
exceptions = ["domain_admins", "mailboxes", "mailbox_aliases"]
names = [
name for name, tpl in utils.get_limit_templates()
name for name, tpl in utils.get_user_limit_templates()
if name not in exceptions and
("required_role" not in tpl or
tpl["required_role"] == user.group)
]
limits = user.objectlimit_set.filter(name__in=names, max_value__gt=0)
limits = user.userobjectlimit_set.filter(name__in=names, max_value__gt=0)
if len(limits) == 0:
return []
return [
@@ -38,17 +41,41 @@ def display_pool_usage(user, target, currentpage):
@events.observe("ExtraAccountForm")
def extra_account_form(user, account=None):
if parameters.get_admin("ENABLE_ADMIN_LIMITS") == "no":
return []
if user.group not in ["SuperAdmins", "Resellers"]:
return []
if account is not None and \
account.group not in ["Resellers", "DomainAdmins"]:
return []
return [
dict(
id="resources", title=_("Resources"), cls=ResourcePoolForm
)
]
return [{
"id": "resources", "title": _("Resources"),
"cls": forms.ResourcePoolForm
}]
@events.observe("ExtraDomainForm")
def extra_domain_form(user, domain):
"""Include domain limits form."""
if parameters.get_admin("ENABLE_DOMAIN_LIMITS") == "no":
return []
if not user.has_perm("admin.change_domain"):
return []
return [{
"id": "resources", "title": _("Resources"),
"cls": forms.DomainLimitsForm
}]
@events.observe("FillDomainInstances")
def fill_domain_instances(user, domain, instances):
"""Set domain instance for resources form."""
if parameters.get_admin("ENABLE_DOMAIN_LIMITS") == "no":
return
if not user.has_perm("admin.change_domain"):
return
instances["resources"] = domain
@events.observe("CheckExtraAccountForm")
@@ -63,16 +90,21 @@ def check_form_access(account, form):
@events.observe("FillAccountInstances")
def fill_account_instances(user, account, instances):
if not user.is_superuser and not user.belongs_to_group("Resellers"):
condition = (
parameters.get_admin("ENABLE_ADMIN_LIMITS") == "no" or
(not user.is_superuser and user.group != "Resellers")
)
if condition:
return
if not account.belongs_to_group("Resellers") and \
not account.belongs_to_group("DomainAdmins"):
if account.group not in ["Resellers", "DomainAdmins"]:
return
instances["resources"] = account
@events.observe("GetStaticContent")
def get_static_content(caller, st_type, user):
if parameters.get_admin("ENABLE_ADMIN_LIMITS") == "no":
return []
if caller not in ['domains', 'identities']:
return []
if user.group == "SimpleUsers":
modoboa/limits/handlers.py
+30 -7
View File
@@ -5,6 +5,7 @@ from django.dispatch import receiver
from django.contrib.contenttypes.models import ContentType
from modoboa.admin import models as admin_models
from modoboa.core import models as core_models
from modoboa.core import signals as core_signals
from modoboa.lib import signals as lib_signals
@@ -16,11 +17,20 @@ from . import utils
@receiver(core_signals.can_create_object)
def check_object_limit(sender, user, object_type, **kwargs):
def check_object_limit(sender, context, object_type, **kwargs):
"""Check if user can create a new object."""
if user.is_superuser:
return True
limit = user.objectlimit_set.get(name=object_type)
if context.__class__.__name__ == "User":
if parameters.get_admin("ENABLE_ADMIN_LIMITS") == "no":
return
if context.is_superuser:
return True
limit = context.userobjectlimit_set.get(name=object_type)
elif context.__class__.__name__ == "Domain":
if parameters.get_admin("ENABLE_DOMAIN_LIMITS") == "no":
return
limit = context.domainobjectlimit_set.get(name=object_type)
else:
raise NotImplementedError
count = kwargs.get("count", 1)
if limit.is_exceeded(count):
raise lib.LimitReached(limit)
@@ -33,13 +43,26 @@ def create_user_limits(sender, instance, **kwargs):
return
request = lib_signals.get_request()
creator = request.user if request else None
for name, definition in utils.get_limit_templates():
for name, definition in utils.get_user_limit_templates():
ct = ContentType.objects.get_by_natural_key(
*definition["content_type"].split("."))
max_value = 0
# creator can be None if user was created by a factory
if not creator or creator.is_superuser:
max_value = int(
parameters.get_admin("DEFLT_{}_LIMIT".format(name.upper())))
models.ObjectLimit.objects.create(
parameters.get_admin(
"DEFLT_USER_{}_LIMIT".format(name.upper())))
models.UserObjectLimit.objects.create(
user=instance, name=name, content_type=ct, max_value=max_value)
@receiver(signals.post_save, sender=admin_models.Domain)
def create_domain_limits(sender, instance, **kwargs):
"""Create limits for new domain."""
if not kwargs.get("created"):
return
for name, definition in utils.get_domain_limit_templates():
max_value = int(
parameters.get_admin("DEFLT_DOMAIN_{}_LIMIT".format(name.upper())))
models.DomainObjectLimit.objects.create(
domain=instance, name=name, max_value=max_value)
modoboa/limits/migrations/0003_auto_20160413_1046.py
+5 -5
View File
@@ -8,7 +8,7 @@ from django.conf import settings
def move_limits_to_user(apps, schema_editor):
"""Move limits from Pool to User."""
LimitsPool = apps.get_model("limits", "LimitsPool")
ObjectLimit = apps.get_model("limits", "ObjectLimit")
UserObjectLimit = apps.get_model("limits", "UserObjectLimit")
ContentType = apps.get_model("contenttypes", "ContentType")
if not ContentType.objects.exists():
@@ -34,12 +34,12 @@ def move_limits_to_user(apps, schema_editor):
.prefetch_related("limit_set"))
for pool in qset:
for limit in pool.limit_set.all():
objlimit = ObjectLimit(
objlimit = UserObjectLimit(
user=pool.user, max_value=limit.maxvalue,
name=limit.name.replace("_limit", ""),
content_type=name_to_content_type[limit.name])
to_create.append(objlimit)
ObjectLimit.objects.bulk_create(to_create)
UserObjectLimit.objects.bulk_create(to_create)
class Migration(migrations.Migration):
@@ -53,7 +53,7 @@ class Migration(migrations.Migration):
operations = [
migrations.CreateModel(
name='ObjectLimit',
name='UserObjectLimit',
fields=[
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
('name', models.CharField(max_length=254)),
@@ -63,7 +63,7 @@ class Migration(migrations.Migration):
],
),
migrations.AlterUniqueTogether(
name='objectlimit',
name='userobjectlimit',
unique_together=set([('user', 'name')]),
),
migrations.RunPython(move_limits_to_user)
modoboa/limits/migrations/0005_auto_20160415_1654.py
+50
View File
@@ -0,0 +1,50 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import migrations, models
import modoboa.limits.models
from .. import utils
def create_limits(apps, schema_editor):
"""Create limits for existing domains."""
Domain = apps.get_model("admin", "Domain")
DomainObjectLimit = apps.get_model("limits", "DomainObjectLimit")
to_create = []
for domain in Domain.objects.all():
for name, tpl in utils.get_domain_limit_templates():
to_create.append(
DomainObjectLimit(domain=domain, name=name, max_value=-1))
DomainObjectLimit.objects.bulk_create(to_create)
class Migration(migrations.Migration):
dependencies = [
('admin', '0003_auto_20151118_1215'),
('limits', '0004_auto_20160413_1312'),
]
operations = [
migrations.CreateModel(
name='DomainObjectLimit',
fields=[
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
('name', models.CharField(max_length=254)),
('max_value', models.IntegerField(default=0)),
('domain', models.ForeignKey(to='admin.Domain')),
],
bases=(modoboa.limits.models.ObjectLimitMixin, models.Model),
),
migrations.AlterField(
model_name='userobjectlimit',
name='max_value',
field=models.IntegerField(default=0),
),
migrations.AlterUniqueTogether(
name='domainobjectlimit',
unique_together=set([('domain', 'name')]),
),
migrations.RunPython(create_limits)
]
modoboa/limits/models.py
+67 -11
View File
@@ -12,14 +12,42 @@ from modoboa.core import models as core_models
from . import utils
class ObjectLimitMixin(object):
"""Common methods."""
@property
def label(self):
"""Return display name."""
return self.definition["label"]
@property
def usage(self):
"""Return current limit usage in %."""
if self.max_value < 0:
return -1
if self.max_value == 0:
return 100
return int(float(self.current_value) / self.max_value * 100)
def is_exceeded(self, count=1):
"""Check if limit will be reached if we add count object(s)."""
return self.current_value + 1 > self.max_value
def __str__(self):
"""Display current usage."""
if self.max_value == -1:
return _("unlimited")
return "{}%".format(self.usage)
@python_2_unicode_compatible
class ObjectLimit(models.Model):
"""Per-user limits on object creation."""
class UserObjectLimit(ObjectLimitMixin, models.Model):
"""Object level limit."""
user = models.ForeignKey(settings.AUTH_USER_MODEL)
name = models.CharField(max_length=254)
content_type = models.ForeignKey("contenttypes.ContentType")
max_value = models.IntegerField(default=-2)
max_value = models.IntegerField(default=0)
class Meta:
unique_together = (("user", "name"), )
@@ -27,7 +55,7 @@ class ObjectLimit(models.Model):
@property
def definition(self):
"""Return the definition of this limit."""
for name, tpl in utils.get_limit_templates():
for name, tpl in utils.get_user_limit_templates():
if name == self.name:
return tpl
return None
@@ -46,6 +74,11 @@ class ObjectLimit(models.Model):
return model_class.objects.filter(
pk__in=id_list, **self.definition["extra_filters"]).count()
@property
def label(self):
"""Return display name."""
return self.definition["label"]
@property
def usage(self):
"""Return current limit usage in %."""
@@ -55,19 +88,42 @@ class ObjectLimit(models.Model):
return 100
return int(float(self.current_value) / self.max_value * 100)
@property
def label(self):
"""Return display name."""
return self.definition["label"]
def is_exceeded(self, count=1):
"""Check if limit will be reached if we add count object(s)."""
return self.current_value + 1 > self.max_value
def __str__(self):
"""Display current usage."""
if self.max_value == -2:
return _("undefined")
if self.max_value == -1:
return _("unlimited")
return "{}%".format(self.usage)
class DomainObjectLimit(ObjectLimitMixin, models.Model):
"""Per-domain limits on object creation."""
domain = models.ForeignKey("admin.Domain")
name = models.CharField(max_length=254)
max_value = models.IntegerField(default=0)
class Meta:
unique_together = (("domain", "name"), )
@property
def definition(self):
"""Return the definition of this limit."""
for name, tpl in utils.get_domain_limit_templates():
if name == self.name:
return tpl
return None
@property
def current_value(self):
"""Return the current number of objects."""
definition = self.definition
if not definition:
raise RuntimeError("Bad limit {}".format(self.name))
relation = getattr(self.domain, definition["relation"])
if "extra_filters" in definition:
relation = relation.filter(**definition["extra_filters"])
return relation.count()
modoboa/limits/tests/__init__.py
View File
modoboa/limits/tests/test_api.py
+237
View File
@@ -0,0 +1,237 @@
# coding: utf-8
"""Test cases for the limits extension."""
from django.core.urlresolvers import reverse
from rest_framework.authtoken.models import Token
from modoboa.admin.factories import populate_database
from modoboa.admin.models import Domain
from modoboa.core import factories as core_factories
from modoboa.core.models import User
from modoboa.lib import parameters
from modoboa.lib import tests as lib_tests
from .. import utils
class APIAdminLimitsTestCase(lib_tests.ModoAPITestCase):
"""Check that limits are used also by the API."""
@classmethod
def setUpTestData(cls):
"""Create test data."""
super(APIAdminLimitsTestCase, cls).setUpTestData()
for name, tpl in utils.get_user_limit_templates():
parameters.save_admin(
"DEFLT_USER_{}_LIMIT".format(name.upper()), 2)
populate_database()
cls.user = User.objects.get(username="admin@test.com")
cls.da_token = Token.objects.create(user=cls.user)
cls.reseller = core_factories.UserFactory(
username="reseller", groups=("Resellers", ),
)
cls.r_token = Token.objects.create(user=cls.reseller)
def test_domadmins_limit(self):
"""Check domain admins limit."""
self.client.credentials(
HTTP_AUTHORIZATION='Token ' + self.r_token.key)
limit = self.reseller.userobjectlimit_set.get(name="domain_admins")
url = reverse("external_api:account-list")
data = {
"username": "fromapi@test.com",
"role": "DomainAdmins",
"password": "Toto1234",
}
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
self.assertFalse(limit.is_exceeded())
data["username"] = "fromapi2@test.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
self.assertTrue(limit.is_exceeded())
data["username"] = "fromapi3@test.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
user = User.objects.get(username="user@test.com")
domain = Domain.objects.get(name="test.com")
domain.add_admin(self.reseller)
url = reverse("external_api:account-detail", args=[user.pk])
data = {
"username": user.username,
"role": "DomainAdmins",
"password": "Toto1234",
"mailbox": {
"full_address": user.username,
"quota": user.mailbox.quota
}
}
response = self.client.put(url, data, format="json")
self.assertEqual(response.status_code, 400)
def test_domains_limit(self):
"""Check domains limit."""
self.client.credentials(
HTTP_AUTHORIZATION='Token ' + self.r_token.key)
limit = self.reseller.userobjectlimit_set.get(name="domains")
url = reverse("external_api:domain-list")
data = {"name": "test3.com", "quota": 10}
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
self.assertFalse(limit.is_exceeded())
data["name"] = "test4.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
self.assertTrue(limit.is_exceeded())
data["username"] = "test5.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
def test_domain_aliases_limit(self):
"""Check domain aliases limit."""
self.client.credentials(
HTTP_AUTHORIZATION='Token ' + self.r_token.key)
domain = Domain.objects.get(name="test.com")
domain.add_admin(self.reseller)
limit = self.reseller.userobjectlimit_set.get(name="domain_aliases")
url = reverse("external_api:domain_alias-list")
data = {"name": "dalias1.com", "target": domain.pk}
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
self.assertFalse(limit.is_exceeded())
data["name"] = "dalias2.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
self.assertTrue(limit.is_exceeded())
data["username"] = "dalias3.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
def test_mailboxes_limit(self):
"""Check mailboxes limit."""
self.client.credentials(
HTTP_AUTHORIZATION='Token ' + self.da_token.key)
limit = self.user.userobjectlimit_set.get(name="mailboxes")
url = reverse("external_api:account-list")
data = {
"username": "fromapi@test.com",
"role": "SimpleUsers",
"password": "Toto1234",
"mailbox": {
"full_address": "fromapi@test.com",
"quota": 10
}
}
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
self.assertFalse(limit.is_exceeded())
data["username"] = "fromapi2@test.com"
data["mailbox"]["full_address"] = "fromapi2@test.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
self.assertTrue(limit.is_exceeded())
data["username"] = "fromapi3@test.com"
data["mailbox"]["full_address"] = "fromapi3@test.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
def test_aliases_limit(self):
"""Check mailbox aliases limit."""
self.client.credentials(
HTTP_AUTHORIZATION='Token ' + self.da_token.key)
limit = self.user.userobjectlimit_set.get(name="mailbox_aliases")
url = reverse("external_api:alias-list")
data = {
"address": "alias_fromapi@test.com",
"recipients": [
"user@test.com", "postmaster@test.com", "user_éé@nonlocal.com"
]
}
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
self.assertFalse(limit.is_exceeded())
data["address"] = "alias_fromapi2@test.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
self.assertTrue(limit.is_exceeded())
data["address"] = "alias_fromapi3@test.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
class APIDomainLimitsTestCase(lib_tests.ModoAPITestCase):
"""Check that limits are used also by the API."""
@classmethod
def setUpTestData(cls):
"""Create test data."""
super(APIDomainLimitsTestCase, cls).setUpTestData()
parameters.save_admin("ENABLE_DOMAIN_LIMITS", "yes")
for name, tpl in utils.get_domain_limit_templates():
parameters.save_admin(
"DEFLT_DOMAIN_{}_LIMIT".format(name.upper()), 2)
populate_database()
def test_mailboxes_limit(self):
"""Check mailboxes limit."""
domain = Domain.objects.get(name="test.com")
limit = domain.domainobjectlimit_set.get(name="mailboxes")
self.assertTrue(limit.is_exceeded())
url = reverse("external_api:account-list")
data = {
"username": "fromapi@test.com",
"role": "SimpleUsers",
"password": "Toto1234",
"mailbox": {
"full_address": "fromapi@test.com",
"quota": 10
}
}
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
def test_domain_aliases_limit(self):
"""Check domain_aliases limit."""
domain = Domain.objects.get(name="test.com")
limit = domain.domainobjectlimit_set.get(name="domain_aliases")
url = reverse("external_api:domain_alias-list")
data = {"name": "dalias1.com", "target": domain.pk}
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
data["name"] = "dalias2.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 201)
self.assertTrue(limit.is_exceeded())
data["name"] = "dalias3.com"
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
def test_mailbox_aliases_limit(self):
"""Check mailbox_aliases limit."""
domain = Domain.objects.get(name="test.com")
limit = domain.domainobjectlimit_set.get(name="mailbox_aliases")
self.assertTrue(limit.is_exceeded())
url = reverse("external_api:alias-list")
data = {
"address": "alias_fromapi@test.com",
"recipients": [
"user@test.com", "postmaster@test.com", "user_éé@nonlocal.com"
]
}
response = self.client.post(url, data, format="json")
self.assertEqual(response.status_code, 400)
modoboa/limits/tests/test_domain_limits.py
+160
View File
@@ -0,0 +1,160 @@
# coding: utf-8
"""Test cases for the limits extension."""
from django.core.urlresolvers import reverse
from modoboa.admin import factories as admin_factories
from modoboa.admin.models import Domain
from modoboa.core import factories as core_factories
from modoboa.core.models import User
from modoboa.lib import parameters
from modoboa.lib import tests as lib_tests
from .. import utils
class DomainLimitsTestCase(lib_tests.ModoTestCase):
"""Per-domain limits tests."""
@classmethod
def setUpTestData(cls):
"""Create test data."""
super(DomainLimitsTestCase, cls).setUpTestData()
parameters.save_admin("ENABLE_ADMIN_LIMITS", "no")
parameters.save_admin("ENABLE_DOMAIN_LIMITS", "yes")
for name, tpl in utils.get_domain_limit_templates():
parameters.save_admin(
"DEFLT_DOMAIN_{}_LIMIT".format(name.upper()), 2)
admin_factories.populate_database()
def test_set_limits(self):
"""Try to set limits for a given domain."""
domain = Domain.objects.get(name="test.com")
values = {
"name": domain.name, "quota": domain.quota,
"enabled": domain.enabled, "type": "domain",
"mailboxes_limit": 3, "mailbox_aliases_limit": 3,
"domain_aliases_limit": 3, "domain_admins_limit": 3
}
self.ajax_post(
reverse("admin:domain_change", args=[domain.id]),
values
)
domain.refresh_from_db()
self.assertEqual(
domain.domainobjectlimit_set.get(name="mailboxes").max_value, 3)
self.assertEqual(
domain.domainobjectlimit_set.get(
name="mailbox_aliases").max_value, 3)
self.assertEqual(
domain.domainobjectlimit_set.get(
name="domain_aliases").max_value, 3)
self.assertEqual(
domain.domainobjectlimit_set.get(
name="domain_admins").max_value, 3)
def test_domain_aliases_limit(self):
"""Try to exceed defined limit."""
domain = Domain.objects.get(name="test.com")
limit = domain.domainobjectlimit_set.get(name="domain_aliases")
self.assertFalse(limit.is_exceeded())
values = {
"name": domain.name, "quota": domain.quota,
"enabled": domain.enabled, "type": "domain",
"mailboxes_limit": 2, "mailbox_aliases_limit": 2,
"domain_aliases_limit": 2, "domain_admins_limit": 2,
"aliases": "alias1.com", "aliases_1": "alias2.com"
}
self.ajax_post(
reverse("admin:domain_change", args=[domain.id]),
values
)
self.assertTrue(limit.is_exceeded())
values["aliases_2"] = "alias3.com"
self.ajax_post(
reverse("admin:domain_change", args=[domain.id]),
values, 403
)
def test_domain_admins_limit(self):
"""Try to exceed defined limit."""
domain = Domain.objects.get(name="test.com")
limit = domain.domainobjectlimit_set.get(name="domain_admins")
self.assertFalse(limit.is_exceeded())
user = User.objects.get(username="admin@test2.com")
values = {
"username": user.username, "role": user.group,
"is_active": user.is_active, "email": user.email,
"quota_act": True, "domains": "test2.com",
"domains_1": "test.com"
}
self.ajax_post(
reverse("admin:account_change", args=[user.id]),
values
)
self.assertTrue(limit.is_exceeded())
user = core_factories.UserFactory(
username="admin1000@test.com", groups=("DomainAdmins", ))
self.ajax_post(
reverse("admin:account_change", args=[user.id]),
values, 400
)
def test_mailboxes_limit(self):
"""Try to exceed defined limits."""
domain = Domain.objects.get(name="test.com")
domain.domainobjectlimit_set.filter(name="mailboxes").update(
max_value=3)
limit = domain.domainobjectlimit_set.get(name="mailboxes")
self.assertFalse(limit.is_exceeded())
username = "toto@test.com"
values = {
"username": username,
"first_name": "Tester", "last_name": "Toto",
"password1": "Toto1234", "password2": "Toto1234",
"role": "SimpleUsers", "quota_act": True,
"is_active": True, "email": username, "stepid": "step2",
}
self.ajax_post(reverse("admin:account_add"), values, 200)
self.assertTrue(limit.is_exceeded())
username = "titi@test.com"
self.ajax_post(reverse("admin:account_add"), values, 400)
def test_mailbox_aliases_limit(self):
"""Try to exceed defined limits."""
domain = Domain.objects.get(name="test.com")
user = User.objects.get(username="user@test.com")
limit = domain.domainobjectlimit_set.get(name="mailbox_aliases")
limit.max_value = 4
limit.save()
self.assertFalse(limit.is_exceeded())
values = {
"username": user.username, "role": user.group,
"is_active": user.is_active, "email": user.email,
"quota_act": True,
"aliases": "alias@test.com", "aliases_1": "alias1@test.com"
}
self.ajax_post(
reverse("admin:account_change", args=[user.id]),
values
)
self.assertTrue(limit.is_exceeded())
values["aliases_2"] = "alias2@test.com"
self.ajax_post(
reverse("admin:account_change", args=[user.id]),
values, 403
)
limit.max_value = 5
limit.save()
values = {
"address": "forward2@test.com", "recipients": "user@test.com",
"enabled": True
}
self.ajax_post(reverse("admin:alias_add"), values)
self.assertTrue(limit.is_exceeded())
values["address"] = "forward3@test.com"
self.ajax_post(reverse("admin:alias_add"), values, 400)
modoboa/limits/tests/test_import.py
+221
View File
@@ -0,0 +1,221 @@
# coding: utf-8
from django.core.files.base import ContentFile
from django.core.urlresolvers import reverse
from modoboa.admin import factories as admin_factories
from modoboa.admin import models as admin_models
from modoboa.core import factories as core_factories
from modoboa.core import models as core_models
from modoboa.lib import parameters
from modoboa.lib.tests import ModoTestCase
from .. import utils
class LimitImportTestCase(ModoTestCase):
"""Base class to test limits."""
@classmethod
def setUpTestData(cls):
"""Create test data."""
super(LimitImportTestCase, cls).setUpTestData()
admin_factories.populate_database()
cls.reseller = core_factories.UserFactory(
username="reseller", groups=("Resellers", ))
cls.dadmin = core_models.User.objects.get(username="admin@test.com")
cls.domain = admin_models.Domain.objects.get(name="test.com")
def _test_domain_alias_import(self, limit):
"""Check domain aliases limit."""
self.client.login(
username=self.reseller.username, password="toto")
self.assertFalse(limit.is_exceeded())
f = ContentFile(b"""domainalias; domalias1.com; {domain}; True
domainalias; domalias2.com; {domain}; True
""".format(domain=self.domain), name="domains.csv")
self.client.post(
reverse("admin:domain_import"), {
"sourcefile": f
}
)
self.assertTrue(limit.is_exceeded())
f = ContentFile(
"domainalias; domalias3.com; {}; True".format(self.domain),
name="domains.csv")
response = self.client.post(
reverse("admin:domain_import"), {
"sourcefile": f
}
)
self.assertContains(response, "Domain aliases: limit reached")
def _test_domain_admins_import(self, limit, initial_count=0):
"""Check domain admins limit."""
self.client.login(
username=self.reseller.username, password="toto")
self.assertFalse(limit.is_exceeded())
content = ""
for cpt in xrange(initial_count, 2):
content += (
"account; admin{cpt}@{domain}; toto; User; One; True; "
"DomainAdmins; user{cpt}@{domain}; 5; {domain}\n"
.format(domain=self.domain, cpt=cpt)
)
f = ContentFile(content, name="domain_admins.csv")
response = self.client.post(
reverse("admin:identity_import"), {
"sourcefile": f
}
)
self.assertTrue(limit.is_exceeded())
f = ContentFile(b"""account; admin3@{domain}; toto; User; One; True; DomainAdmins; admin3@{domain}; 5; {domain}
""".format(domain=self.domain), name="domain_admins.csv")
response = self.client.post(
reverse("admin:identity_import"), {
"sourcefile": f
}
)
self.assertContains(response, "Permission denied")
def _test_mailboxes_import(self, limit):
"""Check mailboxes limit."""
self.client.login(
username=self.dadmin.username, password="toto")
self.assertFalse(limit.is_exceeded())
f = ContentFile(b"""account; user1@{domain}; toto; User; One; True; SimpleUsers; user1@{domain}; 5
account; truc@{domain}; toto; René; Truc; True; SimpleUsers; truc@{domain}; 5
""".format(domain=self.domain), name="domains.csv")
response = self.client.post(
reverse("admin:identity_import"), {
"sourcefile": f
}
)
self.assertTrue(limit.is_exceeded())
f = ContentFile("""
account; user3@{domain}; toto; User; One; True; SimpleUsers; user3@{domain}; 5
""".format(domain=self.domain), name="domains.csv")
response = self.client.post(
reverse("admin:identity_import"), {
"sourcefile": f
}
)
self.assertContains(response, "Mailboxes: limit reached")
def _test_mailbox_aliases_import(self, limit):
"""Check mailbox aliases limit."""
self.client.login(
username=self.dadmin.username, password="toto")
self.assertFalse(limit.is_exceeded())
f = ContentFile(b"""alias; alias1@{domain}; True; user@{domain}
alias; alias2@{domain}; True; user@{domain}
""".format(domain=self.domain), name="aliases.csv")
response = self.client.post(
reverse("admin:identity_import"), {
"sourcefile": f
}
)
self.assertTrue(limit.is_exceeded())
f = ContentFile("""
alias; alias3@{domain}; True; user@{domain}
""".format(domain=self.domain), name="aliases.csv")
response = self.client.post(
reverse("admin:identity_import"), {
"sourcefile": f
}
)
self.assertContains(response, "Mailbox aliases: limit reached")
class UserLimitImportTestCase(LimitImportTestCase):
@classmethod
def setUpTestData(cls):
"""Create test data."""
for name, tpl in utils.get_user_limit_templates():
parameters.save_admin(
"DEFLT_USER_{}_LIMIT".format(name.upper()), 2)
super(UserLimitImportTestCase, cls).setUpTestData()
def test_domains_import(self):
"""Check domains limit."""
self.client.login(
username=self.reseller.username, password="toto")
limit = self.reseller.userobjectlimit_set.get(name="domains")
self.assertFalse(limit.is_exceeded())
f = ContentFile(b"""domain; domain1.com; 100; True
domain; domain2.com; 200; False
""", name="domains.csv")
self.client.post(
reverse("admin:domain_import"), {
"sourcefile": f
}
)
self.assertTrue(limit.is_exceeded())
f = ContentFile(b"""domain; domain3.com; 100; True
""", name="domains.csv")
response = self.client.post(
reverse("admin:domain_import"), {
"sourcefile": f
}
)
self.assertContains(response, "Domains: limit reached")
def test_domain_alias_import(self):
"""Check domain aliases limit."""
limit = self.reseller.userobjectlimit_set.get(name="domain_aliases")
self._test_domain_alias_import(limit)
def test_domain_admins_import(self):
"""Check domain admins limit."""
self.domain.add_admin(self.reseller)
limit = self.reseller.userobjectlimit_set.get(name="domain_admins")
self._test_domain_admins_import(limit)
def test_mailboxes_import(self):
"""Check mailboxes limit."""
limit = self.dadmin.userobjectlimit_set.get(name="mailboxes")
self._test_mailboxes_import(limit)
def test_mailbox_aliases_import(self):
"""Check mailbox aliases limit."""
limit = self.dadmin.userobjectlimit_set.get(name="mailbox_aliases")
self._test_mailbox_aliases_import(limit)
class DomainLimitImportTestCase(LimitImportTestCase):
@classmethod
def setUpTestData(cls):
"""Create test data."""
parameters.save_admin("ENABLE_DOMAIN_LIMITS", "yes")
parameters.save_admin("ENABLE_ADMIN_LIMITS", "no")
for name, tpl in utils.get_domain_limit_templates():
parameters.save_admin(
"DEFLT_DOMAIN_{}_LIMIT".format(name.upper()), 2)
super(DomainLimitImportTestCase, cls).setUpTestData()
mb = admin_factories.MailboxFactory(
user__username="user@test4.com",
domain__name="test4.com", address="user")
cls.domain = mb.domain
def test_domain_alias_import(self):
"""Check domain aliases limit."""
limit = self.domain.domainobjectlimit_set.get(name="domain_aliases")
self._test_domain_alias_import(limit)
def test_mailboxes_import(self):
"""Check mailboxes limit."""
self.domain.add_admin(self.dadmin)
limit = self.domain.domainobjectlimit_set.get(name="mailboxes")
self._test_mailboxes_import(limit)
def test_mailbox_aliases_import(self):
"""Check mailbox aliases limit."""
self.domain.add_admin(self.dadmin)
limit = self.domain.domainobjectlimit_set.get(name="mailbox_aliases")
self._test_mailbox_aliases_import(limit)
modoboa/limits/tests.py → modoboa/limits/tests/test_user_limits.py
+14 -11
View File
@@ -1,3 +1,4 @@
# coding: utf-8
"""Test cases for the limits extension."""
from django.core.urlresolvers import reverse
@@ -7,12 +8,12 @@ from modoboa.admin.models import Alias, Domain
from modoboa.core.factories import UserFactory
from modoboa.core.models import User
from modoboa.lib import parameters
from modoboa.lib.tests import ModoTestCase
from modoboa.lib import tests as lib_tests
from . import utils
from .. import utils
class PermissionsTestCase(ModoTestCase):
class PermissionsTestCase(lib_tests.ModoTestCase):
@classmethod
def setUpTestData(cls):
@@ -41,15 +42,15 @@ class PermissionsTestCase(ModoTestCase):
self.assertEqual(resp, "Permission denied")
class ResourceTestCase(ModoTestCase):
class ResourceTestCase(lib_tests.ModoTestCase):
@classmethod
def setUpTestData(cls):
"""Custom setUpTestData method."""
super(ResourceTestCase, cls).setUpTestData()
for name, tpl in utils.get_limit_templates():
for name, tpl in utils.get_user_limit_templates():
parameters.save_admin(
"DEFLT_{0}_LIMIT".format(name.upper()), 2,
"DEFLT_USER_{0}_LIMIT".format(name.upper()), 2,
app="limits"
)
populate_database()
@@ -107,7 +108,7 @@ class ResourceTestCase(ModoTestCase):
)
def _check_limit(self, name, curvalue, maxvalue):
l = self.user.objectlimit_set.get(name=name)
l = self.user.userobjectlimit_set.get(name=name)
self.assertEqual(l.current_value, curvalue)
self.assertEqual(l.max_value, maxvalue)
@@ -119,7 +120,7 @@ class DomainAdminTestCase(ResourceTestCase):
"""Create test data."""
super(DomainAdminTestCase, cls).setUpTestData()
cls.user = User.objects.get(username='admin@test.com')
cls.user.objectlimit_set.filter(
cls.user.userobjectlimit_set.filter(
name__in=["mailboxes", "mailbox_aliases"]).update(max_value=2)
def setUp(self):
@@ -240,7 +241,8 @@ class ResellerTestCase(ResourceTestCase):
)
self._check_limit('domain_admins', 2, 2)
self.user.objectlimit_set.filter(name="mailboxes").update(max_value=3)
self.user.userobjectlimit_set.filter(
name="mailboxes").update(max_value=3)
self._create_account('user1@domain.tld')
user = User.objects.get(username='user1@domain.tld')
values = {
@@ -265,11 +267,12 @@ class ResellerTestCase(ResourceTestCase):
domadmin = User.objects.get(username='admin1@domain.tld')
for l in ['mailboxes', 'mailbox_aliases']:
self.assertEqual(
domadmin.objectlimit_set.get(name=l).max_value, 0
domadmin.userobjectlimit_set.get(name=l).max_value, 0
)
def test_domain_admins_limit_from_domain_tpl(self):
self.user.objectlimit_set.filter(name="domains").update(max_value=3)
self.user.userobjectlimit_set.filter(
name="domains").update(max_value=3)
self._create_domain('domain1.tld', withtpl=True)
self._create_domain('domain2.tld', withtpl=True)
self._check_limit('domain_admins', 2, 2)
modoboa/limits/utils.py
+7 -2
View File
@@ -3,7 +3,12 @@
from . import constants
def get_limit_templates():
def get_user_limit_templates():
"""Return defined templates."""
# FIXME: change GetExtraLimitTemplates
return constants.DEFAULT_LIMITS.items()
return constants.DEFAULT_USER_LIMITS.items()
def get_domain_limit_templates():
"""Return defined templates."""
return constants.DEFAULT_DOMAIN_LIMITS.items()
modoboa/relaydomains/tests.py
+3 -3
View File
@@ -50,7 +50,7 @@ class Operations(object):
)
def _check_limit(self, name, curvalue, maxvalue):
l = self.user.objectlimit_set.get(name=name)
l = self.user.userobjectlimit_set.get(name=name)
self.assertEqual(l.current_value, curvalue)
self.assertEqual(l.max_value, maxvalue)
@@ -233,9 +233,9 @@ class LimitsTestCase(ModoTestCase, Operations):
"""Create test data."""
super(LimitsTestCase, cls).setUpTestData()
for name, tpl in limits_utils.get_limit_templates():
for name, tpl in limits_utils.get_user_limit_templates():
parameters.save_admin(
"DEFLT_{0}_LIMIT".format(name.upper()), 2, app="limits"
"DEFLT_USER_{0}_LIMIT".format(name.upper()), 2, app="limits"
)
cls.user = UserFactory.create(
username='reseller', groups=('Resellers',)