diff --git a/mopidy/frontends/mpd/server.py b/mopidy/frontends/mpd/server.py index 1be46ef4..87a1cd0a 100644 --- a/mopidy/frontends/mpd/server.py +++ b/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 diff --git a/mopidy/utils/network.py b/mopidy/utils/network.py new file mode 100644 index 00000000..1dedf7d7 --- /dev/null +++ b/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 diff --git a/tests/frontends/mpd/server_test.py b/tests/frontends/mpd/server_test.py index 32e90450..76bf9e33 100644 --- a/tests/frontends/mpd/server_test.py +++ b/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() diff --git a/tests/utils/network_test.py b/tests/utils/network_test.py new file mode 100644 index 00000000..6217e910 --- /dev/null +++ b/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')