diff --git a/mopidy/frontends/mpd/server.py b/mopidy/frontends/mpd/server.py index 4e651ddb..927e2a00 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,22 +20,14 @@ 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)) self.listen(1) - logger.info(u'MPD server running at [%s]:%s', - self._format_hostname(settings.MPD_SERVER_HOSTNAME), - settings.MPD_SERVER_PORT) + logger.info(u'MPD server running at [%s]:%s', hostname, port) except IOError, e: logger.error(u'MPD server startup failed: %s' % str(e).decode('utf-8')) @@ -66,9 +43,3 @@ class MpdServer(asyncore.dispatcher): def handle_close(self): """Called by asyncore when the socket is closed.""" 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 ee363aea..b2e27559 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..66229036 --- /dev/null +++ b/tests/utils/network_test.py @@ -0,0 +1,57 @@ +import mock +import socket +import unittest + +from mopidy.utils import network + +from tests import SkipTest + +class FormatHostnameTest(unittest.TestCase): + @mock.patch('mopidy.utils.network.has_ipv6', True) + 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') + + @mock.patch('mopidy.utils.network.has_ipv6', False) + 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') + + +class TryIPv6SocketTest(unittest.TestCase): + @mock.patch('socket.has_ipv6', False) + def test_system_that_claims_no_ipv6_support(self): + self.assertFalse(network._try_ipv6_socket()) + + @mock.patch('socket.has_ipv6', True) + @mock.patch('socket.socket') + def test_system_with_broken_ipv6(self, socket_mock): + socket_mock.side_effect = IOError() + self.assertFalse(network._try_ipv6_socket()) + + @mock.patch('socket.has_ipv6', True) + @mock.patch('socket.socket') + def test_with_working_ipv6(self, socket_mock): + socket_mock.return_value = mock.Mock() + self.assertTrue(network._try_ipv6_socket()) + + +class CreateSocketTest(unittest.TestCase): + @mock.patch('mopidy.utils.network.has_ipv6', False) + @mock.patch('socket.socket') + def test_ipv4_socket(self, socket_mock): + network.create_socket() + self.assertEqual(socket_mock.call_args[0], + (socket.AF_INET, socket.SOCK_STREAM)) + + @mock.patch('mopidy.utils.network.has_ipv6', True) + @mock.patch('socket.socket') + def test_ipv6_socket(self, socket_mock): + network.create_socket() + self.assertEqual(socket_mock.call_args[0], + (socket.AF_INET6, socket.SOCK_STREAM)) + + @SkipTest + def test_ipv6_only_is_set(self): + pass