diff --git a/tests/unit/misc/test_ipc.py b/tests/unit/misc/test_ipc.py new file mode 100644 index 000000000..4b3f40c8f --- /dev/null +++ b/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) +# +# 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 . + +"""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."