Add some tests for misc.ipc.

tests/unit/misc/test_ipc.py

@@ -0,0 +1,258 @@
+# vim: ft=python fileencoding=utf-8 sts=4 sw=4 et:
+
+# Copyright 2015 Florian Bruhin (The Compiler) <mail@qutebrowser.org>
+#
+# This file is part of qutebrowser.
+#
+# qutebrowser is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# qutebrowser is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with qutebrowser.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Tests for qutebrowser.misc.ipc."""
+
+import getpass
+import collections
+import logging
+from unittest import mock
+
+import pytest
+from PyQt5.QtNetwork import QLocalServer, QLocalSocket
+from PyQt5.QtTest import QSignalSpy
+from PyQt5.QtCore import QIODevice
+
+from qutebrowser.misc import ipc
+
+
+Args = collections.namedtuple('Args', 'basedir')
+
+
+@pytest.yield_fixture
+def ipc_server(qapp, qtbot):
+    server = ipc.IPCServer('qutebrowser-test')
+    yield server
+    if (server._socket is not None and
+            server._socket.state() != QLocalSocket.UnconnectedState):
+        with qtbot.waitSignal(server._socket.disconnected, raising=False):
+            server._socket.abort()
+    try:
+        server.shutdown()
+    except ipc.Error:
+        pass
+
+
+@pytest.yield_fixture
+def qlocalserver(qapp):
+    server = QLocalServer()
+    yield server
+    server.close()
+    server.deleteLater()
+
+
+@pytest.yield_fixture
+def qlocalsocket(qapp):
+    socket = QLocalSocket()
+    yield socket
+    socket.disconnectFromServer()
+    if socket.state() != QLocalSocket.UnconnectedState:
+        disconnected = socket.waitForDisconnected(100)
+        assert disconnected
+
+
+def test_getpass_getuser():
+    """Make sure getpass.getuser() returns something sensible."""
+    assert getpass.getuser()
+
+
+@pytest.mark.parametrize('username, basedir, expected', [
+    ('florian', None, 'qutebrowser-florian'),
+    ('florian', '/x', 'qutebrowser-florian-cc8755609ad61864910f145119713de9'),
+])
+def test_get_socketname(username, basedir, expected):
+    assert ipc._get_socketname(basedir, user=username) == expected
+
+
+def test_get_socketname_no_user():
+    assert ipc._get_socketname(None).startswith('qutebrowser-')
+
+
+class TestListen:
+
+    def test_listen_error_exc(self, qlocalserver):
+        """Tet the ListenError exception."""
+        qlocalserver.listen(None)
+        exc = ipc.ListenError(qlocalserver)
+        assert exc.code == 2
+        assert exc.message == "QLocalServer::listen: Name error"
+        msg = ("Error while listening to IPC server: QLocalServer::listen: Name "
+               "error (error 2)")
+        assert str(exc) == msg
+
+    def test_remove_error(self, ipc_server, monkeypatch):
+        """Simulate an error in _remove_server."""
+        monkeypatch.setattr(ipc_server, '_socketname', None)
+        with pytest.raises(ipc.Error) as excinfo:
+            ipc_server.listen()
+        assert str(excinfo.value) == "Error while removing server None!"
+
+    def test_error(self, ipc_server, monkeypatch):
+        """Simulate an error while listening."""
+        monkeypatch.setattr('qutebrowser.misc.ipc.QLocalServer.removeServer',
+                            lambda self: True)
+        monkeypatch.setattr(ipc_server, '_socketname', None)
+        with pytest.raises(ipc.ListenError):
+            ipc_server.listen()
+
+    def test_in_use(self, qlocalserver, ipc_server, monkeypatch):
+        monkeypatch.setattr('qutebrowser.misc.ipc.QLocalServer.removeServer',
+                            lambda self: True)
+        qlocalserver.listen('qutebrowser-test')
+        with pytest.raises(ipc.AddressInUseError):
+            ipc_server.listen()
+
+    def test_successful(self, ipc_server):
+        ipc_server.listen()
+
+
+class TestOnError:
+
+    def test_closed(self, ipc_server):
+        ipc_server._socket = QLocalSocket()
+        ipc_server._timer.timeout.disconnect()
+        ipc_server._timer.start()
+        ipc_server.on_error(QLocalSocket.PeerClosedError)
+        assert not ipc_server._timer.isActive()
+
+    def test_other_error(self, ipc_server, monkeypatch):
+        socket = QLocalSocket()
+        ipc_server._socket = socket
+        monkeypatch.setattr(socket, 'error',
+                            lambda: QLocalSocket.ConnectionRefusedError)
+        monkeypatch.setattr(socket, 'errorString',
+                            lambda: "Connection refused")
+        socket.setErrorString("Connection refused.")
+
+        with pytest.raises(ipc.Error) as excinfo:
+            ipc_server.on_error(QLocalSocket.ConnectionRefusedError)
+
+        expected = ("Error while handling IPC connection: Connection refused "
+                    "(error 0)")
+        assert str(excinfo.value) == expected
+
+
+class TestHandleConnection:
+
+    def test_ignored(self, ipc_server, monkeypatch):
+        m = mock.Mock(spec=[])
+        monkeypatch.setattr(ipc_server._server, 'nextPendingConnection', m)
+        ipc_server.ignored = True
+        ipc_server.handle_connection()
+        assert not m.called
+
+    def test_no_connection(self, ipc_server, caplog):
+        ipc_server.handle_connection()
+        assert len(caplog.records()) == 1
+        record = caplog.records()[0]
+        assert record.message == "No new connection to handle."
+
+
+class TestRealConnections:
+
+    @pytest.yield_fixture
+    def connected_socket(self, qtbot, qlocalsocket, ipc_server):
+        ipc_server.listen()
+        with qtbot.waitSignal(ipc_server._server.newConnection, raising=True):
+            qlocalsocket.connectToServer('qutebrowser-test')
+        yield qlocalsocket
+        qlocalsocket.disconnectFromServer()
+
+    def test_normal(self, qtbot, tmpdir, ipc_server):
+        ipc_server.listen()
+        spy = QSignalSpy(ipc_server.got_args)
+        error_spy = QSignalSpy(ipc_server.got_invalid_data)
+
+        with qtbot.waitSignal(ipc_server.got_args, raising=True):
+            with tmpdir.as_cwd():
+                sent = ipc.send_to_running_instance('qutebrowser-test',
+                                                    ['foo'])
+
+        assert sent
+        assert len(spy) == 1
+        assert not error_spy
+        assert spy[0] == [['foo'], str(tmpdir)]
+
+    def test_double_connection(self, qtbot, connected_socket, ipc_server,
+                               caplog):
+        spy = QSignalSpy(ipc_server.got_args)
+        error_spy = QSignalSpy(ipc_server.got_invalid_data)
+        with qtbot.waitSignal(ipc_server._server.newConnection, raising=True):
+            sent = ipc.send_to_running_instance('qutebrowser-test', [])
+        assert sent
+        assert not spy
+        assert not error_spy
+        message = ("Got new connection but ignoring it because we're still "
+                   "handling another one.")
+        assert message in [rec.message for rec in caplog.records()]
+
+    def test_disconnected_immediately(self, qtbot, connected_socket,
+                                      ipc_server, caplog):
+        """Disconnect without sending data.
+
+        This means self._socket will be None on on_disconnected.
+        """
+        connected_socket.disconnectFromServer()
+
+    def test_partial_line(self, connected_socket):
+        connected_socket.write(b'foo')
+
+    @pytest.mark.parametrize('data', [
+        b'\x80\n', # invalid UTF8
+        b'\n',
+        b'{"is this invalid json?": true\n',
+        b'{"valid json without args": true}\n',
+    ])
+    def test_invalid_data(self, qtbot, ipc_server, connected_socket, caplog, data):
+        with caplog.atLevel(logging.ERROR):
+            with qtbot.waitSignal(ipc_server.got_invalid_data, raising=True):
+                connected_socket.write(data)
+        messages = [r.message for r in caplog.records()]
+        assert messages[-1] == 'Ignoring invalid IPC data.'
+
+    def test_multiline(self, qtbot, ipc_server, connected_socket):
+        spy = QSignalSpy(ipc_server.got_args)
+        error_spy = QSignalSpy(ipc_server.got_invalid_data)
+
+        with qtbot.waitSignals([ipc_server.got_args, ipc_server.got_args],
+                               raising=True):
+            connected_socket.write(b'{"args": ["one"]}\n{"args": ["two"]}\n')
+
+        assert len(spy) == 2
+        assert spy[0][0] == ['one']
+        assert spy[1][0] == ['two']
+
+    def test_connect_no_server(self, caplog):
+        sent = ipc.send_to_running_instance('qutebrowser-test', [])
+        assert not sent
+        msg = caplog.records()[-1].message
+        assert msg == "No existing instance present (error 2)"
+
+    def test_timeout(self, qtbot, caplog, qlocalsocket, ipc_server):
+        ipc_server._timer.setInterval(100)
+        ipc_server.listen()
+
+        with qtbot.waitSignal(ipc_server._server.newConnection, raising=True):
+            qlocalsocket.connectToServer('qutebrowser-test')
+
+        with caplog.atLevel(logging.ERROR):
+            with qtbot.waitSignal(qlocalsocket.disconnected, raising=True):
+                pass
+
+        assert caplog.records()[-1].message == "IPC connection timed out."