Test raw json data for IPC.

This commit is contained in:
Florian Bruhin 2015-09-06 18:06:23 +02:00
parent e9608a6aea
commit b95fd2c814
2 changed files with 17 additions and 1 deletions

View File

@ -124,10 +124,12 @@ class IPCServer(QObject):
Signals:
got_args: Emitted when there was an IPC connection and arguments were
passed.
got_args: Emitted with the raw data an IPC connection got.
got_invalid_data: Emitted when there was invalid incoming data.
"""
got_args = pyqtSignal(list, str)
got_raw = pyqtSignal(bytes)
got_invalid_data = pyqtSignal()
def __init__(self, socketname, parent=None):
@ -237,6 +239,7 @@ class IPCServer(QObject):
self._timer.start()
while self._socket is not None and self._socket.canReadLine():
data = bytes(self._socket.readLine())
self.got_raw.emit(data)
log.ipc.debug("Read from socket: {}".format(data))
try:
decoded = data.decode('utf-8')

View File

@ -22,6 +22,7 @@
import getpass
import collections
import logging
import json
from unittest import mock
import pytest
@ -29,6 +30,7 @@ from PyQt5.QtCore import pyqtSignal, QObject
from PyQt5.QtNetwork import QLocalServer, QLocalSocket, QAbstractSocket
from PyQt5.QtTest import QSignalSpy
import qutebrowser
from qutebrowser.misc import ipc
from qutebrowser.utils import objreg
from helpers import stubs # pylint: disable=import-error
@ -373,6 +375,7 @@ class TestSendToRunningInstance:
def test_normal(self, qtbot, tmpdir, ipc_server, mocker, has_cwd):
ipc_server.listen()
spy = QSignalSpy(ipc_server.got_args)
raw_spy = QSignalSpy(ipc_server.got_raw)
error_spy = QSignalSpy(ipc_server.got_invalid_data)
with qtbot.waitSignal(ipc_server.got_args, raising=True, timeout=5000):
@ -384,11 +387,21 @@ class TestSendToRunningInstance:
['foo'])
assert sent
assert len(spy) == 1
assert not error_spy
expected_cwd = str(tmpdir) if has_cwd else ''
assert len(spy) == 1
assert spy[0] == [['foo'], expected_cwd]
assert len(raw_spy) == 1
assert len(raw_spy[0]) == 1
raw_expected = {'args': ['foo'], 'version': qutebrowser.__version__}
if has_cwd:
raw_expected['cwd'] = str(tmpdir)
parsed = json.loads(raw_spy[0][0].decode('utf-8'))
assert parsed == raw_expected
def test_socket_error(self):
socket = FakeSocket(error=QLocalSocket.ConnectionError)
with pytest.raises(ipc.Error) as excinfo: