Merge pull request #845 from tonioo/feature/alias_api_844

Feature: Alias api

modoboa/admin/api.py

@@ -64,3 +64,20 @@ class AccountViewSet(viewsets.ModelViewSet):
             data = {"exists": True}
         serializer = serializers.AccountExistsSerializer(data)
         return Response(serializer.data)
+
+
+class AliasViewSet(viewsets.ModelViewSet):
+    """ViewSet for Alias."""
+
+    permission_classes = [DjangoModelPermissions, ]
+    serializer_class = serializers.AliasSerializer
+
+    def get_queryset(self):
+        """Filter queryset based on current user."""
+        user = self.request.user
+        ids = (
+            user.objectaccess_set.filter(
+                content_type=ContentType.objects.get_for_model(models.Alias))
+            .values_list('object_id', flat=True)
+        )
+        return models.Alias.objects.filter(pk__in=ids)

modoboa/admin/models/alias.py

@@ -213,5 +213,5 @@ class AliasRecipient(models.Model):
     def __str__(self):
         """Return alias and recipient."""
         return smart_text(
-            "{} -> {}".format(self.alias.address, self.alias)
+            "{} -> {}".format(self.alias.address, self.address)
         )

modoboa/admin/serializers.py

@@ -161,3 +161,44 @@ class WritableAccountSerializer(AccountSerializer):
             else:
                 self._create_mailbox(creator, instance, mailbox_data)
         return instance
+
+
+class AliasSerializer(serializers.ModelSerializer):
+    """Base Alias serializer."""
+
+    address = lib_fields.DRFEmailFieldUTF8AndEmptyUser()
+    recipients = serializers.ListField(
+        child=lib_fields.DRFEmailFieldUTF8AndEmptyUser(),
+        allow_empty=False)
+
+    class Meta:
+        model = admin_models.Alias
+        fields = ("pk", "address", "enabled", "internal", "recipients")
+
+    def validate_address(self, value):
+        """Check domain."""
+        local_part, domain = email_utils.split_mailbox(value)
+        self.domain = admin_models.Domain.objects.filter(name=domain).first()
+        if self.domain is None:
+            raise serializers.ValidationError(_("Domain not found."))
+        if not self.context["request"].user.can_access(self.domain):
+            raise serializers.ValidationError(_("Permission denied."))
+        return value
+
+    def create(self, validated_data):
+        """Create appropriate objects."""
+        creator = self.context["request"].user
+        recipients = validated_data.pop("recipients", None)
+        alias = admin_models.Alias(domain=self.domain, **validated_data)
+        alias.save(creator=creator)
+        alias.set_recipients(recipients)
+        return alias
+
+    def update(self, instance, validated_data):
+        """Update objects."""
+        recipients = validated_data.pop("recipients", None)
+        for key, value in validated_data.items():
+            setattr(instance, key, value)
+        instance.save()
+        instance.set_recipients(recipients)
+        return instance

modoboa/admin/tests/__init__.py

@@ -2,7 +2,7 @@
 
 from .account import AccountTestCase, PermissionsTestCase
 from .alias import AliasTestCase
-from .api import DomainAPITestCase, AccountAPITestCase
+from .api import DomainAPITestCase, AccountAPITestCase, AliasAPITestCase
 from .domain import DomainTestCase
 from .domain_alias import DomainAliasTestCase
 from .export import ExportTestCase
@@ -14,6 +14,7 @@ from .user import ForwardTestCase
 __all__ = [
     'AccountAPITestCase',
     'AccountTestCase',
+    'AliasAPITestCase',
     'AliasTestCase',
     'DomainAPITestCase',
     'DomainTestCase',

modoboa/admin/tests/api.py

@@ -272,3 +272,125 @@ class AccountAPITestCase(ModoAPITestCase):
         self.assertFalse(content["exists"])
         response = self.client.get(url)
         self.assertEqual(response.status_code, 400)
+
+
+class AliasAPITestCase(ModoAPITestCase):
+    """Check Alias API."""
+
+    ALIAS_DATA = {
+        "address": "alias_fromapi@test.com",
+        "recipients": [
+            "user@test.com", "postmaster@test.com", "user_éé@nonlocal.com"
+        ]
+    }
+
+    @classmethod
+    def setUpTestData(cls):
+        """Create test data."""
+        super(AliasAPITestCase, cls).setUpTestData()
+        factories.populate_database()
+        cls.da_token = Token.objects.create(
+            user=core_models.User.objects.get(username="admin@test.com"))
+
+    def test_get_aliases(self):
+        """Retrieve a list of aliases."""
+        url = reverse("external_api:alias-list")
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        response = json.loads(response.content)
+        self.assertEqual(len(response), 3)
+
+    def test_get_alias(self):
+        """Retrieve an alias."""
+        al = models.Alias.objects.get(address="alias@test.com")
+        url = reverse("external_api:alias-detail", args=[al.pk])
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        response = json.loads(response.content)
+        self.assertEqual(response["recipients"], ["user@test.com"])
+
+    def test_create_alias(self):
+        """Try to create a new alias."""
+        url = reverse("external_api:alias-list")
+        response = self.client.post(url, self.ALIAS_DATA, format="json")
+        self.assertEqual(response.status_code, 201)
+
+        alias = json.loads(response.content)
+        alias = models.Alias.objects.filter(pk=alias["pk"]).first()
+        domadmin = core_models.User.objects.get(username="admin@test.com")
+        self.assertTrue(domadmin.can_access(alias))
+        self.assertEqual(alias.aliasrecipient_set.count(), 3)
+        self.assertTrue(alias.aliasrecipient_set.filter(
+            address="user@test.com", r_mailbox__isnull=False).exists())
+        self.assertTrue(alias.aliasrecipient_set.filter(
+            address="postmaster@test.com", r_alias__isnull=False).exists())
+        self.assertTrue(alias.aliasrecipient_set.filter(
+            address="user_éé@nonlocal.com",
+            r_mailbox__isnull=True, r_alias__isnull=True).exists())
+
+        # Create catchall alias
+        data = copy.deepcopy(self.ALIAS_DATA)
+        data["address"] = "@test.com"
+        response = self.client.post(url, data, format="json")
+        self.assertEqual(response.status_code, 201)
+
+    def test_create_alias_as_domadmin(self):
+        """As DomainAdmin, try to create a new alias."""
+        self.client.credentials(
+            HTTP_AUTHORIZATION='Token ' + self.da_token.key)
+        url = reverse("external_api:alias-list")
+        response = self.client.post(url, self.ALIAS_DATA, format="json")
+        self.assertEqual(response.status_code, 201)
+
+        data = copy.deepcopy(self.ALIAS_DATA)
+        data["address"] = "alias_fromapi@test2.com"
+        response = self.client.post(url, data, format="json")
+        self.assertEqual(response.status_code, 400)
+        errors = json.loads(response.content)
+        self.assertIn("address", errors)
+
+    def test_update_alias(self):
+        """Try to update an alias."""
+        alias = models.Alias.objects.get(address="alias@test.com")
+        url = reverse("external_api:alias-detail", args=[alias.pk])
+        data = {
+            "address": "alias@test.com",
+            "recipients": ["user@test.com", "user@nonlocal.com"]
+        }
+        response = self.client.put(url, data, format="json")
+        self.assertEqual(response.status_code, 200)
+
+        alias.refresh_from_db()
+        self.assertEqual(alias.aliasrecipient_set.count(), 2)
+        data = {
+            "address": "alias@test.com",
+            "recipients": ["user@nonlocal.com"]
+        }
+        response = self.client.put(url, data, format="json")
+        self.assertEqual(response.status_code, 200)
+        alias.refresh_from_db()
+        self.assertEqual(alias.aliasrecipient_set.count(), 1)
+
+        data = {
+            "address": "alias@test.com",
+            "recipients": []
+        }
+        response = self.client.put(url, data, format="json")
+        self.assertEqual(response.status_code, 400)
+
+    def test_delete_alias(self):
+        """Try to delete an existing alias."""
+        alias = models.Alias.objects.get(address="alias@test.com")
+        domadmin = core_models.User.objects.get(username="admin@test.com")
+        self.assertTrue(domadmin.can_access(alias))
+        url = reverse("external_api:alias-detail", args=[alias.pk])
+        response = self.client.delete(url)
+        self.assertEqual(response.status_code, 204)
+        self.assertFalse(
+            models.Alias.objects.filter(pk=alias.pk).exists())
+        self.assertFalse(domadmin.can_access(alias))
+        self.assertFalse(
+            models.AliasRecipient.objects.filter(
+                address="user@test.com", alias__address="alias@test.com")
+            .exists()
+        )

modoboa/admin/urls_api.py

@@ -8,5 +8,6 @@ from . import api
 router = routers.SimpleRouter()
 router.register(r"domains", api.DomainViewSet, base_name="domain")
 router.register(r"accounts", api.AccountViewSet, base_name="account")
+router.register(r"aliases", api.AliasViewSet, base_name="alias")
 
 urlpatterns = router.urls

modoboa/lib/fields.py

@@ -49,3 +49,17 @@ class DRFEmailFieldUTF8(serializers.CharField):
         validator = validators.UTF8EmailValidator(
             message=self.error_messages["invalid"])
         self.validators.append(validator)
+
+
+class DRFEmailFieldUTF8AndEmptyUser(serializers.CharField):
+    """Custom DRF email field to support UTF8 and empty local part."""
+
+    default_error_messages = {
+        "invalid": ugettext_lazy("Enter a valid email address.")
+    }
+
+    def __init__(self, **kwargs):
+        super(DRFEmailFieldUTF8AndEmptyUser, self).__init__(**kwargs)
+        validator = validators.UTF8AndEmptyUserEmailValidator(
+            message=self.error_messages["invalid"])
+        self.validators.append(validator)

modoboa/lib/validators.py

@@ -53,11 +53,11 @@ class UTF8EmailValidator(EmailValidator):
 validate_utf8_email = UTF8EmailValidator()
 
 
-class UT8AndEmptyUserEmailValidator(UTF8EmailValidator):
+class UTF8AndEmptyUserEmailValidator(UTF8EmailValidator):
 
     """Same as upper + allows empty local part."""
 
     user_regex = re.compile(
         ur"(^$|" + UTF8EmailValidator.user_regex_raw, re.IGNORECASE)
 
-validate_utf8_and_empty_user_email = UT8AndEmptyUserEmailValidator()
+validate_utf8_and_empty_user_email = UTF8AndEmptyUserEmailValidator()