Aluode/PerceptionLabPortable

1import contextlib2import errno3import importlib4import itertools5import os6import platform7import subprocess8import sys9import time10from argparse import Namespace11from collections import namedtuple12 13import pytest14 15from pyqtgraph import Qt16 17if __name__ == "__main__" and (__package__ is None or __package__==''):18    parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))19    sys.path.insert(0, parent_dir)20    import examples21    __package__ = "examples"22 23from . import utils24 25 26def buildFileList(examples, files=None):27    if files is None:28        files = []29    for key, val in examples.items():30        if isinstance(val, dict):31            buildFileList(val, files)32        elif isinstance(val, Namespace):33            files.append((key, val.filename))34        else:35            files.append((key, val))36    return files37 38 39path = os.path.abspath(os.path.dirname(__file__))40files = [("Example App", "RunExampleApp.py")]41for ex in [utils.examples_, utils.others]:42    files = buildFileList(ex, files)43files = sorted(set(files))44frontends = {45    Qt.PYQT5: False,46    Qt.PYQT6: False,47    Qt.PYSIDE2: False,48    Qt.PYSIDE6: False,49}50# sort out which of the front ends are available51for frontend in frontends.keys():52    with contextlib.suppress(ImportError):53        importlib.import_module(frontend)54        frontends[frontend] = True55 56installedFrontends = sorted([57    frontend for frontend, isPresent in frontends.items() if isPresent58])59 60 61 62exceptionCondition = namedtuple("exceptionCondition", ["condition", "reason"])63conditionalExamples = {64    "hdf5.py": exceptionCondition(65        False,66        reason="Example requires user interaction"67    ),68    "jupyter_console_example.py": exceptionCondition(69        importlib.util.find_spec("qtconsole") is not None,70        reason="No need to test with qtconsole not being installed"71    ),72    "RemoteSpeedTest.py": exceptionCondition(73        False,74        reason="Test is being problematic on CI machines"75    ),76}77 78 79@pytest.mark.parametrize("frontend, f", [80        pytest.param(81            frontend,82            f,83            marks=pytest.mark.skipif(84                conditionalExamples[f[1]].condition is False,85                reason=conditionalExamples[f[1]].reason86            ) if f[1] in conditionalExamples.keys() else (),87        )88        for frontend, f, in itertools.product(installedFrontends, files)89    ],90    ids=[91        f" {f[1]} - {frontend} " for frontend, f in itertools.product(92            installedFrontends,93            files94        )95    ]96)97def testExamples(frontend, f):98    name, file = f99    global path100    fn = os.path.join(path, file)101    os.chdir(path)102    sys.stdout.write(f"{name}")103    sys.stdout.flush()104    import1 = f"import {frontend}" if frontend != '' else ''105    import2 = os.path.splitext(os.path.split(fn)[1])[0]106    code = """107try:108    {0}109    import faulthandler110    faulthandler.enable()111    import pyqtgraph as pg112    import {1}113    import sys114    print("test complete")115    sys.stdout.flush()116    pg.Qt.QtCore.QTimer.singleShot(1000, pg.Qt.QtWidgets.QApplication.quit)117    pg.exec()118    names = [x for x in dir({1}) if not x.startswith('_')]119    for name in names:120        delattr({1}, name)121except:122    print("test failed")123    raise124 125""".format(import1, import2)126    env = dict(os.environ)127    example_dir = os.path.abspath(os.path.dirname(__file__))128    path = os.path.dirname(os.path.dirname(example_dir))129    env['PYTHONPATH'] = f'{path}{os.pathsep}{example_dir}'130    process = subprocess.Popen([sys.executable],131                                stdin=subprocess.PIPE,132                                stderr=subprocess.PIPE,133                                stdout=subprocess.PIPE,134                                text=True,135                                env=env)136    process.stdin.write(code)137    process.stdin.close()138 139    output = ''140    fail = False141    while True:142        try:143            c = process.stdout.read(1)144        except IOError as err:145            if err.errno == errno.EINTR:146                # Interrupted system call; just try again.147                c = ''148            else:149                raise150        output += c151        if output.endswith('test complete'):152            break153        if output.endswith('test failed'):154            fail = True155            break156    start = time.time()157    killed = False158    while process.poll() is None:159        time.sleep(0.1)160        if time.time() - start > 2.0 and not killed:161            process.kill()162            killed = True163 164    stdout, stderr = (process.stdout.read(), process.stderr.read())165    process.stdout.close()166    process.stderr.close()167 168    if (fail or169        'Exception:' in stderr or170        'Error:' in stderr):171        if (not fail 172            and name == "RemoteGraphicsView" 173            and "pyqtgraph.multiprocess.remoteproxy.ClosedError" in stderr):174            # This test can intermittently fail when the subprocess is killed175            return None176        print(stdout)177        print(stderr)178        pytest.fail(179            f"{stdout}\n{stderr}\nFailed {name} Example Test Located in {file}",180            pytrace=False181        )182 183if __name__ == "__main__":184    pytest.cmdline.main()185