diff options
Diffstat (limited to 'silx/utils/test/test_retry.py')
-rw-r--r-- | silx/utils/test/test_retry.py | 179 |
1 files changed, 0 insertions, 179 deletions
diff --git a/silx/utils/test/test_retry.py b/silx/utils/test/test_retry.py deleted file mode 100644 index d223f44..0000000 --- a/silx/utils/test/test_retry.py +++ /dev/null @@ -1,179 +0,0 @@ -# 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. -# -# ############################################################################*/ -"""Tests for retry utilities""" - -__authors__ = ["W. de Nolf"] -__license__ = "MIT" -__date__ = "05/02/2020" - - -import unittest -import os -import sys -import time -import tempfile - -from .. import retry - - -def _cause_segfault(): - import ctypes - - i = ctypes.c_char(b"a") - j = ctypes.pointer(i) - c = 0 - while True: - j[c] = b"a" - c += 1 - - -def _submain(filename, kwcheck=None, ncausefailure=0, faildelay=0): - assert filename - assert kwcheck - sys.stderr = open(os.devnull, "w") - - with open(filename, mode="r") as f: - failcounter = int(f.readline().strip()) - - if failcounter < ncausefailure: - time.sleep(faildelay) - failcounter += 1 - with open(filename, mode="w") as f: - f.write(str(failcounter)) - if failcounter % 2: - raise retry.RetryError - else: - _cause_segfault() - return True - - -_wsubmain = retry.retry_in_subprocess()(_submain) - - -class TestRetry(unittest.TestCase): - def setUp(self): - self.test_dir = tempfile.mkdtemp() - self.ctr_file = os.path.join(self.test_dir, "failcounter.txt") - - def tearDown(self): - if os.path.exists(self.ctr_file): - os.unlink(self.ctr_file) - os.rmdir(self.test_dir) - - def test_retry(self): - ncausefailure = 3 - faildelay = 0.1 - sufficient_timeout = ncausefailure * (faildelay + 10) - insufficient_timeout = ncausefailure * faildelay * 0.5 - - @retry.retry() - def method(check, kwcheck=None): - assert check - assert kwcheck - nonlocal failcounter - if failcounter < ncausefailure: - time.sleep(faildelay) - failcounter += 1 - raise retry.RetryError - return True - - failcounter = 0 - kw = { - "kwcheck": True, - "retry_timeout": sufficient_timeout, - } - self.assertTrue(method(True, **kw)) - - failcounter = 0 - kw = { - "kwcheck": True, - "retry_timeout": insufficient_timeout, - } - with self.assertRaises(retry.RetryTimeoutError): - method(True, **kw) - - def test_retry_contextmanager(self): - ncausefailure = 3 - faildelay = 0.1 - sufficient_timeout = ncausefailure * (faildelay + 10) - insufficient_timeout = ncausefailure * faildelay * 0.5 - - @retry.retry_contextmanager() - def context(check, kwcheck=None): - assert check - assert kwcheck - nonlocal failcounter - if failcounter < ncausefailure: - time.sleep(faildelay) - failcounter += 1 - raise retry.RetryError - yield True - - failcounter = 0 - kw = {"kwcheck": True, "retry_timeout": sufficient_timeout} - with context(True, **kw) as result: - self.assertTrue(result) - - failcounter = 0 - kw = {"kwcheck": True, "retry_timeout": insufficient_timeout} - with self.assertRaises(retry.RetryTimeoutError): - with context(True, **kw) as result: - pass - - def test_retry_in_subprocess(self): - ncausefailure = 3 - faildelay = 0.1 - sufficient_timeout = ncausefailure * (faildelay + 10) - insufficient_timeout = ncausefailure * faildelay * 0.5 - - kw = { - "ncausefailure": ncausefailure, - "faildelay": faildelay, - "kwcheck": True, - "retry_timeout": sufficient_timeout, - } - with open(self.ctr_file, mode="w") as f: - f.write("0") - self.assertTrue(_wsubmain(self.ctr_file, **kw)) - - kw = { - "ncausefailure": ncausefailure, - "faildelay": faildelay, - "kwcheck": True, - "retry_timeout": insufficient_timeout, - } - with open(self.ctr_file, mode="w") as f: - f.write("0") - with self.assertRaises(retry.RetryTimeoutError): - _wsubmain(self.ctr_file, **kw) - - -def suite(): - test_suite = unittest.TestSuite() - test_suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(TestRetry)) - return test_suite - - -if __name__ == "__main__": - unittest.main(defaultTest="suite") |