# vim: ft=python fileencoding=utf-8 sts=4 sw=4 et:
# Copyright 2014-2015 Florian Bruhin (The Compiler) <mail@qutebrowser.org>
#
# 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 <http://www.gnu.org/licenses/>.
"""Tests for qutebrowser.utils.qtutils."""
import io
import os
import sys
import operator
import os.path
try:
from test import test_file # pylint: disable=no-name-in-module
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."""
@pytest.mark.parametrize('ctype, val',
overflow_test_cases.iter_good_values())
def test_good_values(self, ctype, val):
"""Test values which are inside bounds."""
qtutils.check_overflow(val, ctype)
@pytest.mark.parametrize('ctype, val',
[(ctype, val) for (ctype, val, _) in
overflow_test_cases.iter_bad_values()])
def test_bad_values_fatal(self, ctype, val):
"""Test values which are outside bounds with fatal=True."""
with pytest.raises(OverflowError):
qtutils.check_overflow(val, ctype)
@pytest.mark.parametrize('ctype, val, repl',
overflow_test_cases.iter_bad_values())
def test_bad_values_nonfatal(self, ctype, val, repl):
"""Test values which are outside bounds with fatal=False."""
newval = qtutils.check_overflow(val, ctype, fatal=False)
assert newval == repl
class TestGetQtArgs:
"""Tests for get_args."""
@pytest.fixture
def parser(self, mocker):
"""Fixture to provide an argparser.
Monkey-patches .exit() of the argparser so it doesn't exit on errors.
"""
parser = qutebrowser.get_argparser()
mocker.patch.object(parser, 'exit', side_effect=Exception)
return 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."""
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."""
args = parser.parse_args(['--qt-stylesheet', 'foobar', '--qt-reverse'])
qt_args = qtutils.get_args(args)
assert qt_args[0] == sys.argv[0]
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
errors = ["Filename refers to a directory", # Qt >= 5.4
"Commit failed!"] # older Qt versions
assert str(excinfo.value) in errors
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)
errors = ['Access is denied.', # Linux/OS X
'Is a directory'] # Windows
assert str(excinfo.value) in errors
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 # pylint: disable=no-member
elif hasattr(os, 'SEEK_DATA'):
whence = os.SEEK_DATA # pylint: disable=no-member
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)
@pytest.mark.not_frozen
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 bytes(dev.read()).rstrip() == b'Hello World'
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.posix
@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