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

New tests fixed

Browse Source
This commit is contained in:
Antoine Nguyen
2025-06-08 15:12:48 +02:00
parent 5b1dd9fbed
commit defef9c15f
17 changed files with 483 additions and 747 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modoboa/admin/api/v1/serializers.py
+10 -2
View File
@@ -85,6 +85,10 @@ class DomainSerializer(serializers.ModelSerializer):
def validate_name(self, value):
"""Check name constraints."""
if models.DomainAlias.objects.filter(name=value).exists():
raise serializers.ValidationError(
_("domain alias with this name already exists")
)
domains_must_have_authorized_mx = param_tools.get_global_parameter(
"domains_must_have_authorized_mx"
)
@@ -439,7 +443,11 @@ class WritableAccountSerializer(AccountSerializer):
mb = admin_models.Mailbox(
user=account, address=self.address, domain=self.domain, **data
)
mb.set_quota(quota, creator.has_perm("admin.add_domain"))
override_rules = (
creator.has_perm("admin.add_domain")
and not creator.userobjectlimit_set.get(name="quota").max_value
)
mb.set_quota(quota, override_rules)
default_msg_limit = param_tools.get_global_parameter(
"default_mailbox_message_limit"
)
@@ -527,7 +535,7 @@ class AliasSerializer(serializers.ModelSerializer):
value, self.context["request"].user, instance=self.instance
)
except ValidationError as err:
raise serializers.ValidationError(err) from None
raise serializers.ValidationError(err.message) from None
except AliasExists:
raise serializers.ValidationError(_("This alias already exists")) from None
return value.lower()
modoboa/admin/api/v2/serializers.py
+29 -6
View File
@@ -21,10 +21,12 @@ from modoboa.lib import email_utils
from modoboa.lib import exceptions as lib_exceptions
from modoboa.lib import fields as lib_fields
from modoboa.lib import validators, web_utils
from modoboa.lib.permissions import get_object_owner
from modoboa.lib.sysutils import exec_cmd
from modoboa.parameters import tools as param_tools
from modoboa.limits import constants as limits_constants
from modoboa.limits import models as limits_models
from modoboa.limits.lib import allocate_resources_from_user
from modoboa.transport import models as transport_models
from modoboa.transport.api.v2 import serializers as transport_serializers
@@ -76,6 +78,21 @@ class DomainSerializer(v1_serializers.DomainSerializer):
raise serializers.ValidationError(
{"transport": _("This field is required")}
)
elif dtype == "domain" and data.get("transport"):
raise serializers.ValidationError(
{"transport": _("This field is valid when type is relaydomain")}
)
user = self.context["request"].user
if user.role == "Resellers":
limit = user.userobjectlimit_set.get(name="quota")
if limit.max_value != 0:
quota = data["quota"]
msg = _("You can't define an unlimited quota.")
if quota == 0:
raise serializers.ValidationError({"quota": msg})
default_mailbox_quota = data["default_mailbox_quota"]
if default_mailbox_quota == 0:
raise serializers.ValidationError({"default_mailbox_quota": msg})
return result
def create(self, validated_data):
@@ -89,10 +106,14 @@ class DomainSerializer(v1_serializers.DomainSerializer):
transport.save()
domain.transport = transport
domain.save()
if not domain_admin:
return domain
# 0. Check user limit
user = self.context["request"].user
core_signals.can_create_object.send(
self.__class__, context=user, object_type="domain_admins"
)
# 1. Create a domain administrator
username = "{}@{}".format(domain_admin["username"], domain.name)
try:
@@ -101,7 +122,6 @@ class DomainSerializer(v1_serializers.DomainSerializer):
pass
else:
raise lib_exceptions.Conflict(_("User '%s' already exists") % username)
user = self.context["request"].user
core_signals.can_create_object.send(
self.__class__, context=user, klass=models.Mailbox
)
@@ -150,7 +170,7 @@ class DomainSerializer(v1_serializers.DomainSerializer):
"""Update domain and create/update transport."""
transport_def = validated_data.pop("transport", None)
resources = validated_data.pop("domainobjectlimit_set", None)
super().update(instance, validated_data)
instance = super().update(instance, validated_data)
if transport_def and instance.type == "relaydomain":
transport = getattr(instance, "transport", None)
created = False
@@ -618,10 +638,13 @@ class WritableAccountSerializer(v1_serializers.WritableAccountSerializer):
instance.save()
resources = validated_data.get("resources")
if resources:
owner = get_object_owner(instance)
for resource in resources:
instance.userobjectlimit_set.filter(name=resource["name"]).update(
max_value=resource["max_value"]
)
limit = instance.userobjectlimit_set.get(name=resource["name"])
if not owner.is_superuser:
allocate_resources_from_user(limit, owner, resource["max_value"])
limit.max_value = resource["max_value"]
limit.save()
self.set_permissions(instance, domains)
return instance
modoboa/admin/models/domain.py
+1
View File
@@ -293,6 +293,7 @@ class Domain(mixins.MessageLimitMixin, AdminObject):
"""
from modoboa.lib.permissions import grant_access_to_object
# Check if domain's limit for domain admins is not reached yet
core_signals.can_create_object.send(
sender=self.__class__, context=self, object_type="domain_admins"
)
modoboa/admin/models/mailbox.py
+1 -1
View File
@@ -317,7 +317,7 @@ class Mailbox(mixins.MessageLimitMixin, AdminObject):
or user.has_perm("admin.add_domain")
and not user.userobjectlimit_set.get(name="quota").max_value
)
self.set_quota(values["quota"], override_rules)
self.set_quota(values.get("quota"), override_rules)
if "message_limit" in values:
self.message_limit = values["message_limit"]
if "is_send_only" in values:
modoboa/lib/tests/__init__.py
+12 -4
View File
@@ -103,17 +103,25 @@ class ModoAPITestCase(ParametersMixin, APITestCase):
"""Create a default user."""
super().setUpTestData()
management.call_command("load_initial_data")
cls.token = Token.objects.create(
user=core_models.User.objects.get(username="admin")
)
cls.sadmin = core_models.User.objects.get(username="admin")
cls.token = Token.objects.create(user=cls.sadmin)
def setUp(self):
"""Setup."""
super().setUp()
self.client.credentials(HTTP_AUTHORIZATION="Token " + self.token.key)
self._tokens = {self.sadmin: self.token}
self.authenticate_user(self.sadmin)
self.workdir = tempfile.mkdtemp()
self.set_global_parameter("storage_dir", self.workdir, app="pdfcredentials")
def authenticate_user(self, user: core_models.User) -> Token:
if user not in self._tokens:
self._tokens[user] = Token.objects.create(user=user)
token = self._tokens[user]
self.client.credentials(HTTP_AUTHORIZATION=f"Token {token.key}")
return token
def create_session(self):
"""Enable session storage across requests."""
session_engine = import_module(settings.SESSION_ENGINE)
modoboa/limits/handlers.py
+5 -2
View File
@@ -20,8 +20,11 @@ def check_object_limit(sender, context, **kwargs):
# FIXME: Useless?
if context.is_superuser:
return True
ct = ContentType.objects.get_for_model(kwargs.get("klass"))
limits = context.userobjectlimit_set.filter(content_type=ct)
if "klass" in kwargs:
ct = ContentType.objects.get_for_model(kwargs.get("klass"))
limits = context.userobjectlimit_set.filter(content_type=ct)
else:
limits = context.userobjectlimit_set.filter(name=kwargs["object_type"])
elif context.__class__.__name__ == "Domain":
if not param_tools.get_global_parameter("enable_domain_limits"):
return
modoboa/limits/tests/test_domain_limits.py
+51 -80
View File
@@ -7,16 +7,17 @@ from modoboa.admin.models import Domain
from modoboa.core import factories as core_factories
from modoboa.core.models import User
from modoboa.lib import tests as lib_tests
from modoboa.limits.tests.test_user_limits import ResourceTestCase
from .. import utils
class DomainLimitsTestCase(lib_tests.ModoTestCase):
class DomainLimitsTestCase(ResourceTestCase):
"""Per-domain limits tests."""
@classmethod
def setUpTestData(cls): # NOQA:N802
"""Create test data."""
super().setUpTestData()
lib_tests.ModoAPITestCase.setUpTestData()
cls.localconfig.parameters.set_values(
{"enable_admin_limits": False, "enable_domain_limits": True}
)
@@ -34,12 +35,17 @@ class DomainLimitsTestCase(lib_tests.ModoTestCase):
"default_mailbox_quota": domain.default_mailbox_quota,
"enabled": domain.enabled,
"type": "domain",
"mailboxes_limit": 3,
"mailbox_aliases_limit": 3,
"domain_aliases_limit": 3,
"domain_admins_limit": 3,
"resources": [
{"name": "mailboxes", "max_value": 3},
{"name": "mailbox_aliases", "max_value": 3},
{"name": "domain_aliases", "max_value": 3},
{"name": "domain_admins", "max_value": 3},
],
}
self.ajax_post(reverse("admin:domain_change", args=[domain.id]), values)
response = self.client.put(
reverse("v2:domain-detail", args=[domain.id]), values, format="json"
)
self.assertEqual(response.status_code, 200)
domain.refresh_from_db()
self.assertEqual(
domain.domainobjectlimit_set.get(name="mailboxes").max_value, 3
@@ -59,23 +65,19 @@ class DomainLimitsTestCase(lib_tests.ModoTestCase):
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,
"default_mailbox_quota": domain.default_mailbox_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)
url = reverse("v2:domain_alias-list")
body = {"name": "domainalias1.net", "target": domain.id, "enabled": True}
response = self.client.post(url, body, format="json")
self.assertEqual(response.status_code, 201)
body["name"] = "domainalias2.net"
response = self.client.post(url, body, format="json")
self.assertEqual(response.status_code, 201)
self.assertTrue(limit.is_exceeded())
values["aliases_2"] = "alias3.com"
self.ajax_post(reverse("admin:domain_change", args=[domain.id]), values, 403)
body["name"] = "domainalias3.net"
response = self.client.post(url, body, format="json")
self.assertEqual(response.status_code, 400)
self.assertEqual(response.json()["domain"], "Domain aliases: limit reached")
def test_domain_admins_limit(self):
"""Try to exceed defined limit."""
@@ -83,24 +85,22 @@ class DomainLimitsTestCase(lib_tests.ModoTestCase):
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.role,
"is_active": user.is_active,
"email": user.email,
"quota_act": True,
"domains": "test2.com",
"domains_1": "test.com",
"language": "en",
"subject": "subject",
"content": "content",
}
self.ajax_post(reverse("admin:account_change", args=[user.id]), values)
response = self.client.post(
reverse("v2:domain-add-administrator", args=[domain.id]),
{"account": user.id},
format="json",
)
self.assertEqual(response.status_code, 200)
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)
response = self.client.post(
reverse("v2:domain-add-administrator", args=[domain.id]),
{"account": user.id},
format="json",
)
self.assertEqual(response.status_code, 403)
def test_mailboxes_limit(self):
"""Try to exceed defined limits."""
@@ -108,67 +108,38 @@ class DomainLimitsTestCase(lib_tests.ModoTestCase):
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": "toto@test.com",
"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())
values["username"] = "titi@test.com"
values["email"] = "titi@test.com"
self.ajax_post(reverse("admin:account_add"), values, 400)
# Check if defined limit can't be exceeded
username = "toto@test.com"
self._create_account(username)
self.assertTrue(limit.is_exceeded())
self._create_account("titi@test.com", status=400)
# Set unlimited value
limit.max_value = -1
limit.save(update_fields=["max_value"])
self.ajax_post(reverse("admin:account_add"), values)
self._create_account("titi@test.com")
self.assertFalse(limit.is_exceeded())
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.role,
"is_active": user.is_active,
"email": user.email,
"quota_act": True,
"aliases": "alias@test.com",
"aliases_1": "alias1@test.com",
"language": "en",
"subject": "subject",
"content": "content",
}
self.ajax_post(reverse("admin:account_change", args=[user.id]), values)
self._create_alias("alias1@test.com", ["user@test.com"])
self.assertTrue(limit.is_exceeded())
values["aliases_2"] = "alias2@test.com"
self.ajax_post(reverse("admin:account_change", args=[user.id]), values, 403)
response = self._create_alias("alias2@test.com", ["user@test.com"], status=400)
self.assertEqual(
response.json()["address"][0], "Mailbox aliases: limit reached"
)
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._create_alias("forward2@test.com", ["user@test.com"])
self.assertTrue(limit.is_exceeded())
values["address"] = "forward3@test.com"
self.ajax_post(reverse("admin:alias_add"), values, 400)
self._create_alias("forward3@test.com", ["user@test.com"], status=400)
modoboa/limits/tests/test_import.py
+41 -22
View File
@@ -4,11 +4,11 @@ from django.urls import reverse
from modoboa.admin import factories as admin_factories, models as admin_models
from modoboa.core import factories as core_factories, models as core_models
from modoboa.lib.tests import ModoTestCase
from modoboa.lib.tests import ModoAPITestCase
from .. import utils
class LimitImportTestCase(ModoTestCase):
class LimitImportTestCase(ModoAPITestCase):
"""Base class to test limits."""
@classmethod
@@ -24,7 +24,8 @@ class LimitImportTestCase(ModoTestCase):
def _test_domain_alias_import(self, limit):
"""Check domain aliases limit."""
self.client.force_login(self.reseller)
self.authenticate_user(self.reseller)
self.assertFalse(limit.is_exceeded())
f = ContentFile(
f"""domainalias; domalias1.com; {self.domain}; True
@@ -32,19 +33,24 @@ domainalias; domalias2.com; {self.domain}; True
""",
name="domains.csv",
)
self.client.post(reverse("admin:domain_import"), {"sourcefile": f})
url = reverse("v2:domain-import-from-csv")
response = self.client.post(url, {"sourcefile": f})
self.assertEqual(response.status_code, 200)
self.assertTrue(limit.is_exceeded())
f = ContentFile(
f"domainalias; domalias3.com; {self.domain}; True",
name="domains.csv",
)
response = self.client.post(reverse("admin:domain_import"), {"sourcefile": f})
self.assertContains(response, "Domain aliases: limit reached")
response = self.client.post(url, {"sourcefile": f})
self.assertEqual(response.status_code, 200)
result = response.json()
self.assertFalse(result["status"])
self.assertEqual(result["message"], "Domain aliases: limit reached")
def _test_domain_admins_import(self, limit, initial_count=0):
"""Check domain admins limit."""
self.client.force_login(self.reseller)
self.authenticate_user(self.reseller)
self.assertFalse(limit.is_exceeded())
content = ""
for cpt in range(initial_count, 2):
@@ -53,7 +59,9 @@ domainalias; domalias2.com; {self.domain}; True
f"DomainAdmins; user{cpt}@{self.domain}; 5; {self.domain}\n"
)
f = ContentFile(content, name="domain_admins.csv")
response = self.client.post(reverse("admin:identity_import"), {"sourcefile": f})
url = reverse("v2:identities-import-from-csv")
response = self.client.post(url, {"sourcefile": f})
self.assertEqual(response.status_code, 200)
self.assertTrue(limit.is_exceeded())
f = ContentFile(
@@ -61,12 +69,15 @@ domainalias; domalias2.com; {self.domain}; True
""",
name="domain_admins.csv",
) # NOQA:E501
response = self.client.post(reverse("admin:identity_import"), {"sourcefile": f})
self.assertContains(response, "Permission denied")
response = self.client.post(url, {"sourcefile": f})
self.assertEqual(response.status_code, 200)
result = response.json()
self.assertFalse(result["status"])
def _test_mailboxes_import(self, limit):
"""Check mailboxes limit."""
self.client.force_login(self.dadmin)
self.authenticate_user(self.dadmin)
self.assertFalse(limit.is_exceeded())
f = ContentFile(
f"account; user1@{self.domain}; toto; User; One; True; SimpleUsers; "
@@ -75,7 +86,9 @@ domainalias; domalias2.com; {self.domain}; True
f"truc@{self.domain}; 5\r\n",
name="domains.csv",
)
response = self.client.post(reverse("admin:identity_import"), {"sourcefile": f})
url = reverse("v2:identities-import-from-csv")
response = self.client.post(url, {"sourcefile": f})
self.assertEqual(response.status_code, 200)
self.assertTrue(limit.is_exceeded())
f = ContentFile(
@@ -84,12 +97,15 @@ account; user3@{self.domain}; toto; User; One; True; SimpleUsers; user3@{self.do
""",
name="domains.csv",
)
response = self.client.post(reverse("admin:identity_import"), {"sourcefile": f})
self.assertContains(response, "Mailboxes: limit reached")
response = self.client.post(url, {"sourcefile": f})
self.assertEqual(response.status_code, 200)
result = response.json()
self.assertFalse(result["status"])
self.assertEqual(result["message"], "Mailboxes: limit reached")
def _test_mailbox_aliases_import(self, limit):
"""Check mailbox aliases limit."""
self.client.force_login(self.dadmin)
self.authenticate_user(self.dadmin)
self.assertFalse(limit.is_exceeded())
f = ContentFile(
f"""alias; alias1@{self.domain}; True; user@{self.domain}
@@ -97,7 +113,9 @@ alias; alias2@{self.domain}; True; user@{self.domain}
""",
name="aliases.csv",
)
self.client.post(reverse("admin:identity_import"), {"sourcefile": f})
url = reverse("v2:identities-import-from-csv")
response = self.client.post(url, {"sourcefile": f})
self.assertEqual(response.status_code, 200)
self.assertTrue(limit.is_exceeded())
f = ContentFile(
@@ -107,7 +125,7 @@ alias; alias3@{self.domain}; True; user@{self.domain}
name="aliases.csv",
)
with self.assertRaises(exceptions.ValidationError):
self.client.post(reverse("admin:identity_import"), {"sourcefile": f})
self.client.post(url, {"sourcefile": f})
class UserLimitImportTestCase(LimitImportTestCase):
@@ -123,19 +141,20 @@ class UserLimitImportTestCase(LimitImportTestCase):
def test_domains_import(self):
"""Check domains limit."""
self.client.force_login(self.reseller)
self.authenticate_user(self.reseller)
limit = self.reseller.userobjectlimit_set.get(name="domains")
self.assertFalse(limit.is_exceeded())
f = ContentFile("domain; domain1.com; 1; 1; True", name="domains.csv")
self.client.post(reverse("admin:domain_import"), {"sourcefile": f})
url = reverse("v2:domain-import-from-csv")
self.client.post(url, {"sourcefile": f})
self.assertFalse(limit.is_exceeded())
f = ContentFile("domain; domain2.com; 0; 1; True", name="domains.csv")
response = self.client.post(reverse("admin:domain_import"), {"sourcefile": f})
response = self.client.post(url, {"sourcefile": f})
self.assertContains(response, "not allowed to define unlimited values")
f = ContentFile("domain; domain2.com; 2; 1; True", name="domains.csv")
response = self.client.post(reverse("admin:domain_import"), {"sourcefile": f})
response = self.client.post(url, {"sourcefile": f})
self.assertContains(response, "Quota: limit reached")
f = ContentFile(
@@ -144,7 +163,7 @@ domain; domain3.com; 1000; 100; True
""",
name="domains.csv",
)
response = self.client.post(reverse("admin:domain_import"), {"sourcefile": f})
response = self.client.post(url, {"sourcefile": f})
self.assertContains(response, "Domains: limit reached")
def test_domain_alias_import(self):
modoboa/limits/tests/test_user_limits.py
+174 -139
View File
@@ -1,16 +1,18 @@
"""Test cases for the limits extension."""
from typing import Optional
from django.urls import reverse
from modoboa.admin.factories import populate_database
from modoboa.admin.models import Alias, Domain
from modoboa.admin.models import Alias, Domain, DomainAlias
from modoboa.core.factories import UserFactory
from modoboa.core.models import User
from modoboa.lib import tests as lib_tests
from .. import utils
class PermissionsTestCase(lib_tests.ModoTestCase):
class PermissionsTestCase(lib_tests.ModoAPITestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
@@ -27,24 +29,24 @@ class PermissionsTestCase(lib_tests.ModoTestCase):
"username": "reseller@test.com",
"first_name": "Reseller",
"last_name": "",
"password1": "Toto1234",
"password2": "Toto1234",
"password": "Toto1234",
"role": "Resellers",
"is_active": True,
"email": "reseller@test.com",
"stepid": "step2",
}
self.ajax_post(reverse("admin:account_add"), values)
url = reverse("v2:account-list")
response = self.client.post(url, values, format="json")
self.assertEqual(response.status_code, 201)
account = User.objects.get(username="reseller@test.com")
admin = User.objects.get(username="admin@test.com")
self.client.force_login(admin)
resp = self.ajax_post(
reverse("admin:account_delete", args=[account.id]), {}, status=403
)
self.assertEqual(resp, "Permission denied")
self.authenticate_user(admin)
url = reverse("v2:account-delete", args=[account.id])
response = self.client.post(url, {"keep_folder": False}, format="json")
# account is not visible by admin -> 404
self.assertEqual(response.status_code, 404)
class ResourceTestCase(lib_tests.ModoTestCase):
class ResourceTestCase(lib_tests.ModoAPITestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
@@ -58,67 +60,66 @@ class ResourceTestCase(lib_tests.ModoTestCase):
cls.localconfig.save()
populate_database()
def _create_account(self, username, role="SimpleUsers", status=200, **kwargs):
def _create_account(self, username, role="SimpleUsers", status=201, **kwargs):
values = {
"username": username,
"first_name": "Tester",
"last_name": "Toto",
"password1": "Toto1234",
"password2": "Toto1234",
"password": "Toto1234",
"role": role,
"quota_act": True,
"is_active": True,
"email": username,
"stepid": "step2",
"mailbox": {"use_domain_quota": True},
}
values.update(kwargs)
return self.ajax_post(reverse("admin:account_add"), values, status)
response = self.client.post(reverse("v2:account-list"), values, format="json")
self.assertEqual(response.status_code, status)
return response
def _create_alias(self, email, rcpt="user@test.com", status=200):
def _create_alias(self, email, rcpt: Optional[list] = None, status=201):
if rcpt is None:
rcpt = ["user@test.com"]
values = {"address": email, "recipients": rcpt, "enabled": True}
return self.ajax_post(reverse("admin:alias_add"), values, status)
response = self.client.post(reverse("v2:alias-list"), values, format="json")
self.assertEqual(response.status_code, status)
return response
def _create_domain(self, name, status=200, withtpl=False, **kwargs):
def _create_domain(self, name, status=201, withtpl=False, **kwargs):
values = {
"name": name,
"quota": 100,
"default_mailbox_quota": 10,
"create_dom_admin": False,
"with_mailbox": True,
"create_aliases": False,
"stepid": "step3",
"type": "domain",
}
if withtpl:
values.update(
{
"create_dom_admin": True,
"dom_admin_username": "admin",
"create_aliases": True,
"domain_admin": {
"username": "admin",
"with_mailbox": True,
"with_aliases": True,
}
}
)
values.update(kwargs)
response = self.ajax_post(reverse("admin:domain_add"), values, status)
response = self.client.post(reverse("v2:domain-list"), values, format="json")
self.assertEqual(response.status_code, status)
return response
def _domain_alias_operation(self, optype, domain, name, status=200):
dom = Domain.objects.get(name=domain)
values = {
"name": dom.name,
"quota": dom.quota,
"enabled": dom.enabled,
"type": "domain",
"default_mailbox_quota": dom.default_mailbox_quota,
}
aliases = [alias.name for alias in dom.domainalias_set.all()]
def _domain_alias_operation(self, optype, domain, name, status=201):
if optype == "add":
aliases.append(name)
else:
aliases.remove(name)
for cpt, alias in enumerate(aliases):
fname = "aliases" if not cpt else f"aliases_{cpt}"
values[fname] = alias
self.ajax_post(reverse("admin:domain_change", args=[dom.id]), values, status)
dom = Domain.objects.get(name=domain)
values = {"name": name, "target": dom.pk, "enabled": True}
response = self.client.post(
reverse("v2:domain_alias-list"), values, format="json"
)
self.assertEqual(response.status_code, status)
return response
dalias = DomainAlias.objects.get(name=name)
response = self.client.delete(
reverse("v2:domain_alias-detail", args=[dalias.pk])
)
self.assertEqual(response.status_code, status)
def _check_limit(self, name, curvalue, maxvalue):
limit = self.user.userobjectlimit_set.get(name=name)
@@ -140,22 +141,25 @@ class DomainAdminTestCase(ResourceTestCase):
def setUp(self):
"""Test initialization."""
super().setUp()
self.client.force_login(self.user)
self.authenticate_user(self.user)
def test_mailboxes_limit(self):
self._create_account("tester1@test.com")
self._check_limit("mailboxes", 1, 2)
self._create_account("tester2@test.com")
self._check_limit("mailboxes", 2, 2)
self._create_account("tester3@test.com", status=403)
response = self._create_account("tester3@test.com", status=400)
self.assertEqual(response.json()["mailbox"][0], "Mailboxes: limit reached")
self._check_limit("mailboxes", 2, 2)
self.ajax_post(
response = self.client.post(
reverse(
"admin:account_delete",
"v2:account-delete",
args=[User.objects.get(username="tester2@test.com").id],
),
{},
format="json",
)
self.assertEqual(response.status_code, 204)
self._check_limit("mailboxes", 1, 2)
def test_aliases_limit(self):
@@ -163,7 +167,10 @@ class DomainAdminTestCase(ResourceTestCase):
self._check_limit("mailbox_aliases", 1, 2)
self._create_alias("alias2@test.com")
self._check_limit("mailbox_aliases", 2, 2)
self._create_alias("alias3@test.com", status=403)
response = self._create_alias("alias3@test.com", status=400)
self.assertEqual(
response.json()["address"][0], "Mailbox aliases: limit reached"
)
self._check_limit("mailbox_aliases", 2, 2)
# Set unlimited value
self.user.userobjectlimit_set.filter(name="mailbox_aliases").update(
@@ -171,27 +178,33 @@ class DomainAdminTestCase(ResourceTestCase):
)
self._create_alias("alias3@test.com")
self._check_limit("mailbox_aliases", 3, -1)
self.ajax_delete(
reverse("admin:alias_delete")
+ f"?selection={Alias.objects.get(address='alias2@test.com').id}"
response = self.client.delete(
reverse(
"v2:alias-detail",
args=[Alias.objects.get(address="alias2@test.com").id],
)
)
self.assertEqual(response.status_code, 204)
self._check_limit("mailbox_aliases", 2, -1)
def test_aliases_limit_through_account_form(self):
def test_aliases_limit_through_account_viewset(self):
user = User.objects.get(username="user@test.com")
values = {
"username": user.username,
"role": user.role,
"is_active": user.is_active,
"email": user.email,
"quota_act": True,
"aliases": "alias1@test.com",
"aliases_1": "alias2@test.com",
"mailbox": {"quota": 10},
"aliases": [
"alias1@test.com",
"alias2@test.com",
],
"language": "en",
"subject": "subject",
"content": "content",
}
self.ajax_post(reverse("admin:account_change", args=[user.id]), values)
response = self.client.put(
reverse("v2:account-detail", args=[user.id]), values, format="json"
)
self.assertEqual(response.status_code, 200)
Alias.objects.get(address="alias1@test.com")
self._check_limit("mailbox_aliases", 2, 2)
@@ -209,24 +222,24 @@ class ResellerTestCase(ResourceTestCase):
def setUp(self):
"""Test initialization."""
super().setUp()
self.client.force_login(self.user)
self.authenticate_user(self.user)
def test_domains_limit(self):
response = self.client.get(reverse("admin:domain_list"))
self.assertContains(response, "Domains (0%)")
self.assertContains(response, "Domain aliases (0%)")
self._create_domain("domain1.tld")
self._check_limit("domains", 1, 2)
self._create_domain("domain2.tld")
self._check_limit("domains", 2, 2)
self._create_domain("domain3.tld", 403)
response = self._create_domain("domain3.tld", 403)
self._check_limit("domains", 2, 2)
self.ajax_post(
response = self.client.post(
reverse(
"admin:domain_delete", args=[Domain.objects.get(name="domain2.tld").id]
"v2:domain-delete", args=[Domain.objects.get(name="domain2.tld").id]
),
{},
format="json",
)
self.assertEqual(response.status_code, 204)
self._check_limit("domains", 1, 2)
def test_domain_aliases_limit(self):
@@ -235,17 +248,15 @@ class ResellerTestCase(ResourceTestCase):
self._check_limit("domain_aliases", 1, 2)
self._domain_alias_operation("add", "pouet.com", "domain-alias2.tld")
self._check_limit("domain_aliases", 2, 2)
self._domain_alias_operation("add", "pouet.com", "domain-alias3.tld", 403)
response = self._domain_alias_operation(
"add", "pouet.com", "domain-alias3.tld", 400
)
self.assertEqual(response.json()["domain"], "Domain aliases: limit reached")
self._check_limit("domain_aliases", 2, 2)
self._domain_alias_operation("delete", "pouet.com", "domain-alias2.tld")
self._domain_alias_operation("delete", "pouet.com", "domain-alias2.tld", 204)
self._check_limit("domain_aliases", 1, 2)
def test_domain_admins_limit(self):
response = self.client.get(reverse("admin:identity_list"))
self.assertContains(response, "Domain admins (0%)")
self.assertContains(response, "Mailboxes (0%)")
self.assertContains(response, "Mailbox aliases (0%)")
self._create_domain("domain.tld")
self._create_account("admin1@domain.tld", role="DomainAdmins")
self._check_limit("domain_admins", 1, 2)
@@ -255,9 +266,7 @@ class ResellerTestCase(ResourceTestCase):
"admin3@domain.tld", role="DomainAdmins", status=400
)
self.assertEqual(
resp["form_errors"]["role"][0],
"Select a valid choice. DomainAdmins is not one of the available "
"choices.",
resp.json()["role"][0], '"DomainAdmins" is not a valid choice.'
)
self._check_limit("domain_admins", 2, 2)
@@ -267,20 +276,16 @@ class ResellerTestCase(ResourceTestCase):
values = {
"username": user.username,
"role": "DomainAdmins",
"quota_act": True,
"is_active": user.is_active,
"email": user.email,
"language": "en",
"subject": "subject",
"content": "content",
}
resp = self.ajax_post(
reverse("admin:account_change", args=[user.id]), values, status=400
response = self.client.put(
reverse("v2:account-detail", args=[user.id]), values, format="json"
)
self.assertEqual(response.status_code, 400)
self.assertEqual(
resp["form_errors"]["role"][0],
"Select a valid choice. DomainAdmins is not one of the available "
"choices.",
response.json()["role"][0], '"DomainAdmins" is not a valid choice.'
)
self._check_limit("domain_admins", 2, 2)
@@ -297,9 +302,10 @@ class ResellerTestCase(ResourceTestCase):
self._create_domain("domain2.tld", withtpl=True)
self._check_limit("domain_admins", 2, 2)
self._check_limit("domains", 2, 3)
self._create_domain("domain3.tld", status=200, withtpl=True)
response = self._create_domain("domain3.tld", withtpl=True, status=403)
self.assertEqual(response.json(), "Domain admins: limit reached")
self._check_limit("domain_admins", 2, 2)
self._check_limit("domains", 3, 3)
self._check_limit("domains", 2, 3)
def test_quota(self):
"""Check quota resource."""
@@ -307,9 +313,9 @@ class ResellerTestCase(ResourceTestCase):
response = self._create_domain(
"domain2.tld", status=403, withtpl=True, quota=1000
)
self.assertEqual(response, "Quota: limit reached")
self.assertEqual(response.json(), "Quota: limit reached")
dom1 = Domain.objects.get(name="domain1.tld")
url = reverse("admin:domain_change", args=[dom1.pk])
url = reverse("v2:domain-detail", args=[dom1.pk])
values = {
"name": dom1.name,
"type": dom1.type,
@@ -317,17 +323,18 @@ class ResellerTestCase(ResourceTestCase):
"default_mailbox_quota": dom1.default_mailbox_quota,
"quota": 500,
}
self.ajax_post(url, values)
response = self.client.put(url, values, format="json")
self.assertEqual(response.status_code, 200)
def test_quota_constraints(self):
"""Check reseller can't define unlimited quota."""
response = self._create_domain("domain1.tld", 400, quota=0)
self.assertEqual(
response["form_errors"]["quota"][0], "You can't define an unlimited quota."
response.json()["quota"][0], "You can't define an unlimited quota."
)
response = self._create_domain("domain1.tld", 400, default_mailbox_quota=0)
self.assertEqual(
response["form_errors"]["default_mailbox_quota"][0],
response.json()["default_mailbox_quota"][0],
"You can't define an unlimited quota.",
)
@@ -342,8 +349,7 @@ class ResellerTestCase(ResourceTestCase):
self._create_account(
"admin@domain1.tld",
role="DomainAdmins",
quota_act=False,
quota=0,
mailbox={"use_domain_quota": False, "quota": 0},
status=400,
)
@@ -354,21 +360,27 @@ class ResellerTestCase(ResourceTestCase):
"role": admin.role,
"is_active": admin.is_active,
"email": admin.email,
"mailboxes_limit": 1,
"mailbox_aliases_limit": 2,
"resources": [
{"name": "mailboxes", "max_value": 1},
{"name": "mailbox_aliases", "max_value": 2},
],
"language": "en",
"quota_act": False,
"quota": 0,
"subject": "subject",
"content": "content",
"mailbox": {
"quota": 0,
},
}
self.ajax_post(reverse("admin:account_change", args=[admin.pk]), values, 400)
response = self.client.put(
reverse("v2:account-detail", args=[admin.pk]), values, format="json"
)
self.assertEqual(response.status_code, 400)
def test_reseller_deletes_domain(self):
"""Check if all resources are restored after the deletion."""
self._create_domain("domain.tld", withtpl=True)
dom = Domain.objects.get(name="domain.tld")
self.ajax_post(reverse("admin:domain_delete", args=[dom.id]), {})
url = reverse("v2:domain-delete", args=[dom.id])
response = self.client.post(url, {}, format="json")
self.assertEqual(response.status_code, 204)
self._check_limit("domains", 0, 2)
self._check_limit("domain_admins", 1, 2)
self._check_limit("mailboxes", 0, 2)
@@ -377,14 +389,14 @@ class ResellerTestCase(ResourceTestCase):
def test_sadmin_removes_ownership(self):
self._create_domain("domain.tld", withtpl=True)
dom = Domain.objects.get(name="domain.tld")
self.client.logout()
self.client.login(username="admin", password="password")
self.ajax_delete(
"{}?domid={}&daid={}".format(
reverse("admin:permission_remove"), dom.id, self.user.id
),
{},
self.authenticate_user(self.sadmin)
response = self.client.post(
reverse("v2:domain-remove-administrator", args=[dom.id]),
{"account": self.user.id},
format="json",
)
self.assertEqual(response.status_code, 200)
self._check_limit("domains", 0, 2)
self._check_limit("domain_admins", 0, 2)
self._check_limit("mailboxes", 0, 2)
@@ -399,22 +411,28 @@ class ResellerTestCase(ResourceTestCase):
values = {
"username": user.username,
"role": user.role,
"quota_act": True,
"is_active": user.is_active,
"email": user.email,
"mailboxes_limit": 1,
"mailbox_aliases_limit": 2,
"resources": [
{"name": "mailboxes", "max_value": 1},
{"name": "mailbox_aliases", "max_value": 2},
],
"language": "en",
"subject": "subject",
"content": "content",
"mailbox": {"use_domain_quota": True},
}
self.ajax_post(reverse("admin:account_change", args=[user.id]), values)
response = self.client.put(
reverse("v2:account-detail", args=[user.id]), values, format="json"
)
self.assertEqual(response.status_code, 200)
self._check_limit("mailboxes", 1, 1)
self._check_limit("mailbox_aliases", 0, 0)
# Delete the admin -> resources should go back to the
# reseller's pool
self.ajax_post(reverse("admin:account_delete", args=[user.id]), {})
response = self.client.post(
reverse("v2:account-delete", args=[user.id]), {}, format="json"
)
self.assertEqual(response.status_code, 204)
self._check_limit("mailboxes", 0, 2)
self._check_limit("mailbox_aliases", 0, 2)
@@ -427,25 +445,32 @@ class ResellerTestCase(ResourceTestCase):
values = {
"username": user.username,
"role": user.role,
"quota_act": True,
"is_active": user.is_active,
"email": user.email,
"mailboxes_limit": 1,
"mailbox_aliases_limit": 2,
"resources": [
{"name": "mailboxes", "max_value": 1},
{"name": "mailbox_aliases", "max_value": 2},
],
"language": "en",
}
self.ajax_post(reverse("admin:account_change", args=[user.id]), values)
response = self.client.put(
reverse("v2:account-detail", args=[user.id]), values, format="json"
)
self.assertEqual(response.status_code, 200)
dom.add_admin(user)
self.client.logout()
self.client.login(username="admin1@domain.tld", password="Toto1234")
self._create_account("user1@domain.tld")
self._create_alias("alias1@domain.tld", "user1@domain.tld")
self._create_alias("alias2@domain.tld", "user1@domain.tld")
self.client.logout()
self.client.login(username="reseller", password="toto")
self.authenticate_user(user)
self._create_account("user1@domain.tld", mailbox={"quota": 10})
self._create_alias("alias1@domain.tld", ["user1@domain.tld"])
self._create_alias("alias2@domain.tld", ["user1@domain.tld"])
self.authenticate_user(self.user)
# Delete the admin -> resources should go back to the
# reseller's pool
self.ajax_post(reverse("admin:account_delete", args=[user.id]), {})
response = self.client.post(
reverse("v2:account-delete", args=[user.id]), {}, format="json"
)
self.assertEqual(response.status_code, 204)
self._check_limit("mailboxes", 1, 2)
self._check_limit("mailbox_aliases", 2, 2)
@@ -458,14 +483,18 @@ class ResellerTestCase(ResourceTestCase):
values = {
"username": user.username,
"role": user.role,
"quota_act": True,
"is_active": user.is_active,
"email": user.email,
"mailboxes_limit": 1,
"mailbox_aliases_limit": 2,
"resources": [
{"name": "mailboxes", "max_value": 1},
{"name": "mailbox_aliases", "max_value": 2},
],
"language": "en",
}
self.ajax_post(reverse("admin:account_change", args=[user.id]), values)
response = self.client.put(
reverse("v2:account-detail", args=[user.id]), values, format="json"
)
self.assertEqual(response.status_code, 200)
self._check_limit("mailboxes", 1, 1)
self._check_limit("mailbox_aliases", 0, 0)
@@ -479,7 +508,10 @@ class ResellerTestCase(ResourceTestCase):
"email": user.email,
"language": "en",
}
self.ajax_post(reverse("admin:account_change", args=[user.id]), values)
response = self.client.put(
reverse("v2:account-detail", args=[user.id]), values, format="json"
)
self.assertEqual(response.status_code, 200)
self._check_limit("mailboxes", 1, 2)
self._check_limit("mailbox_aliases", 0, 2)
@@ -495,13 +527,16 @@ class ResellerTestCase(ResourceTestCase):
"quota_act": True,
"is_active": user.is_active,
"email": user.email,
"mailboxes_limit": 2,
"mailbox_aliases_limit": 3,
"resources": [
{"name": "mailboxes", "max_value": 2},
{"name": "mailbox_aliases", "max_value": 3},
],
"language": "en",
}
resp = self.ajax_post(
reverse("admin:account_change", args=[user.id]), values, 424
response = self.client.put(
reverse("v2:account-detail", args=[user.id]), values, format="json"
)
self.assertEqual(resp, "Not enough resources")
self.assertEqual(response.status_code, 424)
self.assertEqual(response.json(), "Not enough resources")
self._check_limit("mailboxes", 1, 2)
self._check_limit("mailbox_aliases", 0, 2)
modoboa/maillog/tests/test_views.py
-87
View File
@@ -7,11 +7,9 @@ import tempfile
from django.conf import settings
from django.core.management import call_command
from django.urls import reverse
from django.test import override_settings
from modoboa.admin import factories as admin_factories
from modoboa.core import models as core_models
from modoboa.lib.tests import ModoTestCase
@@ -48,91 +46,6 @@ class RunCommandsMixin:
call_command("update_statistics", *args)
@override_settings(RRDTOOL_TEST_MODE=True)
class ViewsTestCase(RunCommandsMixin, ModoTestCase):
"""Views test cases."""
@classmethod
def setUpTestData(cls): # noqa
super().setUpTestData()
admin_factories.populate_database()
cls.da = core_models.User.objects.get(username="admin@test.com")
def tearDown(self):
super().tearDown()
self.set_global_parameter("greylist", False)
def test_index(self):
"""Test index view."""
url = reverse("maillog:fullindex")
response = self.client.get(url)
self.assertContains(response, 'id="graphs_accountgraphicset"')
self.assertContains(response, 'id="graphs_mailtraffic"')
self.client.force_login(self.da)
response = self.client.get(url)
self.assertContains(response, 'id="graphs_mailtraffic"')
def test_graphs(self):
"""Test graphs views."""
self.run_logparser()
url = reverse("maillog:graph_list")
self.ajax_get(url, status=404)
response = self.ajax_get(f"{url}?gset=mailtraffic")
self.assertIn("averagetraffic", response["graphs"])
for period in ["week", "month", "year"]:
response = self.ajax_get(f"{url}?gset=mailtraffic&period={period}")
self.assertIn("averagetraffic", response["graphs"])
self.assertEqual(response["period_name"], period)
# custom period
today = datetime.date.today()
start = f"{today} 11:00:00"
end = f"{today} 11:40:00"
response = self.ajax_get(
f"{url}?gset=mailtraffic&period=custom&start={start}&end={end}"
)
self.assertIn("averagetraffic", response["graphs"])
# unknown domain
response = self.ajax_get(
f"{url}?gset=mailtraffic&searchquery=unknown.com", status=400
)
# check with greylist enabled
self.set_global_parameter("greylist", True)
response = self.ajax_get(f"{url}?gset=mailtraffic")
self.assertIn("averagetraffic", response["graphs"])
def test_account_created_graph(self):
"""Check data."""
self.run_update_statistics(rebuild=True)
url = reverse("maillog:graph_list")
response = self.ajax_get(f"{url}?gset=accountgraphicset")
data = response["graphs"]["accountcreationgraphic"]["series"][0]["data"]
self.assertEqual(data[-1]["y"], 5.0)
def test_graphs_as_domainadmin(self):
"""Test graph views as domain admin."""
self.run_logparser()
self.client.force_login(self.da)
url = "{}?gset=mailtraffic".format(reverse("maillog:graph_list"))
response = self.ajax_get(url)
self.assertIn("averagetraffic", response["graphs"])
response = self.ajax_get(f"{url}&searchquery=test.com")
self.assertIn("averagetraffic", response["graphs"])
response = self.ajax_get(f"{url}&searchquery=test2.com", status=403)
def test_get_domain_list(self):
"""Test get_domain_list view."""
url = reverse("maillog:domain_list")
response = self.ajax_get(url)
self.assertIn("test.com", response)
self.assertIn("test2.com", response)
@override_settings(RRDTOOL_TEST_MODE=True)
class ManagementCommandsTestCase(RunCommandsMixin, ModoTestCase):
"""Management command test cases."""
modoboa/maillog/urls.py
-13
View File
@@ -1,13 +0,0 @@
"""Stats urls."""
from django.urls import path
from . import views
app_name = "maillog"
urlpatterns = [
path("", views.index, name="fullindex"),
path("graphs/", views.graphs, name="graph_list"),
path("domains/", views.get_domain_list, name="domain_list"),
]
modoboa/maillog/views.py
-104
View File
@@ -1,104 +0,0 @@
"""Custom views."""
import re
import time
from django.shortcuts import render
from django.utils.translation import gettext as _
from django.contrib.auth.decorators import (
login_required,
user_passes_test,
permission_required,
)
from modoboa.admin.models import Domain
from modoboa.lib.exceptions import BadRequest, NotFound
from modoboa.lib.web_utils import render_to_json_response
from . import signals
from .lib import date_to_timestamp
@login_required
@permission_required("admin.view_mailbox")
def index(request):
"""
FIXME: how to select a default graph set ?
"""
deflocation = "graphs/?gset=mailtraffic"
if not request.user.is_superuser:
if not Domain.objects.get_for_admin(request.user).count():
raise NotFound(_("No statistics available"))
graph_sets = {}
for result in signals.get_graph_sets.send(sender="index", user=request.user):
graph_sets.update(result[1])
periods = [
{"name": "day", "label": _("Day")},
{"name": "week", "label": _("Week")},
{"name": "month", "label": _("Month")},
{"name": "year", "label": _("Year")},
{"name": "custom", "label": _("Custom")},
]
return render(
request,
"maillog/index.html",
{
"periods": periods,
"selection": "stats",
"deflocation": deflocation,
"graph_sets": graph_sets,
},
)
@login_required
@user_passes_test(lambda u: u.role != "SimpleUsers")
def graphs(request):
gset = request.GET.get("gset", None)
searchq = request.GET.get("searchquery", None)
graph_sets = {}
for result in signals.get_graph_sets.send(sender="index", user=request.user):
graph_sets.update(result[1])
if gset not in graph_sets:
raise NotFound(_("Unknown graphic set"))
period = request.GET.get("period", "day")
tplvars = {"graphs": {}, "period": period}
fname = graph_sets[gset].get_file_name(request.user, searchq)
if fname is None:
raise BadRequest(_("Unknown domain"))
tplvars["fname"] = fname
if period == "custom":
if "start" not in request.GET or "end" not in request.GET:
raise BadRequest(_("Bad custom period"))
start = request.GET["start"]
end = request.GET["end"]
expr = re.compile(r"[:\- ]")
period_name = "{}_{}".format(expr.sub("", start), expr.sub("", end))
start = date_to_timestamp(expr.split(start))
end = date_to_timestamp(expr.split(end))
else:
end = int(time.mktime(time.localtime()))
start = f"-1{period}"
period_name = period
tplvars["domain_selector"] = graph_sets[gset].domain_selector
tplvars["graphs"] = graph_sets[gset].export(tplvars["fname"], start, end)
tplvars["period_name"] = period_name
tplvars["start"] = start
tplvars["end"] = end
return render_to_json_response(tplvars)
@login_required
@user_passes_test(lambda u: u.role != "SimpleUsers")
def get_domain_list(request):
"""Get the list of domains the user can see."""
doms = []
for dom in Domain.objects.get_for_admin(request.user):
doms += [dom.name]
doms += [alias.name for alias in dom.domainalias_set.all()]
return render_to_json_response(doms)
modoboa/pdfcredentials/api/v2/tests.py
+1 -1
View File
@@ -162,7 +162,7 @@ class PDFCredentialViewTestCase(ModoAPITestCase):
"""Test validation for unwritable directory."""
self.set_global_parameter("storage_dir", "/nonexistentdir")
self.set_global_parameter("enabled_pdfcredentials", False)
url = reverse("v2:parameter-detail", args=["pdfcredentials"])
url = reverse("v2:parameter-global-detail", args=["pdfcredentials"])
data = {
"webpanel_url": "http://localhost",
"smtp_server_address": "mail.localhost",
modoboa/pdfcredentials/handlers.py
+28 -1
View File
@@ -1,18 +1,21 @@
"""PDF credentials handlers."""
import logging
import os
from django.db.models import signals
from django.dispatch import receiver
from django.urls import reverse
from django.utils.translation import gettext as _
from modoboa.admin import signals as admin_signals
from modoboa.core import models as core_models
from modoboa.core import signals as core_signals
from modoboa.lib.exceptions import InternalError
from modoboa.parameters import tools as param_tools
from .documents import credentials
from .lib import init_storage_dir, delete_credentials
from .lib import init_storage_dir, delete_credentials, get_creds_filename
@receiver(core_signals.account_password_updated)
@@ -40,3 +43,27 @@ def password_updated(sender, account, password, created, **kwargs):
def account_deleted(sender, instance, **kwargs):
"""Remove document."""
delete_credentials(instance)
@receiver(admin_signals.extra_account_identities_actions)
def extra_identities_actions(sender, account, **kwargs):
"""
Add download credential action for identity.
Used for api v2.
"""
if not param_tools.get_global_parameter("enabled_pdfcredentials"):
return []
fname = get_creds_filename(account)
if not os.path.exists(fname):
return []
url = reverse("v2:get-credentials", args=[account.pk])
url = url.replace("/api/v2", "")
return {
"name": "get_credentials",
"icon": "mdi-file-download-outline",
"label": _("Download PDF credentials"),
"type": "download",
"content_type": "application/pdf",
"url": url,
"filename": os.path.basename(fname),
}
modoboa/pdfcredentials/tests/test_views.py
-149
View File
@@ -1,149 +0,0 @@
"""PDF credentials tests."""
import os
import shutil
import tempfile
from django.test import override_settings
from django.urls import reverse
from modoboa.admin import factories as admin_factories
from modoboa.core import models as core_models
from modoboa.lib.tests import ModoTestCase
class EventsTestCase(ModoTestCase):
"""Test event handlers."""
@classmethod
def setUpTestData(cls):
"""Create some data."""
super().setUpTestData()
admin_factories.DomainFactory(name="test.com")
def setUp(self):
"""Create temp. directory to store files."""
super().setUp()
self.workdir = tempfile.mkdtemp()
self.set_global_parameter("storage_dir", self.workdir)
def tearDown(self):
"""Reset test env."""
shutil.rmtree(self.workdir)
def _create_account(self, username, expected_status=200):
"""Create a test account."""
values = {
"username": username,
"first_name": "Tester",
"last_name": "Toto",
"role": "SimpleUsers",
"quota_act": True,
"is_active": True,
"email": username,
"random_password": True,
"stepid": 2,
}
response = self.client.post(
reverse("admin:account_add"), values, HTTP_X_REQUESTED_WITH="XMLHttpRequest"
)
self.assertEqual(response.status_code, expected_status)
return values
def test_password_updated(self):
"""Check that document is generated at account creation/update."""
values = self._create_account("leon@test.com")
fname = os.path.join(self.workdir, "{}.pdf".format(values["username"]))
self.assertTrue(os.path.exists(fname))
account = core_models.User.objects.get(username=values["username"])
# Check if link is present in listing page
response = self.ajax_get(reverse("admin:_identity_list"))
self.assertIn('name="get_credentials"', response["rows"])
# Try to download the file
response = self.client.get(
reverse("pdfcredentials:account_credentials", args=[account.pk])
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response["Content-Type"], "application/pdf")
# File have been deleted?
self.assertFalse(os.path.exists(fname))
# Try to download a second time
response = self.client.get(
reverse("pdfcredentials:account_credentials", args=[account.pk])
)
self.assertContains(response, "No document available for this user")
# Update account
values.update({"language": "en", "subject": "subject", "content": "content"})
self.ajax_post(reverse("admin:account_change", args=[account.pk]), values)
self.assertFalse(os.path.exists(fname))
self.set_global_parameter("generate_at_creation", False)
self.ajax_post(reverse("admin:account_change", args=[account.pk]), values)
self.assertTrue(os.path.exists(fname))
def test_with_connection_settings(self):
"""Add connection settings to documents."""
self.set_global_parameter("include_connection_settings", True)
values = self._create_account("leon@test.com")
fname = os.path.join(self.workdir, "{}.pdf".format(values["username"]))
self.assertTrue(os.path.exists(fname))
def test_with_custom_message(self):
"""Add custom message to documents."""
self.set_global_parameter("custom_message", "This is a test message.")
values = self._create_account("leon@test.com")
fname = os.path.join(self.workdir, "{}.pdf".format(values["username"]))
self.assertTrue(os.path.exists(fname))
def test_account_delete(self):
"""Check that document is deleted with account."""
values = self._create_account("leon@test.com")
fname = os.path.join(self.workdir, "{}.pdf".format(values["username"]))
self.assertTrue(os.path.exists(fname))
account = core_models.User.objects.get(username=values["username"])
self.ajax_post(reverse("admin:account_delete", args=[account.pk]), {})
self.assertFalse(os.path.exists(fname))
@override_settings(MODOBOA_LOGO="modoboa.png")
def test_with_custom_logo(self):
"""Check that document is deleted with account."""
values = self._create_account("leon@test.com")
fname = os.path.join(self.workdir, "{}.pdf".format(values["username"]))
self.assertTrue(os.path.exists(fname))
def test_download_and_delete_account(self):
"""Check that document is deleted with account."""
values = self._create_account("leon@test.com")
fname = os.path.join(self.workdir, "{}.pdf".format(values["username"]))
self.assertTrue(os.path.exists(fname))
account = core_models.User.objects.get(username=values["username"])
# Try to download the file
response = self.client.get(
reverse("pdfcredentials:account_credentials", args=[account.pk])
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response["Content-Type"], "application/pdf")
# File have been deleted?
self.assertFalse(os.path.exists(fname))
# Delete account
self.ajax_post(reverse("admin:account_delete", args=[account.pk]), {})
def test_storage_dir_creation(self):
"""Test storage directory creation."""
self.set_global_parameter("storage_dir", "/nonexistentdir")
with self.assertLogs(logger="modoboa.admin", level="ERROR") as log:
self._create_account("leon@test.com")
self.assertIn(
"ERROR:modoboa.admin:Failed to create PDF_credentials directory. "
"Please check the permissions or the path.",
log.output,
)
modoboa/postfix_autoreply/tests.py
+44 -17
View File
@@ -13,7 +13,7 @@ from django.utils.formats import date_format
from modoboa.admin import factories as admin_factories, models as admin_models
from modoboa.core.models import User
from modoboa.lib.tests import ModoTestCase
from modoboa.lib.tests import ModoAPITestCase, ModoTestCase
from modoboa.transport import models as tr_models
from . import factories, models
@@ -106,7 +106,7 @@ Content-Type: text/html; charset=UTF-8
"""
class EventsTestCase(ModoTestCase):
class EventsTestCase(ModoAPITestCase):
@classmethod
def setUpTestData(cls): # NOQA:N802
@@ -114,23 +114,36 @@ class EventsTestCase(ModoTestCase):
super().setUpTestData()
admin_factories.populate_database()
def _create_armessage(self, account):
values = {
"mbox": account.mailbox.id,
"subject": "I'm off",
"content": "I'll back soon",
"enabled": True,
}
response = self.client.post(reverse("v2:armessage-list"), values, format="json")
self.assertEqual(response.status_code, 201)
return response
def test_domain_created_event(self):
values = {
"name": "domain.tld",
"quota": 100,
"default_mailbox_quota": 1,
"create_dom_admin": False,
"stepid": "step3",
"type": "domain",
}
self.ajax_post(reverse("admin:domain_add"), values)
response = self.client.post(reverse("v2:domain-list"), values, format="json")
self.assertEqual(response.status_code, 201)
self.assertTrue(
tr_models.Transport.objects.filter(pattern="autoreply.domain.tld").exists()
)
def test_domain_deleted_event(self):
dom = admin_models.Domain.objects.get(name="test.com")
self.ajax_post(reverse("admin:domain_delete", args=[dom.id]), {})
response = self.client.post(
reverse("v2:domain-delete", args=[dom.id]), {}, format="json"
)
self.assertEqual(response.status_code, 204)
with self.assertRaises(tr_models.Transport.DoesNotExist):
tr_models.Transport.objects.get(pattern="autoreply.test.com")
@@ -143,7 +156,10 @@ class EventsTestCase(ModoTestCase):
"type": "domain",
}
dom = admin_models.Domain.objects.get(name="test.com")
self.ajax_post(reverse("admin:domain_change", args=[dom.id]), values)
response = self.client.put(
reverse("v2:domain-detail", args=[dom.id]), values, format="json"
)
self.assertEqual(response.status_code, 200)
self.assertTrue(
tr_models.Transport.objects.filter(pattern="autoreply.test.fr").exists()
)
@@ -161,16 +177,20 @@ class EventsTestCase(ModoTestCase):
"first_name": "Tester",
"last_name": "Toto",
"role": "SimpleUsers",
"quota_act": True,
"mailbox": {
"use_domain_quota": True,
},
"is_active": True,
"email": "leon@test.com",
"subject": "test",
"content": "test",
"enabled": True,
"language": "en",
}
account = User.objects.get(username="user@test.com")
self.ajax_post(reverse("admin:account_change", args=[account.id]), values)
self._create_armessage(account)
response = self.client.put(
reverse("v2:account-detail", args=[account.id]), values, format="json"
)
self.assertEqual(response.status_code, 200)
self.assertTrue(
admin_models.AliasRecipient.objects.filter(
alias__address="leon@test.com",
@@ -179,7 +199,9 @@ class EventsTestCase(ModoTestCase):
).exists()
)
values["enabled"] = False
self.ajax_post(reverse("admin:account_change", args=[account.id]), values)
self.client.put(
reverse("v2:account-detail", args=[account.id]), values, format="json"
)
self.assertFalse(
admin_models.AliasRecipient.objects.filter(
alias__address="leon@test.com",
@@ -190,7 +212,10 @@ class EventsTestCase(ModoTestCase):
def test_mailbox_deleted_event(self):
account = User.objects.get(username="user@test.com")
self.ajax_post(reverse("admin:account_delete", args=[account.id]), {})
self._create_armessage(account)
self.client.delete(
reverse("v2:account-delete", args=[account.id]), {}, format="json"
)
self.assertFalse(
admin_models.Alias.objects.filter(
address="user@test.com", internal=True
@@ -211,15 +236,17 @@ class EventsTestCase(ModoTestCase):
"first_name": "Tester",
"last_name": "Toto",
"role": "SimpleUsers",
"quota_act": True,
"mailbox": {
"use_domain_quota": True,
},
"is_active": True,
"email": "leon@test.com",
"subject": "test",
"content": "test",
"enabled": True,
"language": "en",
}
self.ajax_post(reverse("admin:account_change", args=[account.id]), values)
self.client.put(
reverse("v2:account-detail", args=[account.id]), values, format="json"
)
self.assertFalse(
admin_models.AliasRecipient.objects.filter(
alias__address="user@test.com",
modoboa/relaydomains/tests.py
+86 -119
View File
@@ -7,7 +7,7 @@ from django.urls import reverse
from modoboa.admin import factories as admin_factories, models as admin_models
from modoboa.core.factories import UserFactory
from modoboa.lib.test_utils import MapFilesTestCaseMixin
from modoboa.lib.tests import ModoTestCase
from modoboa.lib.tests import ModoAPITestCase
from modoboa.limits import utils as limits_utils
from modoboa.transport import factories as tr_factories, models as tr_models
from . import models
@@ -15,44 +15,27 @@ from . import models
class Operations:
def _create_relay_domain(self, name, status=200, **kwargs):
def _create_relay_domain(self, name, status=201, **kwargs):
url = reverse("v2:domain-list")
values = {
"name": name,
"create_dom_admin": False,
"type": "relaydomain",
"service": "relay",
"relay_target_host": "external.host.tld",
"relay_target_port": 25,
"transport": {
"service": "relay",
"settings": {
"relay_target_host": "external.host.tld",
"relay_target_port": 25,
"relay_verify_recipients": False,
},
},
"enabled": True,
"stepid": "step3",
"quota": 0,
"default_mailbox_quota": 0,
}
values.update(kwargs)
return self.ajax_post(reverse("admin:domain_add"), values, status)
def _relay_domain_alias_operation(self, optype, domain, name, status=200):
transport = tr_models.Transport.objects.get(pattern=domain.name)
values = {
"name": domain.name,
"service": "relay",
"relay_target_host": transport._settings["relay_target_host"],
"relay_target_port": transport._settings["relay_target_port"],
"type": "relaydomain",
"quota": domain.quota,
"default_mailbox_quota": domain.default_mailbox_quota,
}
aliases = [alias.name for alias in domain.domainalias_set.all()]
if optype == "add":
aliases.append(name)
else:
aliases.remove(name)
for cpt, alias in enumerate(aliases):
fname = "aliases" if not cpt else f"aliases_{cpt}"
values[fname] = alias
return self.ajax_post(
reverse("admin:domain_change", args=[domain.id]), values, status
)
response = self.client.post(url, values, format="json")
self.assertEqual(response.status_code, status)
return response
def _check_limit(self, name, curvalue, maxvalue):
limit = self.user.userobjectlimit_set.get(name=name)
@@ -60,7 +43,7 @@ class Operations:
self.assertEqual(limit.max_value, maxvalue)
class RelayDomainsTestCase(ModoTestCase, Operations):
class RelayDomainsTestCase(ModoAPITestCase, Operations):
@classmethod
def setUpTestData(cls): # NOQA:N802
@@ -72,7 +55,7 @@ class RelayDomainsTestCase(ModoTestCase, Operations):
service="relay",
_settings={
"relay_target_host": "external.host.tld",
"relay_target_port": "25",
"relay_target_port": 25,
"relay_verify_recipients": False,
},
)
@@ -87,14 +70,13 @@ class RelayDomainsTestCase(ModoTestCase, Operations):
user__groups=("SimpleUsers",),
)
def test_domain_list_view(self):
def test_domain_list(self):
"""Make sure relaydomain is listed."""
url = reverse("admin:_domain_list")
url = reverse("v2:domain-list")
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
content = response.json()
self.assertIn("relaydomain.tld", content["rows"])
self.assertIn("Relay domain", content["rows"])
self.assertEqual(len(content), 3)
def test_create_relaydomain(self):
"""Test the creation of a relay domain.
@@ -109,34 +91,14 @@ class RelayDomainsTestCase(ModoTestCase, Operations):
self.assertEqual(transport._settings["relay_target_host"], "external.host.tld")
self.assertEqual(transport._settings["relay_verify_recipients"], False)
resp = self._create_relay_domain("test.com", 400)
response = self._create_relay_domain("test.com", 400)
self.assertEqual(
resp["form_errors"]["name"][0], "Domain with this Name already exists."
response.json()["name"][0], "domain with this name already exists."
)
resp = self._create_relay_domain("relaydomainalias.tld", 400)
response = self._create_relay_domain("relaydomainalias.tld", 400)
self.assertEqual(
resp["form_errors"]["name"][0],
"A domain alias with this name already exists",
)
def test_create_relaydomainalias(self):
"""Test the creation of a relay domain alias.
We also check that unique constraints are respected: domain,
relay domain.
FIXME: add a check for domain alias.
"""
self._relay_domain_alias_operation("add", self.dom, "relaydomainalias1.tld")
resp = self._relay_domain_alias_operation("add", self.dom, "test.com", 400)
self.assertEqual(
resp["form_errors"]["aliases_2"][0],
"A domain with this name already exists",
)
resp = self._relay_domain_alias_operation("add", self.dom, self.dom.name, 400)
self.assertEqual(
resp["form_errors"]["aliases_2"][0],
"A domain with this name already exists",
response.json()["name"][0],
"domain alias with this name already exists",
)
def test_edit_relaydomain(self):
@@ -146,23 +108,30 @@ class RelayDomainsTestCase(ModoTestCase, Operations):
"""
values = {
"name": "relaydomain.org",
"service": "relay",
"relay_target_host": self.transport._settings["relay_target_host"],
"relay_target_port": 4040,
"relay_verify_recipients": True,
"transport": {
"service": "relay",
"settings": {
"relay_target_host": self.transport._settings["relay_target_host"],
"relay_target_port": 4040,
"relay_verify_recipients": True,
},
},
"type": "relaydomain",
"enabled": True,
"quota": 0,
"default_mailbox_quota": 0,
}
self.ajax_post(reverse("admin:domain_change", args=[self.dom.id]), values)
url = reverse("v2:domain-detail", args=[self.dom.id])
response = self.client.put(url, values, format="json")
self.assertEqual(response.status_code, 200)
self.transport.refresh_from_db()
self.assertEqual(self.transport._settings["relay_target_port"], 4040)
self.assertTrue(
models.RecipientAccess.objects.filter(pattern=values["name"]).exists()
)
values["relay_verify_recipients"] = False
self.ajax_post(reverse("admin:domain_change", args=[self.dom.id]), values)
values["transport"]["settings"]["relay_verify_recipients"] = False
response = self.client.put(url, values, format="json")
self.assertEqual(response.status_code, 200)
self.assertFalse(
models.RecipientAccess.objects.filter(pattern=values["name"]).exists()
)
@@ -170,17 +139,33 @@ class RelayDomainsTestCase(ModoTestCase, Operations):
def test_relaydomain_domain_switch(self):
"""Check domain <-> relaydomain transitions."""
domain_pk = self.dom.pk
transport_def = {
"service": "relay",
"settings": {
"relay_target_host": self.transport._settings["relay_target_host"],
"relay_target_port": self.transport._settings["relay_target_port"],
},
}
values = {
"name": "relaydomain.tld",
"type": "domain",
"quota": 0,
"default_mailbox_quota": 0,
"enabled": True,
"service": "relay",
"relay_target_host": self.transport._settings["relay_target_host"],
"relay_target_port": self.transport._settings["relay_target_port"],
"transport": transport_def,
}
self.ajax_post(reverse("admin:domain_change", args=[domain_pk]), values)
url = reverse("v2:domain-detail", args=[domain_pk])
response = self.client.put(url, values, format="json")
self.assertEqual(response.status_code, 400)
self.assertEqual(
response.json()["transport"][0],
"This field is valid when type is relaydomain",
)
del values["transport"]
response = self.client.put(url, values, format="json")
self.assertEqual(response.status_code, 200)
with self.assertRaises(tr_models.Transport.DoesNotExist):
self.transport.refresh_from_db()
self.dom.refresh_from_db()
@@ -191,35 +176,20 @@ class RelayDomainsTestCase(ModoTestCase, Operations):
"enabled": True,
"quota": 0,
"default_mailbox_quota": 0,
"transport": transport_def,
}
self.ajax_post(reverse("admin:domain_change", args=[domain_pk]), values)
response = self.client.put(url, values, format="json")
self.assertEqual(response.status_code, 200)
self.assertEqual(
admin_models.Domain.objects.get(name="relaydomain.tld").type, "relaydomain"
)
def test_edit_relaydomainalias(self):
"""Test the modification of a relay domain alias.
Rename 'relaydomainalias.tld' domain to 'relaydomainalias.net'
"""
values = {
"name": "relaydomain.org",
"service": "relay",
"relay_target_host": self.transport._settings["relay_target_host"],
"relay_target_port": self.transport._settings["relay_target_port"],
"type": "relaydomain",
"aliases": "relaydomainalias.net",
"quota": 0,
"default_mailbox_quota": 0,
}
self.ajax_post(reverse("admin:domain_change", args=[self.dom.id]), values)
admin_models.DomainAlias.objects.get(name="relaydomainalias.net")
with self.assertRaises(admin_models.DomainAlias.DoesNotExist):
admin_models.DomainAlias.objects.get(name="relaydomainalias.tld")
def test_delete_relaydomain(self):
"""Test the removal of a relay domain."""
self.ajax_post(reverse("admin:domain_delete", args=[self.dom.id]), {})
url = reverse("v2:domain-delete", args=[self.dom.id])
body = {"keep_folder": False}
response = self.client.post(url, body, format="json")
self.assertEqual(response.status_code, 204)
with self.assertRaises(tr_models.Transport.DoesNotExist):
self.transport.refresh_from_db()
@@ -227,7 +197,10 @@ class RelayDomainsTestCase(ModoTestCase, Operations):
"""Test the removal of a recipient access."""
self.transport._settings["relay_verify_recipients"] = True
self.transport.save()
self.ajax_post(reverse("admin:domain_delete", args=[self.dom.id]), {})
url = reverse("v2:domain-delete", args=[self.dom.id])
body = {"keep_folder": False}
response = self.client.post(url, body, format="json")
self.assertEqual(response.status_code, 204)
self.assertFalse(
models.RecipientAccess.objects.filter(
pattern=self.transport.pattern
@@ -238,25 +211,28 @@ class RelayDomainsTestCase(ModoTestCase, Operations):
"""Create an alias on a relay domain."""
values = {
"address": "alias@relaydomain.tld",
"recipients": "recipient@relaydomain.tld",
"recipients": ["recipient@relaydomain.tld"],
"enabled": True,
}
self.ajax_post(reverse("admin:alias_add"), values)
url = reverse("v2:alias-list")
response = self.client.post(url, values, format="json")
self.assertEqual(response.status_code, 201)
self.assertTrue(
admin_models.Alias.objects.filter(address="alias@relaydomain.tld").exists()
)
values = {
"address": "alias2@relaydomain.tld",
"recipients": "local@relaydomain.tld",
"recipients": ["local@relaydomain.tld"],
"enabled": True,
}
self.ajax_post(reverse("admin:alias_add"), values)
response = self.client.post(url, values, format="json")
self.assertEqual(response.status_code, 201)
self.assertTrue(
admin_models.Alias.objects.filter(address="alias2@relaydomain.tld").exists()
)
class ImportTestCase(ModoTestCase):
class ImportTestCase(ModoAPITestCase):
"""Test import."""
def test_webui_import(self):
@@ -264,7 +240,10 @@ class ImportTestCase(ModoTestCase):
f = ContentFile(
"relaydomain;relay.com;127.0.0.1;25;relay;True;True", name="domains.csv"
)
self.client.post(reverse("admin:domain_import"), {"sourcefile": f})
url = reverse("v2:domain-import-from-csv")
body = {"sourcefile": f}
response = self.client.post(url, body)
self.assertEqual(response.status_code, 200)
self.assertTrue(
admin_models.Domain.objects.filter(
name="relay.com", type="relaydomain"
@@ -272,7 +251,7 @@ class ImportTestCase(ModoTestCase):
)
class LimitsTestCase(ModoTestCase, Operations):
class LimitsTestCase(ModoAPITestCase, Operations):
@classmethod
def setUpTestData(cls): # NOQA:N802
@@ -288,7 +267,7 @@ class LimitsTestCase(ModoTestCase, Operations):
def setUp(self):
"""Initialize test."""
super().setUp()
self.client.force_login(self.user)
self.authenticate_user(self.user)
def test_relay_domains_limit(self):
self._create_relay_domain("relaydomain1.tld", quota=1, default_mailbox_quota=1)
@@ -298,23 +277,11 @@ class LimitsTestCase(ModoTestCase, Operations):
self._create_relay_domain("relaydomain3.tld", 403)
self._check_limit("domains", 2, 2)
domid = admin_models.Domain.objects.get(name="relaydomain2.tld").id
self.ajax_post(reverse("admin:domain_delete", args=[domid]), {})
url = reverse("v2:domain-delete", args=[domid])
response = self.client.post(url, {"keep_folder": False}, format="json")
self.assertEqual(response.status_code, 204)
self._check_limit("domains", 1, 2)
def test_relay_domain_aliases_limit(self):
self._create_relay_domain("relaydomain1.tld", quota=1, default_mailbox_quota=1)
domain = admin_models.Domain.objects.get(name="relaydomain1.tld")
self._relay_domain_alias_operation("add", domain, "relay-domain-alias1.tld")
self._check_limit("domain_aliases", 1, 2)
self._relay_domain_alias_operation("add", domain, "relay-domain-alias2.tld")
self._check_limit("domain_aliases", 2, 2)
self._relay_domain_alias_operation(
"add", domain, "relay-domain-alias3.tld", 403
)
self._check_limit("domain_aliases", 2, 2)
self._relay_domain_alias_operation("delete", domain, "relay-domain-alias2.tld")
self._check_limit("domain_aliases", 1, 2)
class MapFilesTestCase(MapFilesTestCaseMixin, TestCase):
"""Test case for relaydomains."""