Pull network related functions out of mopidy.frontends.mpd.server

mopidy/frontends/mpd/server.py

@@ -1,28 +1,13 @@
 import asyncore
 import logging
-import re
-import socket
 import sys
 
 from mopidy import settings
+from mopidy.utils import network
 from .session import MpdSession
 
 logger = logging.getLogger('mopidy.frontends.mpd.server')
 
-def _try_ipv6_socket():
-    """Determine if system really supports IPv6"""
-    if not socket.has_ipv6:
-        return False
-    try:
-        socket.socket(socket.AF_INET6).close()
-        return True
-    except IOError, e:
-        logger.debug(u'Platform supports IPv6, but socket '
-            'creation failed, disabling: %s', e)
-    return False
-
-has_ipv6 = _try_ipv6_socket()
-
 class MpdServer(asyncore.dispatcher):
     """
     The MPD server. Creates a :class:`mopidy.frontends.mpd.session.MpdSession`
@@ -35,15 +20,9 @@ class MpdServer(asyncore.dispatcher):
     def start(self):
         """Start MPD server."""
         try:
-            if has_ipv6:
-                self.create_socket(socket.AF_INET6, socket.SOCK_STREAM)
-                # Explicitly configure socket to work for both IPv4 and IPv6
-                self.socket.setsockopt(
-                    socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
-            else:
-                self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+            self.socket = network.create_socket()
             self.set_reuse_addr()
-            hostname = self._format_hostname(settings.MPD_SERVER_HOSTNAME)
+            hostname = network.format_hostname(settings.MPD_SERVER_HOSTNAME)
             port = settings.MPD_SERVER_PORT
             logger.debug(u'MPD server is binding to [%s]:%s', hostname, port)
             self.bind((hostname, port))
@@ -65,9 +44,3 @@ class MpdServer(asyncore.dispatcher):
     def handle_close(self):
         """Handle end of client connection."""
         self.close()
-
-    def _format_hostname(self, hostname):
-        if (has_ipv6
-            and re.match('\d+.\d+.\d+.\d+', hostname) is not None):
-            hostname = '::ffff:%s' % hostname
-        return hostname

mopidy/utils/network.py

@@ -0,0 +1,36 @@
+import logging
+import re
+import socket
+
+logger = logging.getLogger('mopidy.utils.server')
+
+def _try_ipv6_socket():
+    """Determine if system really supports IPv6"""
+    if not socket.has_ipv6:
+        return False
+    try:
+        socket.socket(socket.AF_INET6).close()
+        return True
+    except IOError, e:
+        logger.debug(u'Platform supports IPv6, but socket '
+            'creation failed, disabling: %s', e)
+    return False
+
+#: Boolean value that indicates if creating an IPv6 socket will succeed.
+has_ipv6 = _try_ipv6_socket()
+
+def create_socket():
+    """Create a TCP socket with or without IPv6 depending on system support"""
+    if has_ipv6:
+        sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+        # Explicitly configure socket to work for both IPv4 and IPv6
+        sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
+    else:
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    return sock
+
+def format_hostname(hostname):
+    """Format hostname for display."""
+    if (has_ipv6 and re.match('\d+.\d+.\d+.\d+', hostname) is not None):
+        hostname = '::ffff:%s' % hostname
+    return hostname

tests/frontends/mpd/server_test.py

@@ -5,29 +5,6 @@ from mopidy.backends.dummy import DummyBackend
 from mopidy.frontends.mpd import server
 from mopidy.mixers.dummy import DummyMixer
 
-class MpdServerTest(unittest.TestCase):
-    def setUp(self):
-        self.backend = DummyBackend.start().proxy()
-        self.mixer = DummyMixer.start().proxy()
-        self.server = server.MpdServer()
-        self.has_ipv6 = server.has_ipv6
-
-    def tearDown(self):
-        self.backend.stop().get()
-        self.mixer.stop().get()
-        server.has_ipv6 = self.has_ipv6
-
-    def test_format_hostname_prefixes_ipv4_addresses_when_ipv6_available(self):
-        server.has_ipv6 = True
-        self.assertEqual(self.server._format_hostname('0.0.0.0'),
-            '::ffff:0.0.0.0')
-        self.assertEqual(self.server._format_hostname('127.0.0.1'),
-            '::ffff:127.0.0.1')
-
-    def test_format_hostname_does_nothing_when_only_ipv4_available(self):
-        server.has_ipv6 = False
-        self.assertEquals(self.server._format_hostname('0.0.0.0'), '0.0.0.0')
-
 class MpdSessionTest(unittest.TestCase):
     def setUp(self):
         self.backend = DummyBackend.start().proxy()

tests/utils/network_test.py

@@ -0,0 +1,19 @@
+import unittest
+
+from mopidy.utils import network
+
+class FormatHostnameTest(unittest.TestCase):
+    def setUp(self):
+        self.has_ipv6 = network.has_ipv6
+
+    def tearDown(self):
+        network.has_ipv6 = self.has_ipv6
+
+    def test_format_hostname_prefixes_ipv4_addresses_when_ipv6_available(self):
+        network.has_ipv6 = True
+        self.assertEqual(network.format_hostname('0.0.0.0'), '::ffff:0.0.0.0')
+        self.assertEqual(network.format_hostname('1.0.0.1'), '::ffff:1.0.0.1')
+
+    def test_format_hostname_does_nothing_when_only_ipv4_available(self):
+        network.has_ipv6 = False
+        self.assertEquals(network.format_hostname('0.0.0.0'), '0.0.0.0')