diff --git a/tests/utils/network/__init__.py b/tests/utils/network/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/utils/network_test.py b/tests/utils/network/connection_test.py similarity index 53% rename from tests/utils/network_test.py rename to tests/utils/network/connection_test.py index 41dacda1..090d7e3c 100644 --- a/tests/utils/network_test.py +++ b/tests/utils/network/connection_test.py @@ -1,5 +1,3 @@ -#encoding: utf-8 - import errno import gobject import logging @@ -12,239 +10,6 @@ from mopidy.utils import network from mock import patch, sentinel, Mock from tests import SkipTest, any_int, any_unicode -class FormatHostnameTest(unittest.TestCase): - @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') - - @patch('mopidy.utils.network.has_ipv6', False) - def test_format_hostname_does_nothing_when_only_ipv4_available(self): - network.has_ipv6 = False - self.assertEqual(network.format_hostname('0.0.0.0'), '0.0.0.0') - - -class TryIPv6SocketTest(unittest.TestCase): - @patch('socket.has_ipv6', False) - def test_system_that_claims_no_ipv6_support(self): - self.assertFalse(network.try_ipv6_socket()) - - @patch('socket.has_ipv6', True) - @patch('socket.socket') - def test_system_with_broken_ipv6(self, socket_mock): - socket_mock.side_effect = IOError() - self.assertFalse(network.try_ipv6_socket()) - - @patch('socket.has_ipv6', True) - @patch('socket.socket') - def test_with_working_ipv6(self, socket_mock): - socket_mock.return_value = Mock() - self.assertTrue(network.try_ipv6_socket()) - - -class CreateSocketTest(unittest.TestCase): - @patch('mopidy.utils.network.has_ipv6', False) - @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)) - - @patch('mopidy.utils.network.has_ipv6', True) - @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 - - -class ServerTest(unittest.TestCase): - def setUp(self): - self.mock = Mock(spec=network.Server) - - def test_init_calls_create_server_socket(self): - network.Server.__init__(self.mock, sentinel.host, - sentinel.port, sentinel.protocol) - self.mock.create_server_socket.assert_called_once_with( - sentinel.host, sentinel.port) - - def test_init_calls_register_server(self): - sock = Mock(spec=socket.SocketType) - sock.fileno.return_value = sentinel.fileno - self.mock.create_server_socket.return_value = sock - - network.Server.__init__(self.mock, sentinel.host, - sentinel.port, sentinel.protocol) - self.mock.register_server_socket.assert_called_once_with( - sentinel.fileno) - - @SkipTest - def test_init_fails_on_fileno_call(self): - sock = Mock(spec=socket.SocketType) - sock.fileno.side_effect = socket.error - self.mock.create_server_socket.return_value = sock - - network.Server.__init__(self.mock, sentinel.host, - sentinel.port, sentinel.protocol) - - def test_init_stores_values_in_attributes(self): - # This need to be a mock and no a sentinel as fileno() is called on it - sock = Mock(spec=socket.SocketType) - self.mock.create_server_socket.return_value = sock - - network.Server.__init__(self.mock, sentinel.host, sentinel.port, - sentinel.protocol, max_connections=sentinel.max_connections, - timeout=sentinel.timeout) - self.assertEqual(sentinel.protocol, self.mock.protocol) - self.assertEqual(sentinel.max_connections, self.mock.max_connections) - self.assertEqual(sentinel.timeout, self.mock.timeout) - self.assertEqual(sock, self.mock.server_socket) - - @patch.object(network, 'create_socket', spec=socket.SocketType) - def test_create_server_socket_sets_up_listener(self, create_socket): - sock = create_socket.return_value - - network.Server.create_server_socket(self.mock, - sentinel.host, sentinel.port) - sock.setblocking.assert_called_once_with(False) - sock.bind.assert_called_once_with((sentinel.host, sentinel.port)) - sock.listen.assert_called_once_with(any_int) - - @SkipTest # FIXME decide behaviour - @patch.object(network, 'create_socket') - def test_create_server_socket_fails(self): - network.create_socket.side_effect = socket.error - network.Server.create_server_socket(self.mock, - sentinel.host, sentinel.port) - - @SkipTest # FIXME decide behaviour - @patch.object(network, 'create_socket', spec=socket.SocketType) - def test_create_server_bind_fails(self): - sock = create_socket.return_value - sock.bind.side_effect = socket.error - - network.Server.create_server_socket(self.mock, - sentinel.host, sentinel.port) - - @SkipTest # FIXME decide behaviour - @patch.object(network, 'create_socket', spec=socket.SocketType) - def test_create_server_listen_fails(self): - sock = create_socket.return_value - sock.listen.side_effect = socket.error - - network.Server.create_server_socket(self.mock, - sentinel.host, sentinel.port) - - @patch.object(gobject, 'io_add_watch', new=Mock()) - def test_register_server_socket_sets_up_io_watch(self): - network.Server.register_server_socket(self.mock, sentinel.fileno) - gobject.io_add_watch.assert_called_once_with(sentinel.fileno, - gobject.IO_IN, self.mock.handle_connection) - - def test_handle_connection(self): - self.mock.accept_connection.return_value = ( - sentinel.sock, sentinel.addr) - self.mock.maximum_connections_exceeded.return_value = False - - self.assertTrue(network.Server.handle_connection( - self.mock, sentinel.fileno, gobject.IO_IN)) - self.mock.accept_connection.assert_called_once_with() - self.mock.maximum_connections_exceeded.assert_called_once_with() - self.mock.init_connection.assert_called_once_with( - sentinel.sock, sentinel.addr) - self.assertEqual(0, self.mock.reject_connection.call_count) - - def test_handle_connection_exceeded_connections(self): - self.mock.accept_connection.return_value = ( - sentinel.sock, sentinel.addr) - self.mock.maximum_connections_exceeded.return_value = True - - self.assertTrue(network.Server.handle_connection( - self.mock, sentinel.fileno, gobject.IO_IN)) - self.mock.accept_connection.assert_called_once_with() - self.mock.maximum_connections_exceeded.assert_called_once_with() - self.mock.reject_connection.assert_called_once_with( - sentinel.sock, sentinel.addr) - self.assertEqual(0, self.mock.init_connection.call_count) - - def test_accept_connection(self): - sock = Mock(spec=socket.SocketType) - sock.accept.return_value = (sentinel.sock, sentinel.addr) - self.mock.server_socket = sock - - sock, addr = network.Server.accept_connection(self.mock) - self.assertEqual(sentinel.sock, sock) - self.assertEqual(sentinel.addr, addr) - - def test_accept_connection_recoverable_error(self): - sock = Mock(spec=socket.SocketType) - self.mock.server_socket = sock - - for error in (errno.EAGAIN, errno.EINTR): - sock.accept.side_effect = socket.error(error, '') - self.assertRaises(network.ShouldRetrySocketCall, - network.Server.accept_connection, self.mock) - - # FIXME decide if this should be allowed to propegate - def test_accept_connection_unrecoverable_error(self): - sock = Mock(spec=socket.SocketType) - self.mock.server_socket = sock - sock.accept.side_effect = socket.error - self.assertRaises(socket.error, - network.Server.accept_connection, self.mock) - - def test_maximum_connections_exceeded(self): - self.mock.max_connections = 10 - - self.mock.number_of_connections.return_value = 11 - self.assertTrue(network.Server.maximum_connections_exceeded(self.mock)) - - self.mock.number_of_connections.return_value = 10 - self.assertTrue(network.Server.maximum_connections_exceeded(self.mock)) - - self.mock.number_of_connections.return_value = 9 - self.assertFalse(network.Server.maximum_connections_exceeded(self.mock)) - - @patch('pykka.registry.ActorRegistry.get_by_class') - def test_number_of_connections(self, get_by_class): - self.mock.protocol = sentinel.protocol - - get_by_class.return_value = [1, 2, 3] - self.assertEqual(3, network.Server.number_of_connections(self.mock)) - - get_by_class.return_value = [] - self.assertEqual(0, network.Server.number_of_connections(self.mock)) - - @patch.object(network, 'Connection', new=Mock()) - def test_init_connection(self): - self.mock.protocol = sentinel.protocol - self.mock.timeout = sentinel.timeout - - network.Server.init_connection(self.mock, sentinel.sock, sentinel.addr) - network.Connection.assert_called_once_with(sentinel.protocol, - sentinel.sock, sentinel.addr, sentinel.timeout) - - def test_reject_connection(self): - sock = Mock(spec=socket.SocketType) - - network.Server.reject_connection(self.mock, sock, - (sentinel.host, sentinel.port)) - sock.close.assert_called_once_with() - - def test_reject_connection_error(self): - sock = Mock(spec=socket.SocketType) - sock.close.side_effect = socket.error - - network.Server.reject_connection(self.mock, sock, - (sentinel.host, sentinel.port)) - sock.close.assert_called_once_with() - - class ConnectionTest(unittest.TestCase): def setUp(self): self.mock = Mock(spec=network.Connection) @@ -751,222 +516,3 @@ class ConnectionTest(unittest.TestCase): self.assertFalse(network.Connection.timeout_callback(self.mock)) self.mock.stop.assert_called_once_with(any_unicode) - - -class LineProtocolTest(unittest.TestCase): - def setUp(self): - self.mock = Mock(spec=network.LineProtocol) - self.mock.terminator = network.LineProtocol.terminator - self.mock.encoding = network.LineProtocol.encoding - - def test_init_stores_values_in_attributes(self): - network.LineProtocol.__init__(self.mock, sentinel.connection) - self.assertEqual(sentinel.connection, self.mock.connection) - self.assertEqual('', self.mock.recv_buffer) - - def test_on_receive_no_new_lines_adds_to_recv_buffer(self): - self.mock.connection = Mock(spec=network.Connection) - self.mock.recv_buffer = '' - self.mock.parse_lines.return_value = [] - - network.LineProtocol.on_receive(self.mock, {'received': 'data'}) - self.assertEqual('data', self.mock.recv_buffer) - self.mock.parse_lines.assert_called_once_with() - self.assertEqual(0, self.mock.on_line_received.call_count) - - def test_on_receive_no_new_lines_toggles_timeout(self): - self.mock.connection = Mock(spec=network.Connection) - self.mock.recv_buffer = '' - self.mock.parse_lines.return_value = [] - - network.LineProtocol.on_receive(self.mock, {'received': 'data'}) - self.mock.connection.disable_timeout.assert_called_once_with() - self.mock.connection.enable_timeout.assert_called_once_with() - - def test_on_receive_no_new_lines_calls_parse_lines(self): - self.mock.connection = Mock(spec=network.Connection) - self.mock.recv_buffer = '' - self.mock.parse_lines.return_value = [] - - network.LineProtocol.on_receive(self.mock, {'received': 'data'}) - self.mock.parse_lines.assert_called_once_with() - self.assertEqual(0, self.mock.on_line_received.call_count) - - def test_on_receive_with_new_line_calls_decode(self): - self.mock.connection = Mock(spec=network.Connection) - self.mock.recv_buffer = '' - self.mock.parse_lines.return_value = [sentinel.line] - - network.LineProtocol.on_receive(self.mock, {'received': 'data\n'}) - self.mock.parse_lines.assert_called_once_with() - self.mock.decode.assert_called_once_with(sentinel.line) - - def test_on_receive_with_new_line_calls_on_recieve(self): - self.mock.connection = Mock(spec=network.Connection) - self.mock.recv_buffer = '' - self.mock.parse_lines.return_value = [sentinel.line] - self.mock.decode.return_value = sentinel.decoded - - network.LineProtocol.on_receive(self.mock, {'received': 'data\n'}) - self.mock.on_line_received.assert_called_once_with(sentinel.decoded) - - def test_on_receive_with_new_lines_calls_on_recieve(self): - self.mock.connection = Mock(spec=network.Connection) - self.mock.recv_buffer = '' - self.mock.parse_lines.return_value = ['line1', 'line2'] - self.mock.decode.return_value = sentinel.decoded - - network.LineProtocol.on_receive(self.mock, - {'received': 'line1\nline2\n'}) - self.assertEqual(2, self.mock.on_line_received.call_count) - - def test_parse_lines_emtpy_buffer(self): - self.mock.recv_buffer = '' - - lines = network.LineProtocol.parse_lines(self.mock) - self.assertRaises(StopIteration, lines.next) - - def test_parse_lines_no_terminator(self): - self.mock.recv_buffer = 'data' - - lines = network.LineProtocol.parse_lines(self.mock) - self.assertRaises(StopIteration, lines.next) - - def test_parse_lines_termintor(self): - self.mock.recv_buffer = 'data\n' - - lines = network.LineProtocol.parse_lines(self.mock) - self.assertEqual('data', lines.next()) - self.assertRaises(StopIteration, lines.next) - self.assertEqual('', self.mock.recv_buffer) - - def test_parse_lines_no_data_before_terminator(self): - self.mock.recv_buffer = '\n' - - lines = network.LineProtocol.parse_lines(self.mock) - self.assertEqual('', lines.next()) - self.assertRaises(StopIteration, lines.next) - self.assertEqual('', self.mock.recv_buffer) - - def test_parse_lines_extra_data_after_terminator(self): - self.mock.recv_buffer = 'data1\ndata2' - - lines = network.LineProtocol.parse_lines(self.mock) - self.assertEqual('data1', lines.next()) - self.assertRaises(StopIteration, lines.next) - self.assertEqual('data2', self.mock.recv_buffer) - - def test_parse_lines_unicode(self): - self.mock.recv_buffer = u'æøå\n'.encode('utf-8') - - lines = network.LineProtocol.parse_lines(self.mock) - self.assertEqual(u'æøå'.encode('utf-8'), lines.next()) - self.assertRaises(StopIteration, lines.next) - self.assertEqual('', self.mock.recv_buffer) - - def test_parse_lines_multiple_lines(self): - self.mock.recv_buffer = 'abc\ndef\nghi\njkl' - - lines = network.LineProtocol.parse_lines(self.mock) - self.assertEqual('abc', lines.next()) - self.assertEqual('def', lines.next()) - self.assertEqual('ghi', lines.next()) - self.assertRaises(StopIteration, lines.next) - self.assertEqual('jkl', self.mock.recv_buffer) - - def test_parse_lines_multiple_calls(self): - self.mock.recv_buffer = 'data' - - lines = network.LineProtocol.parse_lines(self.mock) - self.assertRaises(StopIteration, lines.next) - self.assertEqual('data', self.mock.recv_buffer) - self.mock.recv_buffer += '\n' - - lines = network.LineProtocol.parse_lines(self.mock) - self.assertEqual('data', lines.next()) - self.assertRaises(StopIteration, lines.next) - self.assertEqual('', self.mock.recv_buffer) - - def test_send_lines_called_with_no_lines(self): - self.mock.connection = Mock(spec=network.Connection) - - network.LineProtocol.send_lines(self.mock, []) - self.assertEqual(0, self.mock.encode.call_count) - self.assertEqual(0, self.mock.connection.send.call_count) - - def test_send_lines_calls_join_lines(self): - self.mock.connection = Mock(spec=network.Connection) - self.mock.join_lines.return_value = 'lines' - - network.LineProtocol.send_lines(self.mock, sentinel.lines) - self.mock.join_lines.assert_called_once_with(sentinel.lines) - - def test_send_line_encodes_joined_lines_with_final_terminator(self): - self.mock.connection = Mock(spec=network.Connection) - self.mock.join_lines.return_value = u'lines\n' - - network.LineProtocol.send_lines(self.mock, sentinel.lines) - self.mock.encode.assert_called_once_with(u'lines\n') - - def test_send_lines_sends_encoded_string(self): - self.mock.connection = Mock(spec=network.Connection) - self.mock.join_lines.return_value = 'lines' - self.mock.encode.return_value = sentinel.data - - network.LineProtocol.send_lines(self.mock, sentinel.lines) - self.mock.connection.send.assert_called_once_with(sentinel.data) - - def test_join_lines_returns_empty_string_for_no_lines(self): - self.assertEqual(u'', network.LineProtocol.join_lines(self.mock, [])) - - def test_join_lines_returns_joined_lines(self): - self.assertEqual(u'1\n2\n', network.LineProtocol.join_lines( - self.mock, [u'1', u'2'])) - - def test_decode_calls_decode_on_string(self): - string = Mock() - - network.LineProtocol.decode(self.mock, string) - string.decode.assert_called_once_with(self.mock.encoding) - - def test_decode_plain_ascii(self): - self.assertEqual(u'abc', network.LineProtocol.decode(self.mock, 'abc')) - - def test_decode_utf8(self): - self.assertEqual(u'æøå', network.LineProtocol.decode( - self.mock, u'æøå'.encode('utf-8'))) - - @SkipTest # FIXME decide behaviour - def test_decode_invalid_data(self): - string = Mock() - string.decode.side_effect = UnicodeError - - network.LineProtocol.decode(self.mock, string) - - def test_encode_calls_encode_on_string(self): - string = Mock() - - network.LineProtocol.encode(self.mock, string) - string.encode.assert_called_once_with(self.mock.encoding) - - def test_encode_plain_ascii(self): - self.assertEqual('abc', network.LineProtocol.encode(self.mock, u'abc')) - - def test_encode_utf8(self): - self.assertEqual(u'æøå'.encode('utf-8'), - network.LineProtocol.encode(self.mock, u'æøå')) - - @SkipTest # FIXME decide behaviour - def test_encode_invalid_data(self): - string = Mock() - string.encode.side_effect = UnicodeError - - network.LineProtocol.encode(self.mock, string) - - @SkipTest - def test_host_property(self): - pass - - @SkipTest - def test_port_property(self): - pass diff --git a/tests/utils/network/lineprotocol_test.py b/tests/utils/network/lineprotocol_test.py new file mode 100644 index 00000000..836c3109 --- /dev/null +++ b/tests/utils/network/lineprotocol_test.py @@ -0,0 +1,226 @@ +#encoding: utf-8 + +import unittest + +from mopidy.utils import network + +from mock import sentinel, Mock +from tests import SkipTest + +class LineProtocolTest(unittest.TestCase): + def setUp(self): + self.mock = Mock(spec=network.LineProtocol) + self.mock.terminator = network.LineProtocol.terminator + self.mock.encoding = network.LineProtocol.encoding + + def test_init_stores_values_in_attributes(self): + network.LineProtocol.__init__(self.mock, sentinel.connection) + self.assertEqual(sentinel.connection, self.mock.connection) + self.assertEqual('', self.mock.recv_buffer) + + def test_on_receive_no_new_lines_adds_to_recv_buffer(self): + self.mock.connection = Mock(spec=network.Connection) + self.mock.recv_buffer = '' + self.mock.parse_lines.return_value = [] + + network.LineProtocol.on_receive(self.mock, {'received': 'data'}) + self.assertEqual('data', self.mock.recv_buffer) + self.mock.parse_lines.assert_called_once_with() + self.assertEqual(0, self.mock.on_line_received.call_count) + + def test_on_receive_no_new_lines_toggles_timeout(self): + self.mock.connection = Mock(spec=network.Connection) + self.mock.recv_buffer = '' + self.mock.parse_lines.return_value = [] + + network.LineProtocol.on_receive(self.mock, {'received': 'data'}) + self.mock.connection.disable_timeout.assert_called_once_with() + self.mock.connection.enable_timeout.assert_called_once_with() + + def test_on_receive_no_new_lines_calls_parse_lines(self): + self.mock.connection = Mock(spec=network.Connection) + self.mock.recv_buffer = '' + self.mock.parse_lines.return_value = [] + + network.LineProtocol.on_receive(self.mock, {'received': 'data'}) + self.mock.parse_lines.assert_called_once_with() + self.assertEqual(0, self.mock.on_line_received.call_count) + + def test_on_receive_with_new_line_calls_decode(self): + self.mock.connection = Mock(spec=network.Connection) + self.mock.recv_buffer = '' + self.mock.parse_lines.return_value = [sentinel.line] + + network.LineProtocol.on_receive(self.mock, {'received': 'data\n'}) + self.mock.parse_lines.assert_called_once_with() + self.mock.decode.assert_called_once_with(sentinel.line) + + def test_on_receive_with_new_line_calls_on_recieve(self): + self.mock.connection = Mock(spec=network.Connection) + self.mock.recv_buffer = '' + self.mock.parse_lines.return_value = [sentinel.line] + self.mock.decode.return_value = sentinel.decoded + + network.LineProtocol.on_receive(self.mock, {'received': 'data\n'}) + self.mock.on_line_received.assert_called_once_with(sentinel.decoded) + + def test_on_receive_with_new_lines_calls_on_recieve(self): + self.mock.connection = Mock(spec=network.Connection) + self.mock.recv_buffer = '' + self.mock.parse_lines.return_value = ['line1', 'line2'] + self.mock.decode.return_value = sentinel.decoded + + network.LineProtocol.on_receive(self.mock, + {'received': 'line1\nline2\n'}) + self.assertEqual(2, self.mock.on_line_received.call_count) + + def test_parse_lines_emtpy_buffer(self): + self.mock.recv_buffer = '' + + lines = network.LineProtocol.parse_lines(self.mock) + self.assertRaises(StopIteration, lines.next) + + def test_parse_lines_no_terminator(self): + self.mock.recv_buffer = 'data' + + lines = network.LineProtocol.parse_lines(self.mock) + self.assertRaises(StopIteration, lines.next) + + def test_parse_lines_termintor(self): + self.mock.recv_buffer = 'data\n' + + lines = network.LineProtocol.parse_lines(self.mock) + self.assertEqual('data', lines.next()) + self.assertRaises(StopIteration, lines.next) + self.assertEqual('', self.mock.recv_buffer) + + def test_parse_lines_no_data_before_terminator(self): + self.mock.recv_buffer = '\n' + + lines = network.LineProtocol.parse_lines(self.mock) + self.assertEqual('', lines.next()) + self.assertRaises(StopIteration, lines.next) + self.assertEqual('', self.mock.recv_buffer) + + def test_parse_lines_extra_data_after_terminator(self): + self.mock.recv_buffer = 'data1\ndata2' + + lines = network.LineProtocol.parse_lines(self.mock) + self.assertEqual('data1', lines.next()) + self.assertRaises(StopIteration, lines.next) + self.assertEqual('data2', self.mock.recv_buffer) + + def test_parse_lines_unicode(self): + self.mock.recv_buffer = u'æøå\n'.encode('utf-8') + + lines = network.LineProtocol.parse_lines(self.mock) + self.assertEqual(u'æøå'.encode('utf-8'), lines.next()) + self.assertRaises(StopIteration, lines.next) + self.assertEqual('', self.mock.recv_buffer) + + def test_parse_lines_multiple_lines(self): + self.mock.recv_buffer = 'abc\ndef\nghi\njkl' + + lines = network.LineProtocol.parse_lines(self.mock) + self.assertEqual('abc', lines.next()) + self.assertEqual('def', lines.next()) + self.assertEqual('ghi', lines.next()) + self.assertRaises(StopIteration, lines.next) + self.assertEqual('jkl', self.mock.recv_buffer) + + def test_parse_lines_multiple_calls(self): + self.mock.recv_buffer = 'data' + + lines = network.LineProtocol.parse_lines(self.mock) + self.assertRaises(StopIteration, lines.next) + self.assertEqual('data', self.mock.recv_buffer) + self.mock.recv_buffer += '\n' + + lines = network.LineProtocol.parse_lines(self.mock) + self.assertEqual('data', lines.next()) + self.assertRaises(StopIteration, lines.next) + self.assertEqual('', self.mock.recv_buffer) + + def test_send_lines_called_with_no_lines(self): + self.mock.connection = Mock(spec=network.Connection) + + network.LineProtocol.send_lines(self.mock, []) + self.assertEqual(0, self.mock.encode.call_count) + self.assertEqual(0, self.mock.connection.send.call_count) + + def test_send_lines_calls_join_lines(self): + self.mock.connection = Mock(spec=network.Connection) + self.mock.join_lines.return_value = 'lines' + + network.LineProtocol.send_lines(self.mock, sentinel.lines) + self.mock.join_lines.assert_called_once_with(sentinel.lines) + + def test_send_line_encodes_joined_lines_with_final_terminator(self): + self.mock.connection = Mock(spec=network.Connection) + self.mock.join_lines.return_value = u'lines\n' + + network.LineProtocol.send_lines(self.mock, sentinel.lines) + self.mock.encode.assert_called_once_with(u'lines\n') + + def test_send_lines_sends_encoded_string(self): + self.mock.connection = Mock(spec=network.Connection) + self.mock.join_lines.return_value = 'lines' + self.mock.encode.return_value = sentinel.data + + network.LineProtocol.send_lines(self.mock, sentinel.lines) + self.mock.connection.send.assert_called_once_with(sentinel.data) + + def test_join_lines_returns_empty_string_for_no_lines(self): + self.assertEqual(u'', network.LineProtocol.join_lines(self.mock, [])) + + def test_join_lines_returns_joined_lines(self): + self.assertEqual(u'1\n2\n', network.LineProtocol.join_lines( + self.mock, [u'1', u'2'])) + + def test_decode_calls_decode_on_string(self): + string = Mock() + + network.LineProtocol.decode(self.mock, string) + string.decode.assert_called_once_with(self.mock.encoding) + + def test_decode_plain_ascii(self): + self.assertEqual(u'abc', network.LineProtocol.decode(self.mock, 'abc')) + + def test_decode_utf8(self): + self.assertEqual(u'æøå', network.LineProtocol.decode( + self.mock, u'æøå'.encode('utf-8'))) + + @SkipTest # FIXME decide behaviour + def test_decode_invalid_data(self): + string = Mock() + string.decode.side_effect = UnicodeError + + network.LineProtocol.decode(self.mock, string) + + def test_encode_calls_encode_on_string(self): + string = Mock() + + network.LineProtocol.encode(self.mock, string) + string.encode.assert_called_once_with(self.mock.encoding) + + def test_encode_plain_ascii(self): + self.assertEqual('abc', network.LineProtocol.encode(self.mock, u'abc')) + + def test_encode_utf8(self): + self.assertEqual(u'æøå'.encode('utf-8'), + network.LineProtocol.encode(self.mock, u'æøå')) + + @SkipTest # FIXME decide behaviour + def test_encode_invalid_data(self): + string = Mock() + string.encode.side_effect = UnicodeError + + network.LineProtocol.encode(self.mock, string) + + @SkipTest + def test_host_property(self): + pass + + @SkipTest + def test_port_property(self): + pass diff --git a/tests/utils/network/server_test.py b/tests/utils/network/server_test.py new file mode 100644 index 00000000..c844a487 --- /dev/null +++ b/tests/utils/network/server_test.py @@ -0,0 +1,190 @@ +import errno +import gobject +import socket +import unittest + +from mopidy.utils import network + +from mock import patch, sentinel, Mock +from tests import SkipTest, any_int + +class ServerTest(unittest.TestCase): + def setUp(self): + self.mock = Mock(spec=network.Server) + + def test_init_calls_create_server_socket(self): + network.Server.__init__(self.mock, sentinel.host, + sentinel.port, sentinel.protocol) + self.mock.create_server_socket.assert_called_once_with( + sentinel.host, sentinel.port) + + def test_init_calls_register_server(self): + sock = Mock(spec=socket.SocketType) + sock.fileno.return_value = sentinel.fileno + self.mock.create_server_socket.return_value = sock + + network.Server.__init__(self.mock, sentinel.host, + sentinel.port, sentinel.protocol) + self.mock.register_server_socket.assert_called_once_with( + sentinel.fileno) + + @SkipTest + def test_init_fails_on_fileno_call(self): + sock = Mock(spec=socket.SocketType) + sock.fileno.side_effect = socket.error + self.mock.create_server_socket.return_value = sock + + network.Server.__init__(self.mock, sentinel.host, + sentinel.port, sentinel.protocol) + + def test_init_stores_values_in_attributes(self): + # This need to be a mock and no a sentinel as fileno() is called on it + sock = Mock(spec=socket.SocketType) + self.mock.create_server_socket.return_value = sock + + network.Server.__init__(self.mock, sentinel.host, sentinel.port, + sentinel.protocol, max_connections=sentinel.max_connections, + timeout=sentinel.timeout) + self.assertEqual(sentinel.protocol, self.mock.protocol) + self.assertEqual(sentinel.max_connections, self.mock.max_connections) + self.assertEqual(sentinel.timeout, self.mock.timeout) + self.assertEqual(sock, self.mock.server_socket) + + @patch.object(network, 'create_socket', spec=socket.SocketType) + def test_create_server_socket_sets_up_listener(self, create_socket): + sock = create_socket.return_value + + network.Server.create_server_socket(self.mock, + sentinel.host, sentinel.port) + sock.setblocking.assert_called_once_with(False) + sock.bind.assert_called_once_with((sentinel.host, sentinel.port)) + sock.listen.assert_called_once_with(any_int) + + @SkipTest # FIXME decide behaviour + @patch.object(network, 'create_socket') + def test_create_server_socket_fails(self): + network.create_socket.side_effect = socket.error + network.Server.create_server_socket(self.mock, + sentinel.host, sentinel.port) + + @SkipTest # FIXME decide behaviour + @patch.object(network, 'create_socket', spec=socket.SocketType) + def test_create_server_bind_fails(self): + sock = create_socket.return_value + sock.bind.side_effect = socket.error + + network.Server.create_server_socket(self.mock, + sentinel.host, sentinel.port) + + @SkipTest # FIXME decide behaviour + @patch.object(network, 'create_socket', spec=socket.SocketType) + def test_create_server_listen_fails(self): + sock = create_socket.return_value + sock.listen.side_effect = socket.error + + network.Server.create_server_socket(self.mock, + sentinel.host, sentinel.port) + + @patch.object(gobject, 'io_add_watch', new=Mock()) + def test_register_server_socket_sets_up_io_watch(self): + network.Server.register_server_socket(self.mock, sentinel.fileno) + gobject.io_add_watch.assert_called_once_with(sentinel.fileno, + gobject.IO_IN, self.mock.handle_connection) + + def test_handle_connection(self): + self.mock.accept_connection.return_value = ( + sentinel.sock, sentinel.addr) + self.mock.maximum_connections_exceeded.return_value = False + + self.assertTrue(network.Server.handle_connection( + self.mock, sentinel.fileno, gobject.IO_IN)) + self.mock.accept_connection.assert_called_once_with() + self.mock.maximum_connections_exceeded.assert_called_once_with() + self.mock.init_connection.assert_called_once_with( + sentinel.sock, sentinel.addr) + self.assertEqual(0, self.mock.reject_connection.call_count) + + def test_handle_connection_exceeded_connections(self): + self.mock.accept_connection.return_value = ( + sentinel.sock, sentinel.addr) + self.mock.maximum_connections_exceeded.return_value = True + + self.assertTrue(network.Server.handle_connection( + self.mock, sentinel.fileno, gobject.IO_IN)) + self.mock.accept_connection.assert_called_once_with() + self.mock.maximum_connections_exceeded.assert_called_once_with() + self.mock.reject_connection.assert_called_once_with( + sentinel.sock, sentinel.addr) + self.assertEqual(0, self.mock.init_connection.call_count) + + def test_accept_connection(self): + sock = Mock(spec=socket.SocketType) + sock.accept.return_value = (sentinel.sock, sentinel.addr) + self.mock.server_socket = sock + + sock, addr = network.Server.accept_connection(self.mock) + self.assertEqual(sentinel.sock, sock) + self.assertEqual(sentinel.addr, addr) + + def test_accept_connection_recoverable_error(self): + sock = Mock(spec=socket.SocketType) + self.mock.server_socket = sock + + for error in (errno.EAGAIN, errno.EINTR): + sock.accept.side_effect = socket.error(error, '') + self.assertRaises(network.ShouldRetrySocketCall, + network.Server.accept_connection, self.mock) + + # FIXME decide if this should be allowed to propegate + def test_accept_connection_unrecoverable_error(self): + sock = Mock(spec=socket.SocketType) + self.mock.server_socket = sock + sock.accept.side_effect = socket.error + self.assertRaises(socket.error, + network.Server.accept_connection, self.mock) + + def test_maximum_connections_exceeded(self): + self.mock.max_connections = 10 + + self.mock.number_of_connections.return_value = 11 + self.assertTrue(network.Server.maximum_connections_exceeded(self.mock)) + + self.mock.number_of_connections.return_value = 10 + self.assertTrue(network.Server.maximum_connections_exceeded(self.mock)) + + self.mock.number_of_connections.return_value = 9 + self.assertFalse(network.Server.maximum_connections_exceeded(self.mock)) + + @patch('pykka.registry.ActorRegistry.get_by_class') + def test_number_of_connections(self, get_by_class): + self.mock.protocol = sentinel.protocol + + get_by_class.return_value = [1, 2, 3] + self.assertEqual(3, network.Server.number_of_connections(self.mock)) + + get_by_class.return_value = [] + self.assertEqual(0, network.Server.number_of_connections(self.mock)) + + @patch.object(network, 'Connection', new=Mock()) + def test_init_connection(self): + self.mock.protocol = sentinel.protocol + self.mock.timeout = sentinel.timeout + + network.Server.init_connection(self.mock, sentinel.sock, sentinel.addr) + network.Connection.assert_called_once_with(sentinel.protocol, + sentinel.sock, sentinel.addr, sentinel.timeout) + + def test_reject_connection(self): + sock = Mock(spec=socket.SocketType) + + network.Server.reject_connection(self.mock, sock, + (sentinel.host, sentinel.port)) + sock.close.assert_called_once_with() + + def test_reject_connection_error(self): + sock = Mock(spec=socket.SocketType) + sock.close.side_effect = socket.error + + network.Server.reject_connection(self.mock, sock, + (sentinel.host, sentinel.port)) + sock.close.assert_called_once_with() diff --git a/tests/utils/network/utils_test.py b/tests/utils/network/utils_test.py new file mode 100644 index 00000000..ada1de01 --- /dev/null +++ b/tests/utils/network/utils_test.py @@ -0,0 +1,57 @@ +import socket +import unittest + +from mopidy.utils import network + +from mock import patch, Mock +from tests import SkipTest + +class FormatHostnameTest(unittest.TestCase): + @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') + + @patch('mopidy.utils.network.has_ipv6', False) + def test_format_hostname_does_nothing_when_only_ipv4_available(self): + network.has_ipv6 = False + self.assertEqual(network.format_hostname('0.0.0.0'), '0.0.0.0') + + +class TryIPv6SocketTest(unittest.TestCase): + @patch('socket.has_ipv6', False) + def test_system_that_claims_no_ipv6_support(self): + self.assertFalse(network.try_ipv6_socket()) + + @patch('socket.has_ipv6', True) + @patch('socket.socket') + def test_system_with_broken_ipv6(self, socket_mock): + socket_mock.side_effect = IOError() + self.assertFalse(network.try_ipv6_socket()) + + @patch('socket.has_ipv6', True) + @patch('socket.socket') + def test_with_working_ipv6(self, socket_mock): + socket_mock.return_value = Mock() + self.assertTrue(network.try_ipv6_socket()) + + +class CreateSocketTest(unittest.TestCase): + @patch('mopidy.utils.network.has_ipv6', False) + @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)) + + @patch('mopidy.utils.network.has_ipv6', True) + @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