Merge pull request #1339 from fyfe/fix/mx-check-ipv6-support

Add IPv6 support to mx checks

modoboa/admin/lib.py

@@ -1,4 +1,4 @@
-# coding: utf-8
+# -*- coding: utf-8 -*-
 
 """Internal library for admin."""
 
@@ -7,14 +7,17 @@ from __future__ import unicode_literals
 from functools import wraps
 import ipaddress
 from itertools import chain
+import logging
 import random
 import socket
 import string
 
+from dns.name import IDNA_2008_UTS_46
 import dns.resolver
 
 from django.contrib.contenttypes.models import ContentType
 from django.db.models import Q
+from django.utils.encoding import smart_text
 from django.utils.translation import ugettext as _
 
 from modoboa.core.models import User
@@ -170,40 +173,56 @@ def import_dlist(user, row, formopts):
     _import_alias(user, row)
 
 
-def get_domain_mx_list(name):
+def get_domain_mx_list(domain):
     """Return a list of MX IP address for domain."""
     result = []
+    logger = logging.getLogger("modoboa.admin")
     try:
-        answers = dns.resolver.query(name, "MX")
-    except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN,
-            dns.resolver.NoNameservers):
-        return result
-    for answer in answers:
-        try:
-            # work if .exchange is a name or IP
-            address = socket.gethostbyname(str(answer.exchange))
-        except socket.gaierror:
-            pass
-        else:
+        dns_answers = dns.resolver.query(domain, "MX")
+    except dns.resolver.NXDOMAIN as e:
+        logger.error(_("No DNS records found for %s") % domain, exc_info=e)
+    except dns.resolver.NoAnswer as e:
+        logger.error(_("No MX record for %s") % domain, exc_info=e)
+    except dns.resolver.NoNameservers as e:
+        logger.error(_("No working name servers found"), exc_info=e)
+    except dns.resolver.Timeout as e:
+        logger.warning(
+            _("DNS resolution timeout, unable to query %s at the moment") % domain,
+            exc_info=e)
+    else:
+        for dns_answer in dns_answers:
             try:
-                # we must have a valid IP
-                address = ipaddress.ip_address("{}".format(address))
-            except ValueError:
-                pass
+                mx_domain = dns_answer.exchange.to_unicode(
+                    omit_final_dot=True, idna_codec=IDNA_2008_UTS_46)
+                ip_answers = socket.getaddrinfo(
+                    mx_domain, 25, 0, socket.SOCK_STREAM)
+            except socket.gaierror as e:
+                logger.warning(
+                    _("Unable to lookup ip addresses for %s; %s") % (
+                        mx_domain, str(e)),
+                    exc_info=e)
             else:
-                result.append((str(answer.exchange), address))
+                for ip_answer in ip_answers:
+                    try:
+                        mx_ip = ipaddress.ip_address(
+                            smart_text(ip_answer[4][0]))
+                    except ValueError as e:
+                        logger.warning(
+                            _("Invalid IP address format for %s; %s") % (
+                                mx_domain, smart_text(ip_answer[4][0])),
+                            exc_info=e)
+                    else:
+                        result.append((mx_domain, mx_ip))
     return result
 
 
 def domain_has_authorized_mx(name):
     """Check if domain has authorized mx record at least."""
     valid_mxs = param_tools.get_global_parameter("valid_mxs")
-    valid_mxs = [ipaddress.ip_network(u"{}".format(v.strip()))
+    valid_mxs = [ipaddress.ip_network(smart_text(v.strip()))
                  for v in valid_mxs.split() if v.strip()]
     domain_mxs = get_domain_mx_list(name)
-    if len(domain_mxs) == 0:
-        return True
-    for mx_addr, mx_ip_addr in domain_mxs:
+    for mx_addr, mx_ip_addr in domain_mxs:  # noqa:B007
         for subnet in valid_mxs:
             if mx_ip_addr in subnet:
                 return True

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

@@ -1,3 +1,5 @@
+# -*- coding: utf-8 -*-
+
 """Management command to check defined domains."""
 
 from __future__ import print_function, unicode_literals
@@ -12,6 +14,7 @@ from django.core import mail
 from django.core.mail import EmailMessage
 from django.core.management.base import BaseCommand
 from django.template.loader import render_to_string
+from django.utils.encoding import smart_text
 from django.utils.functional import cached_property
 from django.utils.translation import ugettext as _
 
@@ -23,7 +26,7 @@ from modoboa.parameters import tools as param_tools
 class CheckMXRecords(BaseCommand):
     """Command class."""
 
-    help = "Check defined domains."
+    help = "Check defined domains."  # noqa:A003
 
     @cached_property
     def providers(self):
@@ -41,7 +44,7 @@ class CheckMXRecords(BaseCommand):
     def valid_mxs(self):
         """Return valid MXs set in admin."""
         valid_mxs = param_tools.get_global_parameter("valid_mxs")
-        return [ipaddress.ip_network(u"{}".format(v.strip()))
+        return [ipaddress.ip_network(smart_text(v.strip()))
                 for v in valid_mxs.split() if v.strip()]
 
     def add_arguments(self, parser):
@@ -69,13 +72,21 @@ class CheckMXRecords(BaseCommand):
     def query_dnsbl(self, mx_list, provider):
         """Check given IP against given DNSBL provider."""
         results = {}
-        for mx in mx_list:
-            reverse = ".".join(reversed(mx.address.split(".")))
+        for ip, mxs in mx_list.items():
+            try:
+                ip = ipaddress.ip_address(smart_text(ip))
+            except ValueError as e:
+                continue
+            else:
+                delim = "." if ip.version == 4 else ":"
+                reverse = delim.join(ip.exploded.split(delim)[::-1])
             pattern = "{}.{}.".format(reverse, provider)
             try:
-                results[mx] = socket.gethostbyname(pattern)
+                result = socket.gethostbyname(pattern)
             except socket.gaierror:
-                results[mx] = False
+                result = False
+            for mx in mxs:
+                results[mx] = result
         return provider, results
 
     def store_dnsbl_result(self, domain, provider, results, **options):
@@ -188,8 +199,16 @@ class CheckMXRecords(BaseCommand):
             options["no_dnsbl"] is True)
         if condition or not mx_list:
             return
+
+        mx_by_ip = {}
+        for mx in mx_list:
+            if mx.address not in mx_by_ip:
+                mx_by_ip[mx.address] = [mx]
+            elif mx not in mx_by_ip[mx.address]:
+                mx_by_ip[mx.address].append(mx)
+
         jobs = [
-            gevent.spawn(self.query_dnsbl, mx_list, provider)
+            gevent.spawn(self.query_dnsbl, mx_by_ip, provider)
             for provider in self.providers]
         gevent.joinall(jobs, timeout)
         for job in jobs:

modoboa/admin/tests/test_api.py

@@ -1,4 +1,5 @@
-# coding: utf-8
+# -*- coding: utf-8 -*-
+
 """Admin API related tests."""
 
 from __future__ import unicode_literals
@@ -27,7 +28,7 @@ class DomainAPITestCase(ModoAPITestCase):
     """Check API."""
 
     @classmethod
-    def setUpTestData(cls):
+    def setUpTestData(cls):  # noqa: N802
         """Create test data."""
         super(DomainAPITestCase, cls).setUpTestData()
         factories.populate_database()
@@ -73,11 +74,11 @@ class DomainAPITestCase(ModoAPITestCase):
         self.assertEqual(response.status_code, 403)
 
     @patch.object(dns.resolver.Resolver, "query")
-    @patch("socket.gethostbyname")
-    def test_create_domain_with_mx_check(self, mock_gethostbyname, mock_query):
+    @patch("socket.getaddrinfo")
+    def test_create_domain_with_mx_check(self, mock_getaddrinfo, mock_query):
         """Check domain creation when MX check is activated."""
         self.set_global_parameter("enable_admin_limits", False, app="limits")
-        self.set_global_parameter("valid_mxs", "1.2.3.4")
+        self.set_global_parameter("valid_mxs", "192.0.2.1 2001:db8::1")
         self.set_global_parameter("domains_must_have_authorized_mx", True)
         reseller = core_factories.UserFactory(
             username="reseller", groups=("Resellers", ))
@@ -85,19 +86,19 @@ class DomainAPITestCase(ModoAPITestCase):
         self.client.credentials(HTTP_AUTHORIZATION='Token ' + token.key)
 
         url = reverse("api:domain-list")
-        mock_query.return_value = [utils.FakeDNSAnswer("mail.ok.com")]
-        mock_gethostbyname.return_value = "1.2.3.5"
+        mock_query.side_effect = utils.mock_dns_query_result
+        mock_getaddrinfo.side_effect = utils.mock_ip_query_result
         response = self.client.post(
-            url, {"name": "test3.com", "quota": 0, "default_mailbox_quota": 10}
+            url, {"name": "no-mx.example.com", "quota": 0, "default_mailbox_quota": 10}
         )
         self.assertEqual(response.status_code, 400)
         self.assertEqual(
             response.json()["name"][0],
             "No authorized MX record found for this domain")
 
-        mock_gethostbyname.return_value = "1.2.3.4"
+        mock_getaddrinfo.side_effect = utils.mock_ip_query_result
         response = self.client.post(
-            url, {"name": "test3.com", "quota": 0, "default_mailbox_quota": 10}
+            url, {"name": "test4.com", "quota": 0, "default_mailbox_quota": 10}
         )
         self.assertEqual(response.status_code, 201)
 
@@ -137,7 +138,7 @@ class DomainAliasAPITestCase(ModoAPITestCase):
     """Check DomainAlias API."""
 
     @classmethod
-    def setUpTestData(cls):
+    def setUpTestData(cls):  # noqa: N802
         """Create test data."""
         super(DomainAliasAPITestCase, cls).setUpTestData()
         factories.populate_database()
@@ -231,7 +232,7 @@ class AccountAPITestCase(ModoAPITestCase):
     }
 
     @classmethod
-    def setUpTestData(cls):
+    def setUpTestData(cls):  # noqa: N802
         """Create test data."""
         super(AccountAPITestCase, cls).setUpTestData()
         factories.populate_database()
@@ -544,7 +545,7 @@ class AliasAPITestCase(ModoAPITestCase):
     }
 
     @classmethod
-    def setUpTestData(cls):
+    def setUpTestData(cls):  # noqa: N802
         """Create test data."""
         super(AliasAPITestCase, cls).setUpTestData()
         cls.localconfig.parameters.set_value(
@@ -667,7 +668,7 @@ class SenderAddressAPITestCase(ModoAPITestCase):
     """Check SenderAddress API."""
 
     @classmethod
-    def setUpTestData(cls):
+    def setUpTestData(cls):  # noqa: N802
         """Create test data."""
         super(SenderAddressAPITestCase, cls).setUpTestData()
         cls.localconfig.parameters.set_value(

modoboa/admin/tests/test_domain.py

@@ -1,4 +1,4 @@
-# coding: utf-8
+# -*- coding: utf-8 -*-
 
 """Domain related test cases."""
 
@@ -22,7 +22,7 @@ class DomainTestCase(ModoTestCase):
     """Test case for Domain."""
 
     @classmethod
-    def setUpTestData(cls):
+    def setUpTestData(cls):  # noqa: N802
         """Create test data."""
         super(DomainTestCase, cls).setUpTestData()
         factories.populate_database()
@@ -199,21 +199,21 @@ class DomainTestCase(ModoTestCase):
         self.assertContains(response, "value=\"50\"")
 
     @patch.object(dns.resolver.Resolver, "query")
-    @patch("socket.gethostbyname")
-    def test_create_and_check_mx(self, mock_gethostbyname, mock_query):
+    @patch("socket.getaddrinfo")
+    def test_create_and_check_mx(self, mock_getaddrinfo, mock_query):
         """Check for authorized MX record."""
         reseller = core_factories.UserFactory(
             username="reseller", groups=("Resellers", ))
         self.client.force_login(reseller)
 
         self.set_global_parameter("enable_admin_limits", False, app="limits")
-        self.set_global_parameter("valid_mxs", "1.2.3.4")
+        self.set_global_parameter("valid_mxs", "192.0.2.1 2001:db8::1")
         self.set_global_parameter("domains_must_have_authorized_mx", True)
 
-        mock_query.return_value = [utils.FakeDNSAnswer("mail.ok.com")]
-        mock_gethostbyname.return_value = "1.2.3.5"
+        mock_query.side_effect = utils.mock_dns_query_result
+        mock_getaddrinfo.side_effect = utils.mock_ip_query_result
         values = {
-            "name": "pouet.com", "quota": 0, "default_mailbox_quota": 0,
+            "name": "no-mx.example.com", "quota": 0, "default_mailbox_quota": 0,
             "create_dom_admin": True, "dom_admin_username": "toto",
             "create_aliases": True, "type": "domain", "stepid": "step3",
             "with_mailbox": True
@@ -222,7 +222,8 @@ class DomainTestCase(ModoTestCase):
         self.assertFalse(
             Domain.objects.filter(name=values["name"]).exists())
 
-        mock_gethostbyname.return_value = "1.2.3.4"
+        values["name"] = "pouet.com"
+        mock_getaddrinfo.side_effect = utils.mock_ip_query_result
         self.ajax_post(reverse("admin:domain_add"), values)
 
     def test_modify(self):

modoboa/admin/tests/test_import_.py

@@ -1,9 +1,9 @@
-# coding: utf-8
+# -*- coding: utf-8 -*-
 
 from __future__ import unicode_literals
 
-from mock import patch
 import dns.resolver
+from mock import patch
 
 from django.core.files.base import ContentFile
 from django.urls import reverse
@@ -20,7 +20,7 @@ from ..models import Domain, Alias, DomainAlias
 class ImportTestCase(ModoTestCase):
 
     @classmethod
-    def setUpTestData(cls):
+    def setUpTestData(cls):  # noqa: N802
         """Create test data."""
         super(ImportTestCase, cls).setUpTestData()
         cls.localconfig.parameters.set_value(
@@ -80,19 +80,19 @@ domainalias; domalias1.com; domain1.com; True
             "Default mailbox quota cannot be greater than domain quota")
 
     @patch.object(dns.resolver.Resolver, "query")
-    @patch("socket.gethostbyname")
-    def test_domain_import_with_mx_check(self, mock_gethostbyname, mock_query):
+    @patch("socket.getaddrinfo")
+    def test_domain_import_with_mx_check(self, mock_getaddrinfo, mock_query):
         """Check domain import when MX check is enabled."""
         reseller = core_factories.UserFactory(
             username="reseller", groups=("Resellers", ))
         self.client.force_login(reseller)
-        self.set_global_parameter("valid_mxs", "1.2.3.4")
+        self.set_global_parameter("valid_mxs", "192.0.2.1 2001:db8::1")
         self.set_global_parameter("domains_must_have_authorized_mx", True)
 
-        mock_query.return_value = [utils.FakeDNSAnswer("mail.ok.com")]
-        mock_gethostbyname.return_value = "1.2.3.5"
+        mock_query.side_effect = utils.mock_dns_query_result
+        mock_getaddrinfo.side_effect = utils.mock_ip_query_result
         f = ContentFile(
-            b"domain; domain1.com; 100; 1; True", name="domains.csv")
+            b"domain; test3.com; 100; 1; True", name="domains.csv")
         resp = self.client.post(
             reverse("admin:domain_import"), {
                 "sourcefile": f
@@ -100,8 +100,9 @@ domainalias; domalias1.com; domain1.com; True
         )
         self.assertContains(resp, "No authorized MX record found for domain")
 
-        mock_gethostbyname.return_value = "1.2.3.4"
-        f.seek(0)
+        mock_getaddrinfo.side_effect = utils.mock_ip_query_result
+        f = ContentFile(
+            b"domain; domain1.com; 100; 1; True", name="domains.csv")
         resp = self.client.post(
             reverse("admin:domain_import"), {
                 "sourcefile": f

modoboa/admin/tests/test_mx.py

@@ -1,18 +1,25 @@
+# -*- coding: utf-8 -*-
+
 """DNSBL related tests."""
 
 from __future__ import unicode_literals
 
+import dns.resolver
 from mock import patch
+from testfixtures import LogCapture
 
 from django.core import mail
 from django.core import management
-from django.urls import reverse
 from django.test import override_settings
+from django.urls import reverse
+from django.utils.translation import ugettext as _
 
 from modoboa.core import factories as core_factories
 from modoboa.lib.tests import ModoTestCase
 
+from . import utils
 from .. import factories
+from ..lib import get_domain_mx_list
 from .. import models
 
 
@@ -20,30 +27,38 @@ class MXTestCase(ModoTestCase):
     """TestCase for DNSBL related features."""
 
     @classmethod
-    def setUpTestData(cls):
+    def setUpTestData(cls):  # noqa:N802
         """Create some data."""
         super(MXTestCase, cls).setUpTestData()
         cls.domain = factories.DomainFactory(name="modoboa.org")
         # should not exist
-        cls.bad_domain = factories.DomainFactory(name="pouet.com")
+        cls.bad_domain = factories.DomainFactory(name="does-not-exist.example.com")
         # Add domain admin with mailbox
         mb = factories.MailboxFactory(
             address="admin", domain=cls.bad_domain,
-            user__username="admin@pouet.com",
+            user__username="admin@does-not-exist.example.com",
             user__groups=("DomainAdmins", )
         )
         cls.bad_domain.add_admin(mb.user)
         # Add domain admin with no mailbox
         admin = core_factories.UserFactory(
-            username="admin2@pouet.com", groups=("DomainAdmins", ))
+            username="admin2@does-not-exist.example.com", groups=("DomainAdmins", ))
         cls.bad_domain.add_admin(admin)
 
-        cls.localconfig.parameters.set_value("valid_mxs", "127.0.0.1")
+        cls.localconfig.parameters.set_value(
+            "valid_mxs", "192.0.2.1 2001:db8::1")
         cls.localconfig.save()
         models.MXRecord.objects.all().delete()
 
-    def test_management_command(self):
+    @patch("gevent.socket.gethostbyname")
+    @patch("socket.getaddrinfo")
+    @patch.object(dns.resolver.Resolver, "query")
+    def test_management_command(
+         self, mock_query, mock_getaddrinfo, mock_g_gethostbyname):
         """Check that command works fine."""
+        mock_query.side_effect = utils.mock_dns_query_result
+        mock_getaddrinfo.side_effect = utils.mock_ip_query_result
+        mock_g_gethostbyname.return_value = "1.2.3.4"
         self.assertEqual(models.MXRecord.objects.count(), 0)
         management.call_command("modo", "check_mx", "--no-dnsbl", "--ttl=0")
         self.assertTrue(
@@ -51,20 +66,27 @@ class MXTestCase(ModoTestCase):
 
         # we passed a ttl to 0. this will reset the cache this time
         qs = models.MXRecord.objects.filter(domain=self.domain)
-        id = qs[0].id
+        id_ = qs[0].id
         management.call_command("modo", "check_mx", "--no-dnsbl", "--ttl=7200")
         qs = models.MXRecord.objects.filter(domain=self.domain)
-        self.assertNotEqual(id, qs[0].id)
-        id = qs[0].id
+        self.assertNotEqual(id_, qs[0].id)
+        id_ = qs[0].id
 
         # assume that mxrecords ids are the same. means that we taking care of
         # ttl
         management.call_command("modo", "check_mx", "--no-dnsbl")
         qs = models.MXRecord.objects.filter(domain=self.domain)
-        self.assertEqual(id, qs[0].id)
+        self.assertEqual(id_, qs[0].id)
 
-    def test_single_domain_update(self):
+    @patch("gevent.socket.gethostbyname")
+    @patch("socket.getaddrinfo")
+    @patch.object(dns.resolver.Resolver, "query")
+    def test_single_domain_update(
+         self, mock_query, mock_getaddrinfo, mock_g_gethostbyname):
         """Update only one domain."""
+        mock_query.side_effect = utils.mock_dns_query_result
+        mock_getaddrinfo.side_effect = utils.mock_ip_query_result
+        mock_g_gethostbyname.return_value = "1.2.3.4"
         management.call_command(
             "modo", "check_mx", "--domain", self.domain.name)
         self.assertTrue(
@@ -81,25 +103,62 @@ class MXTestCase(ModoTestCase):
         management.call_command(
             "modo", "check_mx", "--domain", "toto.com")
 
+    @patch("socket.getaddrinfo")
+    @patch.object(dns.resolver.Resolver, "query")
+    def test_get_domain_mx_list_logging(self, mock_query, mock_getaddrinfo):
+        mock_query.side_effect = utils.mock_dns_query_result
+        mock_getaddrinfo.side_effect = utils.mock_ip_query_result
+        with LogCapture("modoboa.admin") as log:
+            get_domain_mx_list("does-not-exist.example.com")
+            get_domain_mx_list("no-mx.example.com")
+            get_domain_mx_list("no-ns-servers.example.com")
+            get_domain_mx_list("timeout.example.com")
+            get_domain_mx_list("no-lookup.example.com")
+            get_domain_mx_list("bad-response.example.com")
+        log.check(
+            ("modoboa.admin", "ERROR",
+                _("No DNS records found for %s")
+                % "does-not-exist.example.com"),
+            ("modoboa.admin", "ERROR",
+                _("No MX record for %s") % "no-mx.example.com"),
+            ("modoboa.admin", "ERROR", _("No working name servers found")),
+            ("modoboa.admin", "WARNING",
+                _("DNS resolution timeout, unable to query %s at the moment")
+                % "timeout.example.com"),
+            ("modoboa.admin", "WARNING",
+                _("Unable to lookup ip addresses for %s; %s")
+                % ("does-not-exist.example.com", "")),
+            ("modoboa.admin", "WARNING",
+                _("Invalid IP address format for %s; %s")
+                % ("bad-response.example.com", "BAD")),
+            )
+
 
 @override_settings(DNSBL_PROVIDERS=["zen.spamhaus.org"])
 class DNSBLTestCase(ModoTestCase):
     """TestCase for DNSBL related features."""
 
     @classmethod
-    def setUpTestData(cls):
+    def setUpTestData(cls):  # noqa:N802
         """Create some data."""
         super(DNSBLTestCase, cls).setUpTestData()
         cls.domain = factories.DomainFactory(name="modoboa.org")
-        factories.DomainFactory(name="pouet.com")  # should not exist
+        factories.DomainFactory(name="does-not-exist.example.com")
         cls.domain2 = factories.DomainFactory(
             name="test.localhost")  # Should not be checked
         cls.domain3 = factories.DomainFactory(
             name="modoboa.com", enable_dns_checks=False)
         models.DNSBLResult.objects.all().delete()
 
-    def test_management_command(self):
+    @patch("gevent.socket.gethostbyname")
+    @patch("socket.getaddrinfo")
+    @patch.object(dns.resolver.Resolver, "query")
+    def test_management_command(
+         self, mock_query, mock_getaddrinfo, mock_g_gethostbyname):
         """Check that command works fine."""
+        mock_query.side_effect = utils.mock_dns_query_result
+        mock_getaddrinfo.side_effect = utils.mock_ip_query_result
+        mock_g_gethostbyname.return_value = "1.2.3.4"
         self.assertEqual(models.DNSBLResult.objects.count(), 0)
         management.call_command("modo", "check_mx")
         self.assertTrue(
@@ -113,15 +172,30 @@ class DNSBLTestCase(ModoTestCase):
         self.assertTrue(self.domain2.uses_a_reserved_tld)
 
     @patch("gevent.socket.gethostbyname")
-    def test_notifications(self, mock_gethostbyname):
+    @patch("socket.getaddrinfo")
+    @patch.object(dns.resolver.Resolver, "query")
+    def test_notifications(
+         self, mock_query, mock_getaddrinfo, mock_g_gethostbyname):
         """Check notifications."""
-        mock_gethostbyname.return_value = "1.2.3.4"
+        mock_query.side_effect = utils.mock_dns_query_result
+        mock_getaddrinfo.side_effect = utils.mock_ip_query_result
+        mock_g_gethostbyname.return_value = "1.2.3.4"
         management.call_command(
             "modo", "check_mx", "--email", "user@example.test")
+        if len(mail.outbox) != 2:
+            for message in mail.outbox:
+                print(message.subject)
         self.assertEqual(len(mail.outbox), 2)
 
-    def test_management_command_no_dnsbl(self):
+    @patch("gevent.socket.gethostbyname")
+    @patch("socket.getaddrinfo")
+    @patch.object(dns.resolver.Resolver, "query")
+    def test_management_command_no_dnsbl(
+         self, mock_query, mock_getaddrinfo, mock_g_gethostbyname):
         """Check that command works fine without dnsbl."""
+        mock_query.side_effect = utils.mock_dns_query_result
+        mock_getaddrinfo.side_effect = utils.mock_ip_query_result
+        mock_g_gethostbyname.return_value = "1.2.3.4"
         self.assertEqual(models.DNSBLResult.objects.count(), 0)
         management.call_command("modo", "check_mx", "--no-dnsbl")
         self.assertFalse(

modoboa/admin/tests/utils.py

@@ -1,8 +1,62 @@
+# -*- coding: utf-8 -*-
+
 """Test utilities."""
 
+from __future__ import unicode_literals
 
-class FakeDNSAnswer(object):
-    """Fake answer."""
+import socket
 
-    def __init__(self, exchange):
-        self.exchange = exchange
+from dns.name import Name
+from dns.rdtypes.ANY.MX import MX
+from dns.resolver import NoAnswer, NoNameservers, NXDOMAIN, Timeout
+
+_MX_RECORD_1 = MX("IN", "MX", 10, Name("mx.example.com".split(".")))
+_MX_RECORD_2 = MX("IN", "MX", 10, Name("mx2.example.com".split(".")))
+_BAD_MX_RECORD = MX("IN", "MX", 10, Name("bad-response.example.com".split(".")))
+_DNE_MX_RECORD = MX("IN", "MX", 10, Name("does-not-exist.example.com".split(".")))
+_MX_RECORDS = [_MX_RECORD_1]
+
+_IPV4_RECORD_1 = (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.0.2.1", 25))
+_IPV4_RECORD_2 = (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.0.2.2", 25))
+_IPV6_RECORD = (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("2001:db8::1", 25))
+_BAD_IP_RECORD = (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("BAD", 25))
+_IP_RECORDS = [_IPV4_RECORD_1, _IPV6_RECORD]
+
+_POSSIBLE_DNS_RESULTS = {
+    "test3.com": [_MX_RECORD_2],
+    "no-mx.example.com": NoAnswer(),
+    "does-not-exist.example.com": NXDOMAIN(),
+    "timeout.example.com": Timeout(),
+    "no-ns-servers.example.com": NoNameservers(),
+    "bad-response.example.com": [_BAD_MX_RECORD],
+    "no-lookup.example.com": [_DNE_MX_RECORD],
+}
+
+_POSSIBLE_IP_RESULTS = {
+    "test3.com": [_IPV4_RECORD_2],
+    "mx2.example.com": [_IPV4_RECORD_2],
+    "bad-response.example.com": [_BAD_IP_RECORD],
+    "does-not-exist.example.com": socket.gaierror(),
+    "192.0.2.254": socket.gaierror(),
+    "2001:db8:254": socket.gaierror(),
+}
+
+
+def mock_dns_query_result(qname, *args, **kwargs):
+    if qname in _POSSIBLE_DNS_RESULTS:
+        if isinstance(_POSSIBLE_DNS_RESULTS[qname], Exception):
+            raise _POSSIBLE_DNS_RESULTS[qname]
+        else:
+            return _POSSIBLE_DNS_RESULTS[qname]
+
+    return _MX_RECORDS
+
+
+def mock_ip_query_result(host, port, *args, **kwargs):
+    if host in _POSSIBLE_IP_RESULTS:
+        if isinstance(_POSSIBLE_IP_RESULTS[host], Exception):
+            raise _POSSIBLE_IP_RESULTS[host]
+        else:
+            return _POSSIBLE_IP_RESULTS[host]
+
+    return _IP_RECORDS

test_project/test_project/settings.py

@@ -237,6 +237,7 @@ CKEDITOR_CONFIGS = {
 
 LOGGING = {
     'version': 1,
+    'disable_existing_loggers': False,
     'formatters': {
         'syslog': {
             'format': '%(name)s: %(levelname)s %(message)s'