diff options
Diffstat (limited to 'silx/gui/widgets/test/test_threadpoolpushbutton.py')
-rw-r--r-- | silx/gui/widgets/test/test_threadpoolpushbutton.py | 135 |
1 files changed, 0 insertions, 135 deletions
diff --git a/silx/gui/widgets/test/test_threadpoolpushbutton.py b/silx/gui/widgets/test/test_threadpoolpushbutton.py deleted file mode 100644 index e92eb02..0000000 --- a/silx/gui/widgets/test/test_threadpoolpushbutton.py +++ /dev/null @@ -1,135 +0,0 @@ -# coding: utf-8 -# /*########################################################################## -# -# Copyright (c) 2016 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. -# -# ###########################################################################*/ -"""Test for silx.gui.hdf5 module""" - -__authors__ = ["V. Valls"] -__license__ = "MIT" -__date__ = "17/01/2018" - - -import unittest -import time -from silx.gui import qt -from silx.gui.utils.testutils import TestCaseQt -from silx.gui.utils.testutils import SignalListener -from silx.gui.widgets.ThreadPoolPushButton import ThreadPoolPushButton -from silx.utils.testutils import TestLogging - - -class TestThreadPoolPushButton(TestCaseQt): - - def setUp(self): - super(TestThreadPoolPushButton, self).setUp() - self._result = [] - - def waitForPendingOperations(self, object): - for i in range(50): - if not object.hasPendingOperations(): - break - self.qWait(10) - else: - raise RuntimeError("Still waiting for a pending operation") - - def _trace(self, name, delay=0): - self._result.append(name) - if delay != 0: - time.sleep(delay / 1000.0) - - def _compute(self): - return "result" - - def _computeFail(self): - raise Exception("exception") - - def testExecute(self): - button = ThreadPoolPushButton() - button.setCallable(self._trace, "a", 0) - button.executeCallable() - time.sleep(0.1) - self.assertListEqual(self._result, ["a"]) - self.waitForPendingOperations(button) - - def testMultiExecution(self): - button = ThreadPoolPushButton() - button.setCallable(self._trace, "a", 0) - number = qt.silxGlobalThreadPool().maxThreadCount() - for _ in range(number): - button.executeCallable() - self.waitForPendingOperations(button) - self.assertListEqual(self._result, ["a"] * number) - - def testSaturateThreadPool(self): - button = ThreadPoolPushButton() - button.setCallable(self._trace, "a", 100) - number = qt.silxGlobalThreadPool().maxThreadCount() * 2 - for _ in range(number): - button.executeCallable() - self.waitForPendingOperations(button) - self.assertListEqual(self._result, ["a"] * number) - - def testSuccess(self): - listener = SignalListener() - button = ThreadPoolPushButton() - button.setCallable(self._compute) - button.beforeExecuting.connect(listener.partial(test="be")) - button.started.connect(listener.partial(test="s")) - button.succeeded.connect(listener.partial(test="result")) - button.failed.connect(listener.partial(test="Unexpected exception")) - button.finished.connect(listener.partial(test="f")) - button.executeCallable() - self.qapp.processEvents() - time.sleep(0.1) - self.qapp.processEvents() - result = listener.karguments(argumentName="test") - self.assertListEqual(result, ["be", "s", "result", "f"]) - - def testFail(self): - listener = SignalListener() - button = ThreadPoolPushButton() - button.setCallable(self._computeFail) - button.beforeExecuting.connect(listener.partial(test="be")) - button.started.connect(listener.partial(test="s")) - button.succeeded.connect(listener.partial(test="Unexpected success")) - button.failed.connect(listener.partial(test="exception")) - button.finished.connect(listener.partial(test="f")) - with TestLogging('silx.gui.widgets.ThreadPoolPushButton', error=1): - button.executeCallable() - self.qapp.processEvents() - time.sleep(0.1) - self.qapp.processEvents() - result = listener.karguments(argumentName="test") - self.assertListEqual(result, ["be", "s", "exception", "f"]) - listener.clear() - - -def suite(): - test_suite = unittest.TestSuite() - test_suite.addTest( - unittest.defaultTestLoader.loadTestsFromTestCase(TestThreadPoolPushButton)) - return test_suite - - -if __name__ == '__main__': - unittest.main(defaultTest='suite') |