From 1e982a9a84784b870a00fba13c689e146d1ffbdc Mon Sep 17 00:00:00 2001 From: Florian Bruhin Date: Mon, 25 May 2015 01:21:22 +0200 Subject: [PATCH] Add/improve tests for qutebrowser.utils.qtutils. --- tests/utils/test_qtutils.py | 889 +++++++++++++++++++++++++++++++++++- 1 file changed, 876 insertions(+), 13 deletions(-) diff --git a/tests/utils/test_qtutils.py b/tests/utils/test_qtutils.py index 85a99c2f4..220297761 100644 --- a/tests/utils/test_qtutils.py +++ b/tests/utils/test_qtutils.py @@ -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 '' + + 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, + ' is not valid'), + ('ensure_valid', QtObject(valid=False, null=False), True, None, + ' is not valid'), + ('ensure_valid', QtObject(valid=False, null=True, error='Test'), True, + 'Test', ' 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, + ' is null'), + ('ensure_not_null', QtObject(valid=False, null=True), True, None, + ' is null'), + ('ensure_not_null', QtObject(valid=False, null=True, error='Test'), True, + 'Test', ' 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