Add/improve tests for qutebrowser.utils.qtutils.
This commit is contained in:
Florian Bruhin
parent
e60f698615
commit
1e982a9a84
@ -19,15 +19,51 @@
|
||||
|
||||
"""Tests for qutebrowser.utils.qtutils."""
|
||||
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
import operator
|
||||
import os.path
|
||||
try:
|
||||
from test import test_file
|
||||
except ImportError:
|
||||
# Debian patches Python to remove the tests...
|
||||
test_file = None
|
||||
|
||||
import pytest
|
||||
import unittest
|
||||
import unittest.mock
|
||||
from PyQt5.QtCore import (QDataStream, QPoint, QUrl, QByteArray, QIODevice,
|
||||
QTimer, QBuffer, QFile, QProcess)
|
||||
from PyQt5.QtWidgets import QApplication
|
||||
|
||||
from qutebrowser import qutebrowser
|
||||
from qutebrowser.utils import qtutils
|
||||
import overflow_test_cases
|
||||
|
||||
|
||||
@pytest.mark.parametrize('qversion, version, op, expected', [
|
||||
('5.4.0', '5.4.0', operator.ge, True),
|
||||
('5.4.0', '5.4.0', operator.eq, True),
|
||||
('5.4.0', '5.4', operator.eq, True),
|
||||
('5.4.1', '5.4', operator.ge, True),
|
||||
('5.3.2', '5.4', operator.ge, False),
|
||||
('5.3.0', '5.3.2', operator.ge, False),
|
||||
])
|
||||
def test_version_check(monkeypatch, qversion, version, op, expected):
|
||||
"""Test for version_check().
|
||||
|
||||
Args:
|
||||
monkeypatch: The pytest monkeypatch fixture.
|
||||
qversion: The version to set as fake qVersion().
|
||||
version: The version to compare with.
|
||||
op: The operator to use when comparing.
|
||||
expected: The expected result.
|
||||
"""
|
||||
monkeypatch.setattr('qutebrowser.utils.qtutils.qVersion', lambda: qversion)
|
||||
assert qtutils.version_check(version, op) == expected
|
||||
|
||||
|
||||
class TestCheckOverflow:
|
||||
|
||||
"""Test check_overflow."""
|
||||
@ -68,20 +104,18 @@ class TestGetQtArgs:
|
||||
mocker.patch.object(parser, 'exit', side_effect=Exception)
|
||||
return parser
|
||||
|
||||
def test_no_qt_args(self, parser):
|
||||
@pytest.mark.parametrize('args, expected', [
|
||||
# No Qt arguments
|
||||
(['--debug'], [sys.argv[0]]),
|
||||
# Qt flag
|
||||
(['--debug', '--qt-reverse', '--nocolor'], [sys.argv[0], '-reverse']),
|
||||
# Qt argument with value
|
||||
(['--qt-stylesheet', 'foo'], [sys.argv[0], '-stylesheet', 'foo']),
|
||||
])
|
||||
def test_qt_args(self, args, expected, parser):
|
||||
"""Test commandline with no Qt arguments given."""
|
||||
args = parser.parse_args(['--debug'])
|
||||
assert qtutils.get_args(args) == [sys.argv[0]]
|
||||
|
||||
def test_qt_flag(self, parser):
|
||||
"""Test commandline with a Qt flag."""
|
||||
args = parser.parse_args(['--debug', '--qt-reverse', '--nocolor'])
|
||||
assert qtutils.get_args(args) == [sys.argv[0], '-reverse']
|
||||
|
||||
def test_qt_arg(self, parser):
|
||||
"""Test commandline with a Qt argument."""
|
||||
args = parser.parse_args(['--qt-stylesheet', 'foobar'])
|
||||
assert qtutils.get_args(args) == [sys.argv[0], '-stylesheet', 'foobar']
|
||||
parsed = parser.parse_args(args)
|
||||
assert qtutils.get_args(parsed) == expected
|
||||
|
||||
def test_qt_both(self, parser):
|
||||
"""Test commandline with a Qt argument and flag."""
|
||||
@ -91,3 +125,832 @@ class TestGetQtArgs:
|
||||
assert '-reverse' in qt_args
|
||||
assert '-stylesheet' in qt_args
|
||||
assert 'foobar' in qt_args
|
||||
|
||||
|
||||
@pytest.mark.parametrize('os_name, qversion, expected', [
|
||||
('linux', '5.2.1', True), # unaffected OS
|
||||
('linux', '5.4.1', True), # unaffected OS
|
||||
('nt', '5.2.1', False),
|
||||
('nt', '5.3.0', True), # unaffected Qt version
|
||||
('nt', '5.4.1', True), # unaffected Qt version
|
||||
])
|
||||
def test_check_print_compat(os_name, qversion, expected, monkeypatch):
|
||||
"""Test check_print_compat.
|
||||
|
||||
Args:
|
||||
os_name: The fake os.name to set.
|
||||
qversion: The fake qVersion() to set.
|
||||
expected: The expected return value.
|
||||
"""
|
||||
monkeypatch.setattr('qutebrowser.utils.qtutils.os.name', os_name)
|
||||
monkeypatch.setattr('qutebrowser.utils.qtutils.qVersion', lambda: qversion)
|
||||
assert qtutils.check_print_compat() == expected
|
||||
|
||||
|
||||
class QtObject:
|
||||
|
||||
"""Fake Qt object for test_ensure."""
|
||||
|
||||
def __init__(self, valid=True, null=False, error=None):
|
||||
self._valid = valid
|
||||
self._null = null
|
||||
self._error = error
|
||||
|
||||
def __repr__(self):
|
||||
return '<QtObject>'
|
||||
|
||||
def errorString(self):
|
||||
"""Get the fake error, or raise AttributeError if set to None."""
|
||||
if self._error is None:
|
||||
raise AttributeError
|
||||
else:
|
||||
return self._error
|
||||
|
||||
def isValid(self):
|
||||
return self._valid
|
||||
|
||||
def isNull(self):
|
||||
return self._null
|
||||
|
||||
|
||||
@pytest.mark.parametrize('func_name, obj, raising, exc_reason, exc_str', [
|
||||
# ensure_valid, good examples
|
||||
('ensure_valid', QtObject(valid=True, null=True), False, None, None),
|
||||
('ensure_valid', QtObject(valid=True, null=False), False, None, None),
|
||||
# ensure_valid, bad examples
|
||||
('ensure_valid', QtObject(valid=False, null=True), True, None,
|
||||
'<QtObject> is not valid'),
|
||||
('ensure_valid', QtObject(valid=False, null=False), True, None,
|
||||
'<QtObject> is not valid'),
|
||||
('ensure_valid', QtObject(valid=False, null=True, error='Test'), True,
|
||||
'Test', '<QtObject> is not valid: Test'),
|
||||
# ensure_not_null, good examples
|
||||
('ensure_not_null', QtObject(valid=True, null=False), False, None, None),
|
||||
('ensure_not_null', QtObject(valid=False, null=False), False, None, None),
|
||||
# ensure_not_null, bad examples
|
||||
('ensure_not_null', QtObject(valid=True, null=True), True, None,
|
||||
'<QtObject> is null'),
|
||||
('ensure_not_null', QtObject(valid=False, null=True), True, None,
|
||||
'<QtObject> is null'),
|
||||
('ensure_not_null', QtObject(valid=False, null=True, error='Test'), True,
|
||||
'Test', '<QtObject> is null: Test'),
|
||||
])
|
||||
def test_ensure(func_name, obj, raising, exc_reason, exc_str):
|
||||
"""Test ensure_valid and ensure_not_null.
|
||||
|
||||
The function is parametrized as they do nearly the same.
|
||||
|
||||
Args:
|
||||
func_name: The name of the function to call.
|
||||
obj: The object to test with.
|
||||
raising: Whether QtValueError is expected to be raised.
|
||||
exc_reason: The expected .reason attribute of the exception.
|
||||
exc_str: The expected string of the exception.
|
||||
"""
|
||||
func = getattr(qtutils, func_name)
|
||||
if raising:
|
||||
with pytest.raises(qtutils.QtValueError) as excinfo:
|
||||
func(obj)
|
||||
assert excinfo.value.reason == exc_reason
|
||||
assert str(excinfo.value) == exc_str
|
||||
else:
|
||||
func(obj)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('status, raising, message', [
|
||||
(QDataStream.Ok, False, None),
|
||||
(QDataStream.ReadPastEnd, True, "The data stream has read past the end of "
|
||||
"the data in the underlying device."),
|
||||
(QDataStream.ReadCorruptData, True, "The data stream has read corrupt "
|
||||
"data."),
|
||||
(QDataStream.WriteFailed, True, "The data stream cannot write to the "
|
||||
"underlying device."),
|
||||
])
|
||||
def test_check_qdatastream(status, raising, message):
|
||||
"""Test check_qdatastream.
|
||||
|
||||
Args:
|
||||
status: The status to set on the QDataStream we test with.
|
||||
raising: Whether check_qdatastream is expected to raise OSError.
|
||||
message: The expected exception string.
|
||||
"""
|
||||
stream = QDataStream()
|
||||
stream.setStatus(status)
|
||||
if raising:
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
qtutils.check_qdatastream(stream)
|
||||
assert str(excinfo.value) == message
|
||||
else:
|
||||
qtutils.check_qdatastream(stream)
|
||||
|
||||
|
||||
def test_qdatastream_status_count():
|
||||
"""Make sure no new members are added to QDataStream.Status."""
|
||||
values = vars(QDataStream).values()
|
||||
status_vals = [e for e in values if isinstance(e, QDataStream.Status)]
|
||||
assert len(status_vals) == 4
|
||||
|
||||
|
||||
@pytest.mark.parametrize('obj', [
|
||||
QPoint(23, 42),
|
||||
QUrl('http://www.qutebrowser.org/'),
|
||||
])
|
||||
def test_serialize(obj):
|
||||
"""Test a serialize/deserialize round trip.
|
||||
|
||||
Args:
|
||||
obj: The object to test with.
|
||||
"""
|
||||
new_obj = type(obj)()
|
||||
qtutils.deserialize(qtutils.serialize(obj), new_obj)
|
||||
assert new_obj == obj
|
||||
|
||||
|
||||
class TestSerializeStream:
|
||||
|
||||
"""Tests for serialize_stream and deserialize_stream."""
|
||||
|
||||
def _set_status(self, stream, status):
|
||||
"""Helper function so mocks can set an error status when used."""
|
||||
stream.status.return_value = status
|
||||
|
||||
@pytest.fixture
|
||||
def stream_mock(self):
|
||||
"""Fixture providing a QDataStream-like mock."""
|
||||
m = unittest.mock.MagicMock(spec=QDataStream)
|
||||
m.status.return_value = QDataStream.Ok
|
||||
return m
|
||||
|
||||
def test_serialize_pre_error_mock(self, stream_mock):
|
||||
"""Test serialize_stream with an error already set."""
|
||||
stream_mock.status.return_value = QDataStream.ReadCorruptData
|
||||
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
qtutils.serialize_stream(stream_mock, QPoint())
|
||||
|
||||
assert not stream_mock.__lshift__.called
|
||||
assert str(excinfo.value) == "The data stream has read corrupt data."
|
||||
|
||||
def test_serialize_post_error_mock(self, stream_mock):
|
||||
"""Test serialize_stream with an error while serializing."""
|
||||
obj = QPoint()
|
||||
stream_mock.__lshift__.side_effect = lambda _other: self._set_status(
|
||||
stream_mock, QDataStream.ReadCorruptData)
|
||||
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
qtutils.serialize_stream(stream_mock, obj)
|
||||
|
||||
assert stream_mock.__lshift__.called_once_with(obj)
|
||||
assert str(excinfo.value) == "The data stream has read corrupt data."
|
||||
|
||||
def test_deserialize_pre_error_mock(self, stream_mock):
|
||||
"""Test deserialize_stream with an error already set."""
|
||||
stream_mock.status.return_value = QDataStream.ReadCorruptData
|
||||
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
qtutils.deserialize_stream(stream_mock, QPoint())
|
||||
|
||||
assert not stream_mock.__rshift__.called
|
||||
assert str(excinfo.value) == "The data stream has read corrupt data."
|
||||
|
||||
def test_deserialize_post_error_mock(self, stream_mock):
|
||||
"""Test deserialize_stream with an error while deserializing."""
|
||||
obj = QPoint()
|
||||
stream_mock.__rshift__.side_effect = lambda _other: self._set_status(
|
||||
stream_mock, QDataStream.ReadCorruptData)
|
||||
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
qtutils.deserialize_stream(stream_mock, obj)
|
||||
|
||||
assert stream_mock.__rshift__.called_once_with(obj)
|
||||
assert str(excinfo.value) == "The data stream has read corrupt data."
|
||||
|
||||
def test_round_trip_real_stream(self):
|
||||
"""Test a round trip with a real QDataStream."""
|
||||
src_obj = QPoint(23, 42)
|
||||
dest_obj = QPoint()
|
||||
data = QByteArray()
|
||||
|
||||
write_stream = QDataStream(data, QIODevice.WriteOnly)
|
||||
qtutils.serialize_stream(write_stream, src_obj)
|
||||
|
||||
read_stream = QDataStream(data, QIODevice.ReadOnly)
|
||||
qtutils.deserialize_stream(read_stream, dest_obj)
|
||||
|
||||
assert src_obj == dest_obj
|
||||
|
||||
@pytest.mark.qt_log_ignore('^QIODevice::write: ReadOnly device')
|
||||
def test_serialize_readonly_stream(self):
|
||||
"""Test serialize_stream with a read-only stream."""
|
||||
data = QByteArray()
|
||||
stream = QDataStream(data, QIODevice.ReadOnly)
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
qtutils.serialize_stream(stream, QPoint())
|
||||
assert str(excinfo.value) == ("The data stream cannot write to the "
|
||||
"underlying device.")
|
||||
|
||||
@pytest.mark.qt_log_ignore('QIODevice::read: WriteOnly device')
|
||||
def test_deserialize_writeonly_stream(self):
|
||||
"""Test deserialize_stream with a write-only stream."""
|
||||
data = QByteArray()
|
||||
obj = QPoint()
|
||||
stream = QDataStream(data, QIODevice.WriteOnly)
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
qtutils.deserialize_stream(stream, obj)
|
||||
assert str(excinfo.value) == ("The data stream has read past the end "
|
||||
"of the data in the underlying device.")
|
||||
|
||||
|
||||
class SavefileTestException(Exception):
|
||||
|
||||
"""Exception raised in TestSavefileOpen for testing."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class TestSavefileOpen:
|
||||
|
||||
"""Tests for savefile_open."""
|
||||
|
||||
## Tests with a mock testing that the needed methods are called.
|
||||
|
||||
@pytest.yield_fixture
|
||||
def qsavefile_mock(self, mocker):
|
||||
"""Mock for QSaveFile."""
|
||||
m = mocker.patch('qutebrowser.utils.qtutils.QSaveFile')
|
||||
instance = m()
|
||||
yield instance
|
||||
instance.commit.assert_called_once_with()
|
||||
|
||||
def test_mock_open_error(self, qsavefile_mock):
|
||||
"""Test with a mock and a failing open()."""
|
||||
qsavefile_mock.open.return_value = False
|
||||
qsavefile_mock.errorString.return_value = "Hello World"
|
||||
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
with qtutils.savefile_open('filename'):
|
||||
pass
|
||||
|
||||
qsavefile_mock.open.assert_called_once_with(QIODevice.WriteOnly)
|
||||
qsavefile_mock.cancelWriting.assert_called_once_with()
|
||||
assert str(excinfo.value) == "Hello World"
|
||||
|
||||
def test_mock_exception(self, qsavefile_mock):
|
||||
"""Test with a mock and an exception in the block."""
|
||||
qsavefile_mock.open.return_value = True
|
||||
|
||||
with pytest.raises(SavefileTestException):
|
||||
with qtutils.savefile_open('filename'):
|
||||
raise SavefileTestException
|
||||
|
||||
qsavefile_mock.open.assert_called_once_with(QIODevice.WriteOnly)
|
||||
qsavefile_mock.cancelWriting.assert_called_once_with()
|
||||
|
||||
def test_mock_commit_failed(self, qsavefile_mock):
|
||||
"""Test with a mock and an exception in the block."""
|
||||
qsavefile_mock.open.return_value = True
|
||||
qsavefile_mock.commit.return_value = False
|
||||
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
with qtutils.savefile_open('filename'):
|
||||
pass
|
||||
|
||||
qsavefile_mock.open.assert_called_once_with(QIODevice.WriteOnly)
|
||||
assert not qsavefile_mock.cancelWriting.called
|
||||
assert not qsavefile_mock.errorString.called
|
||||
assert str(excinfo.value) == "Commit failed!"
|
||||
|
||||
def test_mock_successful(self, qsavefile_mock):
|
||||
"""Test with a mock and a successful write."""
|
||||
qsavefile_mock.open.return_value = True
|
||||
qsavefile_mock.errorString.return_value = "Hello World"
|
||||
qsavefile_mock.commit.return_value = True
|
||||
qsavefile_mock.write.side_effect = len
|
||||
qsavefile_mock.isOpen.return_value = True
|
||||
|
||||
with qtutils.savefile_open('filename') as f:
|
||||
f.write("Hello World")
|
||||
|
||||
qsavefile_mock.open.assert_called_once_with(QIODevice.WriteOnly)
|
||||
assert not qsavefile_mock.cancelWriting.called
|
||||
qsavefile_mock.write.assert_called_once_with(b"Hello World")
|
||||
|
||||
## Tests with real files
|
||||
|
||||
@pytest.mark.parametrize('data', ["Hello World", "Snowman! ☃"])
|
||||
def test_utf8(self, data, tmpdir):
|
||||
"""Test with UTF8 data."""
|
||||
filename = tmpdir / 'foo'
|
||||
filename.write("Old data")
|
||||
with qtutils.savefile_open(str(filename)) as f:
|
||||
f.write(data)
|
||||
assert tmpdir.listdir() == [filename]
|
||||
assert filename.read_text(encoding='utf-8') == data
|
||||
|
||||
def test_binary(self, tmpdir):
|
||||
"""Test with binary data."""
|
||||
filename = tmpdir / 'foo'
|
||||
with qtutils.savefile_open(str(filename), binary=True) as f:
|
||||
f.write(b'\xde\xad\xbe\xef')
|
||||
assert tmpdir.listdir() == [filename]
|
||||
assert filename.read_binary() == b'\xde\xad\xbe\xef'
|
||||
|
||||
def test_exception(self, tmpdir):
|
||||
"""Test with an exception in the block."""
|
||||
filename = tmpdir / 'foo'
|
||||
filename.write("Old content")
|
||||
with pytest.raises(SavefileTestException):
|
||||
with qtutils.savefile_open(str(filename)) as f:
|
||||
f.write("Hello World!")
|
||||
raise SavefileTestException
|
||||
assert tmpdir.listdir() == [filename]
|
||||
assert filename.read_text(encoding='utf-8') == "Old content"
|
||||
|
||||
def test_existing_dir(self, tmpdir):
|
||||
"""Test with the filename already occupied by a directory."""
|
||||
filename = tmpdir / 'foo'
|
||||
filename.mkdir()
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
with qtutils.savefile_open(str(filename)):
|
||||
pass
|
||||
assert str(excinfo.value) == "Filename refers to a directory"
|
||||
assert tmpdir.listdir() == [filename]
|
||||
|
||||
def test_failing_commit(self, tmpdir):
|
||||
"""Test with the file being closed before comitting."""
|
||||
filename = tmpdir / 'foo'
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
with qtutils.savefile_open(str(filename), binary=True) as f:
|
||||
f.write(b'Hello')
|
||||
f.dev.commit() # provoke failing "real" commit
|
||||
|
||||
assert str(excinfo.value) == "Commit failed!"
|
||||
assert tmpdir.listdir() == [filename]
|
||||
|
||||
def test_line_endings(self, tmpdir):
|
||||
"""Make sure line endings are translated correctly.
|
||||
|
||||
See https://github.com/The-Compiler/qutebrowser/issues/309
|
||||
"""
|
||||
filename = tmpdir / 'foo'
|
||||
with qtutils.savefile_open(str(filename)) as f:
|
||||
f.write('foo\nbar\nbaz')
|
||||
data = filename.read_binary()
|
||||
if os.name == 'nt':
|
||||
assert data == b'foo\r\nbar\r\nbaz'
|
||||
else:
|
||||
assert data == b'foo\nbar\nbaz'
|
||||
|
||||
|
||||
@pytest.mark.parametrize('orgname, expected', [(None, ''), ('test', 'test')])
|
||||
def test_unset_organization(orgname, expected):
|
||||
"""Test unset_organization.
|
||||
|
||||
Args:
|
||||
orgname: The organizationName to set initially.
|
||||
expected: The organizationName which is expected when reading back.
|
||||
"""
|
||||
app = QApplication.instance()
|
||||
app.setOrganizationName(orgname)
|
||||
assert app.organizationName() == expected # sanity check
|
||||
with qtutils.unset_organization():
|
||||
assert app.organizationName() == ''
|
||||
assert app.organizationName() == expected
|
||||
|
||||
|
||||
if test_file is not None:
|
||||
# If we were able to import Python's test_file module, we run some code
|
||||
# here which defines unittest TestCases to run the python tests over
|
||||
# PyQIODevice.
|
||||
|
||||
@pytest.yield_fixture(scope='session', autouse=True)
|
||||
def clean_up_python_testfile():
|
||||
"""Clean up the python testfile after tests if tests didn't."""
|
||||
yield
|
||||
try:
|
||||
os.remove(test_file.TESTFN)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
class PyIODeviceTestMixin:
|
||||
|
||||
"""Some helper code to run Python's tests with PyQIODevice.
|
||||
|
||||
Attributes:
|
||||
_data: A QByteArray containing the data in memory.
|
||||
f: The opened PyQIODevice.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up self.f using a PyQIODevice instead of a real file."""
|
||||
self._data = QByteArray()
|
||||
self.f = self.open(test_file.TESTFN, 'wb')
|
||||
|
||||
def open(self, _fname, mode):
|
||||
"""Open an in-memory PyQIODevice instead of a real file."""
|
||||
modes = {
|
||||
'wb': QIODevice.WriteOnly | QIODevice.Truncate,
|
||||
'w': QIODevice.WriteOnly | QIODevice.Text | QIODevice.Truncate,
|
||||
'rb': QIODevice.ReadOnly,
|
||||
'r': QIODevice.ReadOnly | QIODevice.Text,
|
||||
}
|
||||
try:
|
||||
qt_mode = modes[mode]
|
||||
except KeyError:
|
||||
raise ValueError("Invalid mode {}!".format(mode))
|
||||
f = QBuffer(self._data)
|
||||
f.open(qt_mode)
|
||||
qiodev = qtutils.PyQIODevice(f)
|
||||
# Make sure tests using name/mode don't blow up.
|
||||
qiodev.name = test_file.TESTFN
|
||||
qiodev.mode = mode
|
||||
# Create empty TESTFN file because the Python tests try to unlink
|
||||
# it.after the test.
|
||||
open(test_file.TESTFN, 'w', encoding='utf-8').close()
|
||||
return qiodev
|
||||
|
||||
class PyAutoFileTests(PyIODeviceTestMixin, test_file.AutoFileTests,
|
||||
unittest.TestCase):
|
||||
|
||||
"""Unittest testcase to run Python's AutoFileTests."""
|
||||
|
||||
def testReadinto_text(self):
|
||||
"""Skip this test as BufferedIOBase seems to fail it."""
|
||||
pass
|
||||
|
||||
class PyOtherFileTests(PyIODeviceTestMixin, test_file.OtherFileTests,
|
||||
unittest.TestCase):
|
||||
|
||||
"""Unittest testcase to run Python's OtherFileTests."""
|
||||
|
||||
def testSetBufferSize(self):
|
||||
"""Skip this test as setting buffer size is unsupported."""
|
||||
pass
|
||||
|
||||
def testTruncateOnWindows(self):
|
||||
"""Skip this test truncating is unsupported."""
|
||||
pass
|
||||
|
||||
|
||||
class FailingQIODevice(QIODevice):
|
||||
|
||||
"""A fake QIODevice where reads/writes fail."""
|
||||
|
||||
def isOpen(self):
|
||||
return True
|
||||
|
||||
def isReadable(self):
|
||||
return True
|
||||
|
||||
def isWritable(self):
|
||||
return True
|
||||
|
||||
def write(self, _data):
|
||||
"""Simulate failed write."""
|
||||
self.setErrorString("Writing failed")
|
||||
return -1
|
||||
|
||||
def read(self, _maxsize):
|
||||
"""Simulate failed read."""
|
||||
self.setErrorString("Reading failed")
|
||||
return None
|
||||
|
||||
def readAll(self):
|
||||
return self.read(0)
|
||||
|
||||
def readLine(self, maxsize):
|
||||
return self.read(maxsize)
|
||||
|
||||
|
||||
class TestPyQIODevice:
|
||||
|
||||
"""Tests for PyQIODevice."""
|
||||
|
||||
@pytest.yield_fixture
|
||||
def pyqiodev(self):
|
||||
"""Fixture providing a PyQIODevice with a QByteArray to test."""
|
||||
data = QByteArray()
|
||||
f = QBuffer(data)
|
||||
qiodev = qtutils.PyQIODevice(f)
|
||||
yield qiodev
|
||||
qiodev.close()
|
||||
|
||||
@pytest.fixture
|
||||
def pyqiodev_failing(self):
|
||||
"""Fixture providing a PyQIODevice with a FailingQIODevice to test."""
|
||||
failing = FailingQIODevice()
|
||||
return qtutils.PyQIODevice(failing)
|
||||
|
||||
@pytest.mark.parametrize('method, args', [
|
||||
('seek', [0]),
|
||||
('flush', []),
|
||||
('isatty', []),
|
||||
('readline', []),
|
||||
('tell', []),
|
||||
('write', [b'']),
|
||||
('read', []),
|
||||
])
|
||||
def test_closed_device(self, pyqiodev, method, args):
|
||||
"""Test various methods with a closed device.
|
||||
|
||||
Args:
|
||||
method: The name of the method to call.
|
||||
args: The arguments to pass.
|
||||
"""
|
||||
func = getattr(pyqiodev, method)
|
||||
with pytest.raises(ValueError) as excinfo:
|
||||
func(*args)
|
||||
assert str(excinfo.value) == "IO operation on closed device!"
|
||||
|
||||
@pytest.mark.parametrize('method', ['readline', 'read'])
|
||||
def test_unreadable(self, pyqiodev, method):
|
||||
"""Test methods with an unreadable device.
|
||||
|
||||
Args:
|
||||
method: The name of the method to call.
|
||||
"""
|
||||
pyqiodev.open(QIODevice.WriteOnly)
|
||||
func = getattr(pyqiodev, method)
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
func()
|
||||
assert str(excinfo.value) == "Trying to read unreadable file!"
|
||||
|
||||
def test_unwritable(self, pyqiodev):
|
||||
"""Test writing with a read-only device."""
|
||||
pyqiodev.open(QIODevice.ReadOnly)
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
pyqiodev.write(b'')
|
||||
assert str(excinfo.value) == "Trying to write to unwritable file!"
|
||||
|
||||
@pytest.mark.parametrize('data', [b'12345', b''])
|
||||
def test_len(self, pyqiodev, data):
|
||||
"""Test len()/__len__.
|
||||
|
||||
Args:
|
||||
data: The data to write before checking if the length equals
|
||||
len(data).
|
||||
"""
|
||||
pyqiodev.open(QIODevice.WriteOnly)
|
||||
pyqiodev.write(data)
|
||||
assert len(pyqiodev) == len(data)
|
||||
|
||||
def test_failing_open(self, tmpdir):
|
||||
"""Test open() which fails (because it's an existant directory)."""
|
||||
qf = QFile(str(tmpdir))
|
||||
dev = qtutils.PyQIODevice(qf)
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
dev.open(QIODevice.WriteOnly)
|
||||
assert str(excinfo.value) == 'Is a directory'
|
||||
assert dev.closed
|
||||
|
||||
def test_fileno(self, pyqiodev):
|
||||
with pytest.raises(io.UnsupportedOperation):
|
||||
pyqiodev.fileno()
|
||||
|
||||
@pytest.mark.qt_log_ignore('^QBuffer::seek: Invalid pos:')
|
||||
@pytest.mark.parametrize('offset, whence, pos, data, raising', [
|
||||
(0, io.SEEK_SET, 0, b'1234567890', False),
|
||||
(42, io.SEEK_SET, 0, b'1234567890', True),
|
||||
(8, io.SEEK_CUR, 8, b'90', False),
|
||||
(-5, io.SEEK_CUR, 0, b'1234567890', True),
|
||||
(-2, io.SEEK_END, 8, b'90', False),
|
||||
(2, io.SEEK_END, 0, b'1234567890', True),
|
||||
(0, io.SEEK_END, 10, b'', False),
|
||||
])
|
||||
def test_seek_tell(self, pyqiodev, offset, whence, pos, data, raising):
|
||||
"""Test seek() and tell().
|
||||
|
||||
The initial position when these tests run is 0.
|
||||
|
||||
Args:
|
||||
offset: The offset to pass to .seek().
|
||||
whence: The whence argument to pass to .seek().
|
||||
pos: The expected position after seeking.
|
||||
data: The expected data to read after seeking.
|
||||
raising: Whether seeking should raise OSError.
|
||||
"""
|
||||
with pyqiodev.open(QIODevice.WriteOnly) as f:
|
||||
f.write(b'1234567890')
|
||||
pyqiodev.open(QIODevice.ReadOnly)
|
||||
if raising:
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
pyqiodev.seek(offset, whence)
|
||||
assert str(excinfo.value) == "seek failed!"
|
||||
else:
|
||||
pyqiodev.seek(offset, whence)
|
||||
assert pyqiodev.tell() == pos
|
||||
assert pyqiodev.read() == data
|
||||
|
||||
def test_seek_unsupported(self, pyqiodev):
|
||||
"""Test seeking with unsupported whence arguments."""
|
||||
if hasattr(os, 'SEEK_HOLE'):
|
||||
whence = os.SEEK_HOLE
|
||||
elif hasattr(os, 'SEEK_DATA'):
|
||||
whence = os.SEEK_DATA
|
||||
else:
|
||||
pytest.skip("Needs os.SEEK_HOLE or os.SEEK_DATA available.")
|
||||
pyqiodev.open(QIODevice.ReadOnly)
|
||||
with pytest.raises(io.UnsupportedOperation):
|
||||
pyqiodev.seek(0, whence)
|
||||
|
||||
def test_qprocess(self):
|
||||
"""Test PyQIODevice with a QProcess which is non-sequential.
|
||||
|
||||
This also verifies seek() and tell() behave as expected.
|
||||
"""
|
||||
proc = QProcess()
|
||||
proc.start(sys.executable, ['-c', 'print("Hello World")'])
|
||||
dev = qtutils.PyQIODevice(proc)
|
||||
assert not dev.closed
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
dev.seek(0)
|
||||
assert str(excinfo.value) == 'Random access not allowed!'
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
dev.tell()
|
||||
assert str(excinfo.value) == 'Random access not allowed!'
|
||||
proc.waitForFinished(1000)
|
||||
proc.kill()
|
||||
assert dev.read() == b'Hello World\n'
|
||||
|
||||
def test_truncate(self, pyqiodev):
|
||||
with pytest.raises(io.UnsupportedOperation):
|
||||
pyqiodev.truncate()
|
||||
|
||||
def test_closed(self, pyqiodev):
|
||||
"""Test the closed attribute."""
|
||||
assert pyqiodev.closed
|
||||
pyqiodev.open(QIODevice.ReadOnly)
|
||||
assert not pyqiodev.closed
|
||||
pyqiodev.close()
|
||||
assert pyqiodev.closed
|
||||
|
||||
def test_contextmanager(self, pyqiodev):
|
||||
"""Make sure using the PyQIODevice as context manager works."""
|
||||
assert pyqiodev.closed
|
||||
with pyqiodev.open(QIODevice.ReadOnly) as f:
|
||||
assert not f.closed
|
||||
assert f is pyqiodev
|
||||
assert pyqiodev.closed
|
||||
|
||||
def test_flush(self, pyqiodev):
|
||||
"""Make sure flushing doesn't raise an exception."""
|
||||
pyqiodev.open(QIODevice.WriteOnly)
|
||||
pyqiodev.write(b'test')
|
||||
pyqiodev.flush()
|
||||
|
||||
@pytest.mark.parametrize('method, ret', [
|
||||
('isatty', False),
|
||||
('seekable', True),
|
||||
])
|
||||
def test_bools(self, method, ret, pyqiodev):
|
||||
"""Make sure simple bool arguments return the right thing.
|
||||
|
||||
Args:
|
||||
method: The name of the method to call.
|
||||
ret: The return value we expect.
|
||||
"""
|
||||
pyqiodev.open(QIODevice.WriteOnly)
|
||||
func = getattr(pyqiodev, method)
|
||||
assert func() == ret
|
||||
|
||||
@pytest.mark.parametrize('mode, readable, writable', [
|
||||
(QIODevice.ReadOnly, True, False),
|
||||
(QIODevice.ReadWrite, True, True),
|
||||
(QIODevice.WriteOnly, False, True),
|
||||
])
|
||||
def test_readable_writable(self, mode, readable, writable, pyqiodev):
|
||||
"""Test readable() and writable().
|
||||
|
||||
Args:
|
||||
mode: The mode to open the PyQIODevice in.
|
||||
readable: Whether the device should be readable.
|
||||
writable: Whether the device should be writable.
|
||||
"""
|
||||
assert not pyqiodev.readable()
|
||||
assert not pyqiodev.writable()
|
||||
pyqiodev.open(mode)
|
||||
assert pyqiodev.readable() == readable
|
||||
assert pyqiodev.writable() == writable
|
||||
|
||||
@pytest.mark.parametrize('size, chunks', [
|
||||
(-1, [b'one\n', b'two\n', b'three', b'']),
|
||||
(0, [b'', b'', b'', b'']),
|
||||
(2, [b'on', b'e\n', b'tw', b'o\n', b'th', b're', b'e']),
|
||||
(10, [b'one\n', b'two\n', b'three', b'']),
|
||||
])
|
||||
def test_readline(self, size, chunks, pyqiodev):
|
||||
"""Test readline() with different sizes.
|
||||
|
||||
Args:
|
||||
size: The size to pass to readline()
|
||||
chunks: A list of expected chunks to read.
|
||||
"""
|
||||
with pyqiodev.open(QIODevice.WriteOnly) as f:
|
||||
f.write(b'one\ntwo\nthree')
|
||||
pyqiodev.open(QIODevice.ReadOnly)
|
||||
for i, chunk in enumerate(chunks, start=1):
|
||||
print("Expecting chunk {}: {!r}".format(i, chunk))
|
||||
assert pyqiodev.readline(size) == chunk
|
||||
|
||||
def test_write(self, pyqiodev):
|
||||
"""Make sure writing and re-reading works."""
|
||||
with pyqiodev.open(QIODevice.WriteOnly) as f:
|
||||
f.write(b'foo\n')
|
||||
f.write(b'bar\n')
|
||||
pyqiodev.open(QIODevice.ReadOnly)
|
||||
assert pyqiodev.read() == b'foo\nbar\n'
|
||||
|
||||
def test_write_error(self, pyqiodev_failing):
|
||||
"""Test writing with FailingQIODevice."""
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
pyqiodev_failing.write(b'x')
|
||||
assert str(excinfo.value) == 'Writing failed'
|
||||
|
||||
@pytest.mark.skipif(os.name != 'posix', reason="Needs a POSIX OS.")
|
||||
@pytest.mark.skipif(not os.path.exists('/dev/full'),
|
||||
reason="Needs /dev/full.")
|
||||
def test_write_error_real(self):
|
||||
"""Test a real write error with /dev/full on supported systems."""
|
||||
qf = QFile('/dev/full')
|
||||
qf.open(QIODevice.WriteOnly | QIODevice.Unbuffered)
|
||||
dev = qtutils.PyQIODevice(qf)
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
dev.write(b'foo')
|
||||
qf.close()
|
||||
assert str(excinfo.value) == 'No space left on device'
|
||||
|
||||
@pytest.mark.parametrize('size, chunks', [
|
||||
(-1, [b'1234567890']),
|
||||
(0, [b'']),
|
||||
(3, [b'123', b'456', b'789', b'0']),
|
||||
(20, [b'1234567890'])
|
||||
])
|
||||
def test_read(self, size, chunks, pyqiodev):
|
||||
"""Test reading with different sizes.
|
||||
|
||||
Args:
|
||||
size: The size to pass to read()
|
||||
chunks: A list of expected data chunks.
|
||||
"""
|
||||
with pyqiodev.open(QIODevice.WriteOnly) as f:
|
||||
f.write(b'1234567890')
|
||||
pyqiodev.open(QIODevice.ReadOnly)
|
||||
for i, chunk in enumerate(chunks):
|
||||
print("Expecting chunk {}: {!r}".format(i, chunk))
|
||||
assert pyqiodev.read(size) == chunk
|
||||
|
||||
@pytest.mark.parametrize('method, args', [
|
||||
('read', []),
|
||||
('read', [5]),
|
||||
('readline', []),
|
||||
('readline', [5]),
|
||||
])
|
||||
def test_failing_reads(self, method, args, pyqiodev_failing):
|
||||
"""Test reading with a FailingQIODevice.
|
||||
|
||||
Args:
|
||||
method: The name of the method to call.
|
||||
args: A list of arguments to pass.
|
||||
"""
|
||||
func = getattr(pyqiodev_failing, method)
|
||||
with pytest.raises(OSError) as excinfo:
|
||||
func(*args)
|
||||
assert str(excinfo.value) == 'Reading failed'
|
||||
|
||||
|
||||
class TestEventLoop:
|
||||
|
||||
"""Tests for EventLoop.
|
||||
|
||||
Attributes:
|
||||
loop: The EventLoop we're testing.
|
||||
"""
|
||||
|
||||
# pylint: disable=protected-access
|
||||
|
||||
def _assert_executing(self):
|
||||
"""Slot which gets called from timers to be sure the loop runs."""
|
||||
assert self.loop._executing
|
||||
|
||||
def _double_exec(self):
|
||||
"""Slot which gets called from timers to assert double-exec fails."""
|
||||
with pytest.raises(AssertionError):
|
||||
self.loop.exec_()
|
||||
|
||||
def test_normal_exec(self):
|
||||
"""Test exec_ without double-executing."""
|
||||
self.loop = qtutils.EventLoop()
|
||||
QTimer.singleShot(100, self._assert_executing)
|
||||
QTimer.singleShot(200, self.loop.quit)
|
||||
self.loop.exec_()
|
||||
assert not self.loop._executing
|
||||
|
||||
def test_double_exec(self):
|
||||
"""Test double-executing."""
|
||||
self.loop = qtutils.EventLoop()
|
||||
QTimer.singleShot(100, self._assert_executing)
|
||||
QTimer.singleShot(200, self._double_exec)
|
||||
QTimer.singleShot(300, self._assert_executing)
|
||||
QTimer.singleShot(400, self.loop.quit)
|
||||
self.loop.exec_()
|
||||
assert not self.loop._executing
|
||||
|
||||
Loading…
Reference in New Issue
Block a user