From dd6b678cd855490960a8d78e480de85f2b9ae158 Mon Sep 17 00:00:00 2001 From: Christian Herdtweck Date: Wed, 13 Jan 2016 18:09:24 +0100 Subject: [PATCH] start testing the DiscCheckerThread, not as easy as expected... --- test/test_helper_unittest.py | 31 ++++++++++++++++++++++++++++++- 1 files changed, 30 insertions(+), 1 deletions(-) diff --git a/test/test_helper_unittest.py b/test/test_helper_unittest.py index cf2fc37..0040224 100644 --- a/test/test_helper_unittest.py +++ b/test/test_helper_unittest.py @@ -15,6 +15,8 @@ import unittest from datetime import timedelta, datetime as dt from file_helpers import FilesystemFillState, get_filesystem_fill_states import warnings +from time import sleep +import threading import test_helpers @@ -161,6 +163,9 @@ class TestHelperTester(unittest.TestCase): def test_thread(self): """ test the DiscFillCheckerThread """ + #threading.setprofile(self.threading_profile_func) + # somehow damages df output ... need to check! + # since we do not want to fill up the disc, we incrementally decrease # the tolerance so even our hopefully half-empty filesystems will # trigger warnings / errors @@ -168,13 +173,37 @@ class TestHelperTester(unittest.TestCase): min_free = min(fill_state.available for fill_state in actual_states) print(min_free) - sleep_time = 0 + sleep_time_checker = 0 + sleep_time_tester = 1 + time_tolerance = 0.1 # run something with default settings. Should give no trouble + decision_function = test_helpers.default_disc_full_decision_function + print('before: threads are {0}' + .format([thread.name for thread in threading.enumerate()])) + start_time = dt.now() + with test_helpers.disc_fill_checked(sleep_time_checker, + decision_function): + print('within: threads are {0}' + .format([thread.name for thread in threading.enumerate()])) + self.nonsense_function(sleep_time_tester) + time_taken = (dt.now() - start_time).total_seconds() + print('after: threads are {0}' + .format([thread.name for thread in threading.enumerate()])) + self.assertTrue(time_taken >= sleep_time_tester) + self.assertTrue(time_taken < sleep_time_tester + time_tolerance) # run with thresholds set so we get warnings # run with thresholds set so we get errors + def nonsense_function(self, sleep_time): + """ function to run while checking disc fill state; just sleeps """ + sleep(sleep_time) + + def threading_profile_func(self, *args, **kwargs): + print('profiling with args {0} and kwargs {1}' + .format([arg for arg in args if len(args) < 20], kwargs)) + if __name__ == '__main__': unittest.main() -- 1.7.1