Move example test code such that pytest is required
This commit is contained in:
Ogi
parent
eb90616ae2
commit
5ff409ba4b
@ -9,8 +9,8 @@ import subprocess
|
||||
from pyqtgraph.python2_3 import basestring
|
||||
from pyqtgraph.Qt import QtGui, QT_LIB
|
||||
|
||||
from .utils import buildFileList, path, examples
|
||||
|
||||
from .utils import buildFileList, testFile, path, examples
|
||||
|
||||
if QT_LIB == 'PySide':
|
||||
from .exampleLoaderTemplate_pyside import Ui_Form
|
||||
@ -117,32 +117,7 @@ class ExampleLoader(QtGui.QMainWindow):
|
||||
def run():
|
||||
app = QtGui.QApplication([])
|
||||
loader = ExampleLoader()
|
||||
|
||||
app.exec_()
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
args = sys.argv[1:]
|
||||
|
||||
if '--test' in args:
|
||||
# get rid of orphaned cache files first
|
||||
pg.renamePyc(path)
|
||||
|
||||
files = buildFileList(examples)
|
||||
if '--pyside' in args:
|
||||
lib = 'PySide'
|
||||
elif '--pyqt' in args or '--pyqt4' in args:
|
||||
lib = 'PyQt4'
|
||||
elif '--pyqt5' in args:
|
||||
lib = 'PyQt5'
|
||||
elif '--pyside2' in args:
|
||||
lib = 'PySide2'
|
||||
else:
|
||||
lib = ''
|
||||
|
||||
exe = sys.executable
|
||||
print("Running tests:", lib, sys.executable)
|
||||
for f in files:
|
||||
testFile(f[0], f[1], exe, lib)
|
||||
else:
|
||||
run()
|
||||
run()
|
||||
|
||||
@ -1,9 +1,87 @@
|
||||
from __future__ import print_function, division, absolute_import
|
||||
from pyqtgraph import Qt
|
||||
from . import utils
|
||||
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__))
|
||||
|
||||
|
||||
def runExampleFile(name, f, exe, lib, graphicsSystem=None):
|
||||
global path
|
||||
fn = os.path.join(path,f)
|
||||
os.chdir(path)
|
||||
sys.stdout.write("{} ".format(name))
|
||||
sys.stdout.flush()
|
||||
import1 = "import %s" % lib if lib != '' 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([exe],
|
||||
stdin=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE)
|
||||
else:
|
||||
process = subprocess.Popen(['exec %s -i' % (exe)],
|
||||
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())
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
# printing on travis ci frequently leads to "interrupted system call" errors.
|
||||
@ -32,16 +110,7 @@ if os.getenv('TRAVIS') is not None:
|
||||
print("Installed wrapper for flaky print.")
|
||||
|
||||
|
||||
# apparently importlib does not exist in python 2.6...
|
||||
try:
|
||||
import importlib
|
||||
except ImportError:
|
||||
# we are on python 2.6
|
||||
print("If you want to test the examples, please install importlib from "
|
||||
"pypi\n\npip install importlib\n\n")
|
||||
pass
|
||||
|
||||
files = utils.buildFileList(utils.tested_examples)
|
||||
files = 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():
|
||||
@ -50,16 +119,32 @@ for frontend in frontends.keys():
|
||||
frontends[frontend] = True
|
||||
except ImportError:
|
||||
pass
|
||||
except ModuleNotFoundError:
|
||||
pass
|
||||
|
||||
installed = sorted([frontend for frontend, isPresent in frontends.items() if isPresent])
|
||||
|
||||
# keep a dictionary of example files and their non-standard dependencies
|
||||
specialExamples = {
|
||||
"hdf5.py": ["h5py"]
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"frontend, f", itertools.product(sorted(list(frontends.keys())), files))
|
||||
def test_examples(frontend, f):
|
||||
# Test the examples with all available front-ends
|
||||
print('frontend = %s. f = %s' % (frontend, f))
|
||||
if not frontends[frontend]:
|
||||
pytest.skip('%s is not installed. Skipping tests' % frontend)
|
||||
utils.testFile(f[0], f[1], utils.sys.executable, frontend)
|
||||
"frontend, f",
|
||||
[
|
||||
pytest.param(
|
||||
frontend,
|
||||
f,
|
||||
marks=pytest.mark.skipif(any(pkgutil.find_loader(pkg) is None for pkg in specialExamples[f[1]]),
|
||||
reason="Skipping Example for Missing Dependencies") if f[1] in specialExamples.keys() else (),
|
||||
)
|
||||
for frontend, f, in itertools.product(installed, files)
|
||||
],
|
||||
ids = [" {} - {} ".format(f[1], frontend) for frontend, f in itertools.product(installed, files)]
|
||||
)
|
||||
def testExamples(frontend, f):
|
||||
assert runExampleFile(f[0], f[1], sys.executable, frontend)
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.cmdline.main()
|
||||
|
||||
@ -1,10 +1,5 @@
|
||||
from __future__ import division, print_function, absolute_import
|
||||
import subprocess
|
||||
import time
|
||||
import os
|
||||
import sys
|
||||
import errno
|
||||
import copy
|
||||
from pyqtgraph.pgcollections import OrderedDict
|
||||
from pyqtgraph.python2_3 import basestring
|
||||
|
||||
@ -87,16 +82,10 @@ examples = OrderedDict([
|
||||
#('VerticalLabel', '../widgets/VerticalLabel.py'),
|
||||
('JoystickButton', 'JoystickButton.py'),
|
||||
])),
|
||||
|
||||
('Flowcharts', 'Flowchart.py'),
|
||||
('Custom Flowchart Nodes', 'FlowchartCustomNode.py'),
|
||||
])
|
||||
|
||||
not_tested = ['HDF5 big data']
|
||||
|
||||
tested_examples = copy.deepcopy(examples)
|
||||
all(map(tested_examples.pop, not_tested))
|
||||
|
||||
|
||||
def buildFileList(examples, files=None):
|
||||
if files == None:
|
||||
@ -109,73 +98,3 @@ def buildFileList(examples, files=None):
|
||||
else:
|
||||
buildFileList(val, files)
|
||||
return files
|
||||
|
||||
def testFile(name, f, exe, lib, graphicsSystem=None):
|
||||
global path
|
||||
fn = os.path.join(path,f)
|
||||
#print "starting process: ", fn
|
||||
os.chdir(path)
|
||||
sys.stdout.write(name)
|
||||
sys.stdout.flush()
|
||||
|
||||
import1 = "import %s" % lib if lib != '' 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([exe], stdin=subprocess.PIPE, stderr=subprocess.PIPE, stdout=subprocess.PIPE)
|
||||
process.stdin.write(code.encode('UTF-8'))
|
||||
process.stdin.close()
|
||||
else:
|
||||
process = subprocess.Popen(['exec %s -i' % (exe)], 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
|
||||
#sys.stdout.write(c)
|
||||
#sys.stdout.flush()
|
||||
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('.' * (50-len(name)) + 'FAILED')
|
||||
print(res[0].decode())
|
||||
print(res[1].decode())
|
||||
else:
|
||||
print('.' * (50-len(name)) + 'passed')
|
||||
|
||||
Loading…
Reference in New Issue
Block a user