Commit c688345a authored by payno's avatar payno
Browse files

[core][process] add process function

parent b1a2b274
File mode changed from 100644 to 100755
......@@ -2,7 +2,7 @@
# coding: utf-8
# /*##########################################################################
#
# Copyright (c) 2015-2017 European Synchrotron Radiation Facility
# Copyright (c) 2015-2018 European Synchrotron Radiation Facility
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
......@@ -28,11 +28,11 @@
This script expects a suite function in <project_package>.test,
which returns a unittest.TestSuite.
test coverage dependencies: coverage, lxml.
Test coverage dependencies: coverage, lxml.
"""
__authors__ = ["Jérôme Kieffer", "Thomas Vincent"]
__date__ = "29/11/2016"
__date__ = "02/03/2018"
__license__ = "MIT"
import distutils.util
......@@ -42,16 +42,52 @@ import subprocess
import sys
import time
import unittest
# from silx.gui import plot
# needed for gitlab unit test. Otherwise fail in an
# `ImportError: dlopen: cannot load any more object with static TLS`
import collections
from argparse import ArgumentParser
class StreamHandlerUnittestReady(logging.StreamHandler):
"""The unittest class TestResult redefine sys.stdout/err to capture
stdout/err from tests and to display them only when a test fail.
This class allow to use unittest stdout-capture by using the last sys.stdout
and not a cached one.
"""
def emit(self, record):
"""
:type record: logging.LogRecord
"""
self.stream = sys.stderr
super(StreamHandlerUnittestReady, self).emit(record)
def flush(self):
pass
def createBasicHandler():
"""Create the handler using the basic configuration"""
hdlr = StreamHandlerUnittestReady()
fs = logging.BASIC_FORMAT
dfs = None
fmt = logging.Formatter(fs, dfs)
hdlr.setFormatter(fmt)
return hdlr
# Use an handler compatible with unittests, else use_buffer is not working
logging.root.addHandler(createBasicHandler())
# Capture all default warnings
logging.captureWarnings(True)
import warnings
warnings.simplefilter('default')
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger("run_tests")
logger.setLevel(logging.ERROR)
logger.setLevel(logging.WARNING)
logger.info("Python %s %s", sys.version, tuple.__itemsize__ * 8)
try:
import resource
except ImportError:
......@@ -60,10 +96,16 @@ except ImportError:
try:
import importlib
except:
importer = __import__
else:
importer = importlib.import_module
except ImportError:
def importer(name):
module = __import__(name)
# returns the leaf module, instead of the root module
subnames = name.split(".")
subnames.pop(0)
for subname in subnames:
module = getattr(module, subname)
return module
try:
......@@ -73,7 +115,6 @@ except Exception as error:
else:
logger.info("Numpy %s", numpy.version.version)
try:
import h5py
except Exception as error:
......@@ -96,9 +137,31 @@ def get_project_name(root_dir):
return name.split()[-1].decode('ascii')
PROJECT_DIR = os.path.dirname(os.path.abspath(__file__))
PROJECT_NAME = "tomwer"
logger.info("Project name: %s", PROJECT_NAME)
class TextTestResultWithSkipList(unittest.TextTestResult):
"""Override default TextTestResult to display list of skipped tests at the
end
"""
def printErrors(self):
unittest.TextTestResult.printErrors(self)
# Print skipped tests at the end
self.printGroupedList("SKIPPED", self.skipped)
def printGroupedList(self, flavour, errors):
grouped = collections.OrderedDict()
for test, err in errors:
if err in grouped:
grouped[err] = grouped[err] + [test]
else:
grouped[err] = [test]
for err, tests in grouped.items():
self.stream.writeln(self.separator1)
for test in tests:
self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
self.stream.writeln(self.separator2)
self.stream.writeln("%s" % err)
class ProfileTextTestResult(unittest.TextTestRunner.resultclass):
......@@ -129,7 +192,8 @@ class ProfileTextTestResult(unittest.TextTestRunner.resultclass):
else:
memusage = 0
self.logger.info("Time: %.3fs \t RAM: %.3f Mb\t%s",
time.time() - self.__time_start, memusage, test.id())
time.time() - self.__time_start,
memusage, test.id())
def report_rst(cov, package, version="0.0.0", base=""):
......@@ -150,12 +214,12 @@ def report_rst(cov, package, version="0.0.0", base=""):
xml = etree.parse(fn)
classes = xml.xpath("//class")
line0 = "test coverage report for %s" % package
line0 = "Test coverage report for %s" % package
res = [line0, "=" * len(line0), ""]
res.append("Measured on *%s* version %s, %s" %
(package, version, time.strftime("%d/%m/%Y")))
res += ["",
".. csv-table:: test suite coverage",
".. csv-table:: Test suite coverage",
' :header: "Name", "Stmts", "Exec", "Cover"',
' :widths: 35, 8, 8, 8',
'']
......@@ -190,6 +254,20 @@ def report_rst(cov, package, version="0.0.0", base=""):
return os.linesep.join(res)
def is_debug_python():
"""Returns true if the Python interpreter is in debug mode."""
try:
import sysconfig
except ImportError: # pragma nocover
# Python < 2.7
import distutils.sysconfig as sysconfig
if sysconfig.get_config_var("Py_DEBUG"):
return True
return hasattr(sys, "gettotalrefcount")
def build_project(name, root_dir):
"""Run python setup.py build for the project.
......@@ -202,6 +280,8 @@ def build_project(name, root_dir):
platform = distutils.util.get_platform()
architecture = "lib.%s-%i.%i" % (platform,
sys.version_info[0], sys.version_info[1])
if is_debug_python():
architecture += "-pydebug"
if os.environ.get("PYBUILD_NAME") == name:
# we are in the debian packaging way
......@@ -215,176 +295,166 @@ def build_project(name, root_dir):
p = subprocess.Popen([sys.executable, "setup.py", "build"],
shell=False, cwd=root_dir)
logger.debug("subprocess ended with rc= %s", p.wait())
return home
from argparse import ArgumentParser
epilog = """
"""
parser = ArgumentParser(description='Run the tests.',
epilog=epilog)
parser.add_argument("-i", "--insource",
action="store_true", dest="insource", default=False,
help="Use the build source and not the installed version")
parser.add_argument("-c", "--coverage", dest="coverage",
action="store_true", default=False,
help=("Report code coverage" +
"(requires 'coverage' and 'lxml' module)"))
parser.add_argument("-m", "--memprofile", dest="memprofile",
action="store_true", default=False,
help="Report memory profiling")
parser.add_argument("-v", "--verbose", default=0,
action="count", dest="verbose",
help="Increase verbosity. Option -v prints additional " +
"INFO messages. Use -vv for full verbosity, " +
"including debug messages and test help strings.")
parser.add_argument("-x", "--no-gui", dest="gui", default=True,
action="store_false",
help="Disable the test of the graphical use interface")
parser.add_argument("-g", "--no-opengl", dest="opengl", default=True,
action="store_false",
help="Disable tests using OpenGL")
parser.add_argument("-o", "--no-opencl", dest="opencl", default=True,
action="store_false",
help="Disable the test of the OpenCL part")
parser.add_argument("-l", "--low-mem", dest="low_mem", default=False,
action="store_true",
help="Disable test with large memory consumption (>100Mbyte")
parser.add_argument("--qt-binding", dest="qt_binding", default=None,
help="Force using a Qt binding, from 'PyQt4', 'PyQt5', or 'PySide'")
parser.add_argument("--web", dest="web_log", default=False,
help="Force unit test to export his log to graylog'",
action="store_true")
default_test_name = PROJECT_NAME
parser.add_argument("test_name", nargs='*',
default=(default_test_name,),
help="test names to run (Default: %s)" % default_test_name)
options = parser.parse_args()
sys.argv = [sys.argv[0]]
test_verbosity = 1
if options.verbose == 1:
logging.root.setLevel(logging.INFO)
logger.info("Set log level: INFO")
test_verbosity = 2
elif options.verbose > 1:
logging.root.setLevel(logging.DEBUG)
logger.info("Set log level: DEBUG")
test_verbosity = 2
if options.coverage:
logger.info("Running test-coverage")
import coverage
try:
cov = coverage.Coverage(omit=["*test*", "*third_party*", "*/setup.py"])
except AttributeError:
cov = coverage.coverage(omit=["*test*", "*third_party*", "*/setup.py"])
cov.start()
previous_web_log_value = os.environ.get("ORANGE_WEB_LOG")
os.environ["ORANGE_WEB_LOG"] = str(options.web_log)
if options.web_log is False:
from tomwer.web import config
config.grayport_host = None
if options.qt_binding and options.gui:
binding = options.qt_binding.lower()
if binding == "pyqt4":
logger.info("Force using PyQt4")
import PyQt4 #noqa
elif binding == "pyqt5":
logger.info("Force using PyQt5")
import PyQt5 #noqa
elif binding == "pyside":
logger.info("Force using PySide")
import PySide #noqa
# else:
# raise ValueError("Qt binding '%s' is unknown" % options.qt_binding)
os.environ["_TOMWER_NO_GUI_UNIT_TESTS"] = str(not options.gui)
# Prevent importing from source directory
if (os.path.dirname(os.path.abspath(__file__)) ==
os.path.abspath(sys.path[0])):
removed_from_sys_path = sys.path.pop(0)
logger.info("Patched sys.path, removed: '%s'", removed_from_sys_path)
# import module
if not options.insource:
if os.path.isdir(home):
return home
alt_home = os.path.join(os.path.dirname(home), "lib")
if os.path.isdir(alt_home):
return alt_home
def import_project_module(project_name, project_dir):
"""Import project module, from the system of from the project directory"""
# Prevent importing from source directory
if (os.path.dirname(os.path.abspath(__file__)) == os.path.abspath(sys.path[0])):
removed_from_sys_path = sys.path.pop(0)
logger.info("Patched sys.path, removed: '%s'", removed_from_sys_path)
if "--installed" in sys.argv:
try:
module = importer(project_name)
except Exception:
logger.error("Cannot run tests on installed version: %s not installed or raising error.",
project_name)
raise
else: # Use built source
build_dir = build_project(project_name, project_dir)
if build_dir is None:
logging.error("Built project is not available !!! investigate")
sys.path.insert(0, build_dir)
logger.warning("Patched sys.path, added: '%s'", build_dir)
module = importer(project_name)
return module
def get_test_options(project_module):
"""Returns the test options if available, else None"""
module_name = project_module.__name__ + '.test.utils'
logger.info('Import %s', module_name)
try:
module = importer(PROJECT_NAME)
except:
logger.warning(
"%s missing, using built (i.e. not installed) version",
PROJECT_NAME)
options.insource = True
if options.insource:
build_dir = build_project(PROJECT_NAME, PROJECT_DIR)
sys.path.insert(0, build_dir)
logger.warning("Patched sys.path, added: '%s'", build_dir)
module = importer(PROJECT_NAME)
PROJECT_VERSION = getattr(module, 'version', '')
PROJECT_PATH = module.__path__[0]
test_utils = importer(module_name)
except ImportError:
logger.warning("No module named '%s'. No test options available.", module_name)
return None
test_options = getattr(test_utils, "test_options", None)
return test_options
if __name__ == "__main__": # Needed for multiprocessing support on Windows
PROJECT_DIR = os.path.dirname(os.path.abspath(__file__))
PROJECT_NAME = get_project_name(PROJECT_DIR)
logger.info("Project name: %s", PROJECT_NAME)
project_module = import_project_module(PROJECT_NAME, PROJECT_DIR)
PROJECT_VERSION = getattr(project_module, 'version', '')
PROJECT_PATH = project_module.__path__[0]
test_options = get_test_options(project_module)
"""Contains extra configuration for the tests."""
parser = ArgumentParser(description='Run the tests.')
parser.add_argument("--installed",
action="store_true", dest="installed", default=False,
help=("Test the installed version instead of" +
"building from the source"))
parser.add_argument("-v", "--verbose", default=0,
action="count", dest="verbose",
help="Increase verbosity. Option -v prints additional " +
"INFO messages. Use -vv for full verbosity, " +
"including debug messages and test help strings.")
parser.add_argument("--qt-binding", dest="qt_binding", default=None,
help="Force using a Qt binding, from 'PyQt4', 'PyQt5', or 'PySide'")
if test_options is not None:
test_options.add_parser_argument(parser)
default_test_name = "%s.test.suite" % PROJECT_NAME
parser.add_argument("test_name", nargs='*',
default=(default_test_name,),
help="Test names to run (Default: %s)" % default_test_name)
options = parser.parse_args()
sys.argv = [sys.argv[0]]
test_verbosity = 1
use_buffer = True
if options.verbose == 1:
logging.root.setLevel(logging.INFO)
logger.info("Set log level: INFO")
test_verbosity = 2
use_buffer = False
elif options.verbose > 1:
logging.root.setLevel(logging.DEBUG)
logger.info("Set log level: DEBUG")
test_verbosity = 2
use_buffer = False
if options.qt_binding:
binding = options.qt_binding.lower()
if binding == "pyqt4":
logger.info("Force using PyQt4")
if sys.version < "3.0.0":
try:
import sip
sip.setapi("QString", 2)
sip.setapi("QVariant", 2)
except Exception:
logger.warning("Cannot set sip API")
import PyQt4.QtCore # noqa
elif binding == "pyqt5":
logger.info("Force using PyQt5")
import PyQt5.QtCore # noqa
elif binding == "pyside":
logger.info("Force using PySide")
import PySide.QtCore # noqa
elif binding == "pyside2":
logger.info("Force using PySide2")
import PySide2.QtCore # noqa
else:
raise ValueError("Qt binding '%s' is unknown" % options.qt_binding)
# Run the tests
runnerArgs = {}
runnerArgs["verbosity"] = test_verbosity
if options.memprofile:
runnerArgs["resultclass"] = ProfileTextTestResult
runner = unittest.TextTestRunner(**runnerArgs)
# Run the tests
runnerArgs = {}
runnerArgs["verbosity"] = test_verbosity
runnerArgs["buffer"] = use_buffer
runner = unittest.TextTestRunner(**runnerArgs)
logger.warning("test %s %s from %s",
PROJECT_NAME, PROJECT_VERSION, PROJECT_PATH)
logger.warning("Test %s %s from %s",
PROJECT_NAME, PROJECT_VERSION, PROJECT_PATH)
test_module_name = PROJECT_NAME + '.test'
logger.info('Import %s', test_module_name)
test_module = importer(test_module_name)
test_module_name = PROJECT_NAME + '.test'
logger.info('Import %s', test_module_name)
test_module = importer(test_module_name)
test_suite = unittest.TestSuite()
test_suite = unittest.TestSuite()
if test_options is not None:
# Configure the test options according to the command lines and the the environment
test_options.configure(options)
else:
logger.warning("No test options available.")
if options.test_name[0] == PROJECT_NAME:
main_test_suite = "%s.test.suite" % PROJECT_NAME
project_test_suite = getattr(test_module, 'suite')
test_suite.addTest(project_test_suite())
if options.gui is True:
from orangecontrib.xas.test import suite as orangecontrib_suite
test_suite.addTest(orangecontrib_suite())
else:
test_suite.addTest(
unittest.defaultTestLoader.loadTestsFromNames(options.test_name))
result = runner.run(test_suite)
for test, reason in result.skipped:
logger.warning('Skipped %s (%s): %s',
test.id(), test.shortDescription() or '', reason)
if not options.test_name:
# Do not use test loader to avoid cryptic exception
# when an error occur during import
project_test_suite = getattr(test_module, 'suite')
test_suite.addTest(project_test_suite())
else:
test_suite.addTest(
unittest.defaultTestLoader.loadTestsFromNames(options.test_name))
if result.wasSuccessful():
logger.info("test suite succeeded")
exit_status = 0
else:
logger.warning("test suite failed")
exit_status = 1
# Display the result when using CTRL-C
unittest.installHandler()
result = runner.run(test_suite)
if options.coverage:
cov.stop()
cov.save()
with open("coverage.rst", "w") as fn:
fn.write(report_rst(cov, PROJECT_NAME, PROJECT_VERSION, PROJECT_PATH))
if result.wasSuccessful():
exit_status = 0
else:
exit_status = 1
# reset the if ORANGE_WEB_LOG value
if previous_web_log_value is None:
del os.environ["ORANGE_WEB_LOG"]
else:
os.environ["ORANGE_WEB_LOG"] = previous_web_log_value
del os.environ["_TOMWER_NO_GUI_UNIT_TESTS"]
sys.exit(exit_status)
sys.exit(exit_status)
......@@ -90,9 +90,9 @@ def get_version():
def get_readme():
"""Returns content of README.rst file"""
"""Returns content of README.md file"""
dirname = os.path.dirname(os.path.abspath(__file__))
filename = os.path.join(dirname, "README.rst")
filename = os.path.join(dirname, "README.md")
with io.open(filename, "r", encoding="utf-8") as fp:
long_description = fp.read()
return long_description
......
# coding: utf-8
# /*##########################################################################
#
# Copyright (c) 2015-2017 European Synchrotron Radiation Facility
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
# ###########################################################################*/
from __future__ import absolute_import, print_function, division
__authors__ = ["Jérôme Kieffer"]
__license__ = "MIT"
__date__ = "23/05/2016"
import os as _os
import logging as _logging
# Attach a do nothing logging handler for silx
_logging.getLogger(__name__).addHandler(_logging.NullHandler())
project = _os.path.basename(_os.path.dirname(_os.path.abspath(__file__)))
try:
from ._version import __date__ as date # noqa
from ._version import version, version_info, hexversion, strictversion # noqa
except ImportError:
pass
# not raised for now because fail with pip install -e option. Something to
# look at.
# raise RuntimeError("Do NOT use %s from its sources: build it and use the built version" % project)
# coding: utf-8
# /*##########################################################################
#
# Copyright (c) 2016-2017 European Synchrotron Radiation Facility
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
# ###########################################################################*/
__authors__ = ["H. Payno"]
__license__ = "MIT"
__date__ = "06/11/2019"
from PyMca5.PyMcaPhysics.xas.XASClass import XASClass
from PyMca5.PyMcaPhysics.xas.XASClass import e2k
import logging