pyqtgraph/examples/test_examples.py

from __future__ import print_function, division, absolute_import
from pyqtgraph import Qt
from . import utils
from collections import namedtuple
import errno
import importlib
import itertools
import pkgutil
import pytest
import os, sys
import subprocess
import time


path = os.path.abspath(os.path.dirname(__file__))

# printing on travis ci frequently leads to "interrupted system call" errors.
# as a workaround, we overwrite the built-in print function (bleh)
if os.getenv('TRAVIS') is not None:
    if sys.version_info[0] < 3:
        import __builtin__ as builtins
    else:
        import builtins

    def flaky_print(*args):
        """Wrapper for print that retries in case of IOError.
        """
        count = 0
        while count < 5:
            count += 1
            try:
                orig_print(*args)
                break
            except IOError:
                if count >= 5:
                    raise
                pass
    orig_print = builtins.print
    builtins.print = flaky_print
    print("Installed wrapper for flaky print.")


files = sorted(set(utils.buildFileList(utils.examples)))
frontends = {Qt.PYQT4: False, Qt.PYQT5: False, Qt.PYSIDE: False, Qt.PYSIDE2: False}
# sort out which of the front ends are available
for frontend in frontends.keys():
    try:
        importlib.import_module(frontend)
        frontends[frontend] = True
    except ImportError:
        pass

installedFrontends = sorted([frontend for frontend, isPresent in frontends.items() if isPresent])

exceptionCondition = namedtuple("exceptionCondition", ["condition", "reason"])
conditionalExampleTests = {
    "hdf5.py": exceptionCondition(False, reason="Example requires user interaction and is not suitable for testing"),
    "RemoteSpeedTest.py": exceptionCondition(False, reason="Test is being problematic on CI machines"),
    "optics_demos.py": exceptionCondition(not frontends[Qt.PYSIDE], reason="Test fails due to PySide bug: https://bugreports.qt.io/browse/PYSIDE-671")
}


@pytest.mark.parametrize(
	"frontend, f",
	[
		pytest.param(
			frontend,
            f,
            marks=pytest.mark.skipif(conditionalExampleTests[f[1]].condition is False,
                                     reason=conditionalExampleTests[f[1]].reason) if f[1] in conditionalExampleTests.keys() else (),
		)
		for frontend, f, in itertools.product(installedFrontends, files)
	],
    ids = [" {} - {} ".format(f[1], frontend) for frontend, f in itertools.product(installedFrontends, files)]
)
def testExamples(frontend, f, graphicsSystem=None):
    # runExampleFile(f[0], f[1], sys.executable, frontend)

    name, file = f
    global path
    fn = os.path.join(path,file)
    os.chdir(path)
    sys.stdout.write("{} ".format(name))
    sys.stdout.flush()
    import1 = "import %s" % frontend if frontend != '' else ''
    import2 = os.path.splitext(os.path.split(fn)[1])[0]
    graphicsSystem = '' if graphicsSystem is None else "pg.QtGui.QApplication.setGraphicsSystem('%s')" % graphicsSystem
    code = """
try:
    %s
    import initExample
    import pyqtgraph as pg
    %s
    import %s
    import sys
    print("test complete")
    sys.stdout.flush()
    import time
    while True:  ## run a little event loop
        pg.QtGui.QApplication.processEvents()
        time.sleep(0.01)
except:
    print("test failed")
    raise

""" % (import1, graphicsSystem, import2)
    if sys.platform.startswith('win'):
        process = subprocess.Popen([sys.executable],
                                    stdin=subprocess.PIPE,
                                    stderr=subprocess.PIPE,
                                    stdout=subprocess.PIPE)
    else:
        process = subprocess.Popen(['exec %s -i' % (sys.executable)],
                                   shell=True,
                                   stdin=subprocess.PIPE,
                                   stderr=subprocess.PIPE,
                                   stdout=subprocess.PIPE)
    process.stdin.write(code.encode('UTF-8'))
    process.stdin.close() ##?
    output = ''
    fail = False
    while True:
        try:
            c = process.stdout.read(1).decode()
        except IOError as err:
            if err.errno == errno.EINTR:
                # Interrupted system call; just try again.
                c = ''
            else:
                raise
        output += c

        if output.endswith('test complete'):
            break
        if output.endswith('test failed'):
            fail = True
            break
    time.sleep(1)
    process.kill()
    #res = process.communicate()
    res = (process.stdout.read(), process.stderr.read())
    if fail or 'exception' in res[1].decode().lower() or 'error' in res[1].decode().lower():
        print(res[0].decode())
        print(res[1].decode())
        pytest.fail("{}\n{}\nFailed {} Example Test Located in {} ".format(res[0].decode(), res[1].decode(), name, file), pytrace=False)

if __name__ == "__main__":
    pytest.cmdline.main()