ipc: Use a custom class for exceptions.

This commit is contained in:
Florian Bruhin 2015-09-04 07:12:23 +02:00
parent 2a4cd02704
commit ea0cbea1dd
2 changed files with 51 additions and 20 deletions

View File

@ -50,7 +50,34 @@ def _get_socketname(basedir, user=None):
class Error(Exception):
"""Exception raised when there was a problem with IPC."""
"""Base class for IPC exceptions."""
class SocketError(Error):
"""Exception raised when there was an error with a QLocalSocket.
Args:
code: The error code.
message: The error message.
action: The action which was taken when the error happened.
"""
def __init__(self, action, socket):
"""Constructor.
Args:
action: The action which was taken when the error happened.
socket: The QLocalSocket which has the error set.
"""
super().__init__()
self.action = action
self.code = socket.error()
self.message = socket.errorString()
def __str__(self):
return "Error while {}: {} (error {})".format(
self.action, self.message, self.code)
class ListenError(Error):
@ -139,7 +166,7 @@ class IPCServer(QObject):
@pyqtSlot(int)
def on_error(self, err):
"""Convenience method which calls _socket_error on an error."""
"""Raise SocketError on fatal errors."""
if self._socket is None:
# Sometimes this gets called from stale sockets.
msg = "In on_error with None socket!"
@ -153,7 +180,7 @@ class IPCServer(QObject):
log.ipc.debug("Socket error {}: {}".format(
self._socket.error(), self._socket.errorString()))
if err != QLocalSocket.PeerClosedError:
_socket_error("handling IPC connection", self._socket)
raise SocketError("handling IPC connection", self._socket)
@pyqtSlot()
def handle_connection(self):
@ -259,17 +286,6 @@ class IPCServer(QObject):
self._remove_server()
def _socket_error(action, socket):
"""Raise an Error based on an action and a QLocalSocket.
Args:
action: A string like "writing to running instance".
socket: A QLocalSocket.
"""
raise Error("Error while {}: {} (error {})".format(
action, socket.errorString(), socket.error()))
def send_to_running_instance(socketname, command, *, socket=None):
"""Try to send a commandline to a running instance.
@ -303,7 +319,7 @@ def send_to_running_instance(socketname, command, *, socket=None):
socket.writeData(data)
socket.waitForBytesWritten(WRITE_TIMEOUT)
if socket.error() != QLocalSocket.UnknownSocketError:
_socket_error("writing to running instance", socket)
raise SocketError("writing to running instance", socket)
else:
socket.disconnectFromServer()
if socket.state() != QLocalSocket.UnconnectedState:
@ -312,7 +328,7 @@ def send_to_running_instance(socketname, command, *, socket=None):
else:
if socket.error() not in (QLocalSocket.ConnectionRefusedError,
QLocalSocket.ServerNotFoundError):
_socket_error("connecting to running instance", socket)
raise SocketError("connecting to running instance", socket)
else:
log.ipc.debug("No existing instance present (error {})".format(
socket.error()))

View File

@ -169,10 +169,9 @@ def test_get_socketname_no_user():
assert ipc._get_socketname(None).startswith('qutebrowser-')
class TestListen:
class TestExceptions:
def test_listen_error_exc(self, qlocalserver):
"""Tet the ListenError exception."""
def test_listen_error(self, qlocalserver):
qlocalserver.listen(None)
exc = ipc.ListenError(qlocalserver)
assert exc.code == 2
@ -181,6 +180,22 @@ class TestListen:
"Name error (error 2)")
assert str(exc) == msg
with pytest.raises(ipc.Error):
raise exc
def test_socket_error(self, qlocalserver):
socket = FakeSocket(error=QLocalSocket.ConnectionRefusedError)
exc = ipc.SocketError("testing", socket)
assert exc.code == QLocalSocket.ConnectionRefusedError
assert exc.message == "Error string"
assert str(exc) == "Error while testing: Error string (error 0)"
with pytest.raises(ipc.Error):
raise exc
class TestListen:
@pytest.mark.posix
def test_remove_error(self, ipc_server, monkeypatch):
"""Simulate an error in _remove_server."""
@ -503,7 +518,7 @@ class TestSendOrListen:
assert "Got AddressInUseError, trying again." in msgs
@pytest.mark.parametrize('has_error, excname', [
(True, 'Error'),
(True, 'SocketError'),
(False, 'AddressInUseError')
])
def test_address_in_use_error(self, qlocalserver_mock, qlocalsocket_mock,