diff options
Diffstat (limited to 'tests/unit/misc/test_ipc.py')
| -rw-r--r-- | tests/unit/misc/test_ipc.py | 85 |
1 files changed, 42 insertions, 43 deletions
diff --git a/tests/unit/misc/test_ipc.py b/tests/unit/misc/test_ipc.py index 9b6aa286c..a58920544 100644 --- a/tests/unit/misc/test_ipc.py +++ b/tests/unit/misc/test_ipc.py @@ -30,14 +30,13 @@ from unittest import mock from typing import Optional, List import pytest -from qutebrowser.qt.core import pyqtSignal, QObject -from qutebrowser.qt.network import QLocalServer, QLocalSocket, QAbstractSocket -from qutebrowser.qt.test import QSignalSpy +from qutebrowser.qt import test import qutebrowser from qutebrowser.misc import ipc from qutebrowser.utils import standarddir, utils, version from helpers import stubs, testutils +from qutebrowser.qt import network, core pytestmark = pytest.mark.usefixtures('qapp') @@ -56,7 +55,7 @@ def ipc_server(qapp, qtbot): server = ipc.IPCServer('qute-test') yield server if (server._socket is not None and - server._socket.state() != QLocalSocket.LocalSocketState.UnconnectedState): + server._socket.state() != network.QLocalSocket.LocalSocketState.UnconnectedState): with qtbot.wait_signal(server._socket.disconnected, raising=False): server._socket.abort() try: @@ -67,7 +66,7 @@ def ipc_server(qapp, qtbot): @pytest.fixture def qlocalserver(qapp): - server = QLocalServer() + server = network.QLocalServer() yield server server.close() server.deleteLater() @@ -75,10 +74,10 @@ def qlocalserver(qapp): @pytest.fixture def qlocalsocket(qapp): - socket = QLocalSocket() + socket = network.QLocalSocket() yield socket socket.disconnectFromServer() - if socket.state() != QLocalSocket.LocalSocketState.UnconnectedState: + if socket.state() != network.QLocalSocket.LocalSocketState.UnconnectedState: socket.waitForDisconnected(1000) @@ -89,7 +88,7 @@ def fake_runtime_dir(monkeypatch, short_tmpdir): return short_tmpdir -class FakeSocket(QObject): +class FakeSocket(core.QObject): """A stub for a QLocalSocket. @@ -100,11 +99,11 @@ class FakeSocket(QObject): _connect_successful: The value returned for waitForConnected(). """ - readyRead = pyqtSignal() # noqa: N815 - disconnected = pyqtSignal() - errorOccurred = pyqtSignal(QLocalSocket.LocalSocketError) # noqa: N815 + readyRead = core.pyqtSignal() # noqa: N815 + disconnected = core.pyqtSignal() + errorOccurred = core.pyqtSignal(network.QLocalSocket.LocalSocketError) # noqa: N815 - def __init__(self, *, error=QLocalSocket.LocalSocketError.UnknownSocketError, state=None, + def __init__(self, *, error=network.QLocalSocket.LocalSocketError.UnknownSocketError, state=None, data=None, connect_successful=True, parent=None): super().__init__(parent) self._error_val = error @@ -270,7 +269,7 @@ class TestExceptions: def test_listen_error(self, qlocalserver): qlocalserver.listen(None) exc = ipc.ListenError(qlocalserver) - assert exc.code == QAbstractSocket.SocketError.HostNotFoundError + assert exc.code == network.QAbstractSocket.SocketError.HostNotFoundError assert exc.message == "QLocalServer::listen: Name error" msg = ("Error while listening to IPC server: QLocalServer::listen: " "Name error (HostNotFoundError)") @@ -280,9 +279,9 @@ class TestExceptions: raise exc def test_socket_error(self, qlocalserver): - socket = FakeSocket(error=QLocalSocket.LocalSocketError.ConnectionRefusedError) + socket = FakeSocket(error=network.QLocalSocket.LocalSocketError.ConnectionRefusedError) exc = ipc.SocketError("testing", socket) - assert exc.code == QLocalSocket.LocalSocketError.ConnectionRefusedError + assert exc.code == network.QLocalSocket.LocalSocketError.ConnectionRefusedError assert exc.message == "Error string" assert str(exc) == "Error while testing: Error string (ConnectionRefusedError)" @@ -322,7 +321,7 @@ class TestListen: @pytest.mark.windows def test_permissions_windows(self, ipc_server): opts = ipc_server._server.socketOptions() - assert opts == QLocalServer.SocketOption.UserAccessOption + assert opts == network.QLocalServer.SocketOption.UserAccessOption @pytest.mark.posix def test_permissions_posix(self, ipc_server): @@ -403,24 +402,24 @@ class TestListen: class TestOnError: def test_closed(self, ipc_server): - ipc_server._socket = QLocalSocket() + ipc_server._socket = network.QLocalSocket() ipc_server._timer.timeout.disconnect() ipc_server._timer.start() - ipc_server.on_error(QLocalSocket.LocalSocketError.PeerClosedError) + ipc_server.on_error(network.QLocalSocket.LocalSocketError.PeerClosedError) assert not ipc_server._timer.isActive() def test_other_error(self, ipc_server, monkeypatch): - socket = QLocalSocket() + socket = network.QLocalSocket() ipc_server._socket = socket monkeypatch.setattr(socket, 'error', - lambda: QLocalSocket.LocalSocketError.ConnectionRefusedError) + lambda: network.QLocalSocket.LocalSocketError.ConnectionRefusedError) monkeypatch.setattr(socket, 'errorString', lambda: "Connection refused") socket.setErrorString("Connection refused.") with pytest.raises(ipc.Error, match=r"Error while handling IPC " r"connection: Connection refused \(ConnectionRefusedError\)"): - ipc_server.on_error(QLocalSocket.LocalSocketError.ConnectionRefusedError) + ipc_server.on_error(network.QLocalSocket.LocalSocketError.ConnectionRefusedError) class TestHandleConnection: @@ -444,13 +443,13 @@ class TestHandleConnection: assert any(message.startswith(msg) for message in caplog.messages) def test_disconnected_immediately(self, ipc_server, caplog): - socket = FakeSocket(state=QLocalSocket.LocalSocketState.UnconnectedState) + socket = FakeSocket(state=network.QLocalSocket.LocalSocketState.UnconnectedState) ipc_server._server = FakeServer(socket) ipc_server.handle_connection() assert "Socket was disconnected immediately." in caplog.messages def test_error_immediately(self, ipc_server, caplog): - socket = FakeSocket(error=QLocalSocket.LocalSocketError.ConnectionError) + socket = FakeSocket(error=network.QLocalSocket.LocalSocketError.ConnectionError) ipc_server._server = FakeServer(socket) with pytest.raises(ipc.Error, match=r"Error while handling IPC " @@ -529,7 +528,7 @@ def test_invalid_data(qtbot, ipc_server, connected_socket, caplog, data, msg): def test_multiline(qtbot, ipc_server, connected_socket): - spy = QSignalSpy(ipc_server.got_args) + spy = test.QSignalSpy(ipc_server.got_args) data = ('{{"args": ["one"], "target_arg": "tab",' ' "protocol_version": {version}}}\n' @@ -589,7 +588,7 @@ class TestSendToRunningInstance: assert parsed == raw_expected def test_socket_error(self): - socket = FakeSocket(error=QLocalSocket.LocalSocketError.ConnectionError) + socket = FakeSocket(error=network.QLocalSocket.LocalSocketError.ConnectionError) with pytest.raises(ipc.Error, match=r"Error while writing to running " r"instance: Error string \(ConnectionError\)"): ipc.send_to_running_instance('qute-test', [], None, socket=socket) @@ -599,7 +598,7 @@ class TestSendToRunningInstance: ipc.send_to_running_instance('qute-test', [], None, socket=socket) def test_socket_error_no_server(self): - socket = FakeSocket(error=QLocalSocket.LocalSocketError.ConnectionError, + socket = FakeSocket(error=network.QLocalSocket.LocalSocketError.ConnectionError, connect_successful=False) with pytest.raises(ipc.Error, match=r"Error while connecting to " r"running instance: Error string \(ConnectionError\)"): @@ -658,7 +657,7 @@ class TestSendOrListen: def qlocalserver_mock(self, mocker): m = mocker.patch('qutebrowser.misc.ipc.QLocalServer', autospec=True) m().errorString.return_value = "Error string" - m.SocketOption = QLocalServer.SocketOption + m.SocketOption = network.QLocalServer.SocketOption m().newConnection = stubs.FakeSignal() return m @@ -666,8 +665,8 @@ class TestSendOrListen: def qlocalsocket_mock(self, mocker): m = mocker.patch('qutebrowser.misc.ipc.QLocalSocket', autospec=True) m().errorString.return_value = "Error string" - m.LocalSocketError = QLocalSocket.LocalSocketError - m.LocalSocketState = QLocalSocket.LocalSocketState + m.LocalSocketError = network.QLocalSocket.LocalSocketError + m.LocalSocketState = network.QLocalSocket.LocalSocketState return m @pytest.mark.linux(reason="Flaky on Windows and macOS") @@ -701,14 +700,14 @@ class TestSendOrListen: -> success """ qlocalserver_mock().listen.return_value = False - err = QAbstractSocket.SocketError.AddressInUseError + err = network.QAbstractSocket.SocketError.AddressInUseError qlocalserver_mock().serverError.return_value = err qlocalsocket_mock().waitForConnected.side_effect = [False, True] qlocalsocket_mock().error.side_effect = [ - QLocalSocket.LocalSocketError.ServerNotFoundError, - QLocalSocket.LocalSocketError.UnknownSocketError, - QLocalSocket.LocalSocketError.UnknownSocketError, # error() gets called twice + network.QLocalSocket.LocalSocketError.ServerNotFoundError, + network.QLocalSocket.LocalSocketError.UnknownSocketError, + network.QLocalSocket.LocalSocketError.UnknownSocketError, # error() gets called twice ] ret = ipc.send_or_listen(args) @@ -734,17 +733,17 @@ class TestSendOrListen: -> not sent / error """ qlocalserver_mock().listen.return_value = False - err = QAbstractSocket.SocketError.AddressInUseError + err = network.QAbstractSocket.SocketError.AddressInUseError qlocalserver_mock().serverError.return_value = err # If the second connection succeeds, we will have an error later. # If it fails, that's the "not sent" case above. qlocalsocket_mock().waitForConnected.side_effect = [False, has_error] qlocalsocket_mock().error.side_effect = [ - QLocalSocket.LocalSocketError.ServerNotFoundError, - QLocalSocket.LocalSocketError.ServerNotFoundError, - QLocalSocket.LocalSocketError.ConnectionRefusedError, - QLocalSocket.LocalSocketError.ConnectionRefusedError, # error() gets called twice + network.QLocalSocket.LocalSocketError.ServerNotFoundError, + network.QLocalSocket.LocalSocketError.ServerNotFoundError, + network.QLocalSocket.LocalSocketError.ConnectionRefusedError, + network.QLocalSocket.LocalSocketError.ConnectionRefusedError, # error() gets called twice ] # For debug.qenum_key() on Qt 5 value_to_key = qlocalsocket_mock.staticMetaObject.enumerator().valueToKey @@ -769,7 +768,7 @@ class TestSendOrListen: def test_error_while_listening(self, qlocalserver_mock, caplog, args): """Test an error with the first listen call.""" qlocalserver_mock().listen.return_value = False - err = QAbstractSocket.SocketError.SocketResourceError + err = network.QAbstractSocket.SocketError.SocketResourceError qlocalserver_mock().serverError.return_value = err with caplog.at_level(logging.ERROR): @@ -812,7 +811,7 @@ def test_connect_inexistent(qlocalsocket): would not work properly. """ qlocalsocket.connectToServer('qute-test-inexistent') - assert qlocalsocket.error() == QLocalSocket.LocalSocketError.ServerNotFoundError + assert qlocalsocket.error() == network.QLocalSocket.LocalSocketError.ServerNotFoundError @pytest.mark.posix @@ -824,12 +823,12 @@ def test_socket_options_address_in_use_problem(qlocalserver, short_tmpdir): """ servername = str(short_tmpdir / 'x') - s1 = QLocalServer() + s1 = network.QLocalServer() ok = s1.listen(servername) assert ok - s2 = QLocalServer() - s2.setSocketOptions(QLocalServer.SocketOption.UserAccessOption) + s2 = network.QLocalServer() + s2.setSocketOptions(network.QLocalServer.SocketOption.UserAccessOption) ok = s2.listen(servername) print(s2.errorString()) # We actually would expect ok == False here - but we want the test to fail |