From: Christian Herdtweck <christian.herdtweck@intra2net.com>
Date: Tue, 12 Jan 2016 12:32:29 +0000 (+0100)
Subject: created test_helpers for simplifying quick test development; started disc usage checker
X-Git-Tag: v1.2~78
X-Git-Url: http://developer.intra2net.com/git/?a=commitdiff_plain;h=cf4e57fdbbd41711401db3c814ae3972db58c8d0;p=pyi2ncommon

created test_helpers for simplifying quick test development; started disc usage checker
---

diff --git a/test_helpers.py b/test_helpers.py
new file mode 100644
index 0000000..42eff5c
--- /dev/null
+++ b/test_helpers.py
@@ -0,0 +1,70 @@
+""" Helpers for developping quick test scripts
+
+Creation motivated by fear of filling disc space during long-running stress
+tests
+
+.. codeauthor:: Christian Herdtweck, christian.herdtweck@intra2net.com
+"""
+
+import contextlib
+from threading import Thread
+from time import sleep
+from file_helpers import get_disc_stats
+from datetime import datetime as dt
+from buffers import LogarithmicBuffer
+
+class DiscFillCheckerThread(Thread):
+    """ a thread that checks disc fill in regular intervals """
+
+    def __init__(self, interval=10, warn_func=None, err_func=None):
+        super(DiscFillCheckerThread, self).__init__(name='discFillChck',
+                                                    daemon=True)
+        # set variables
+        self.interval = interval
+        if warn_func is None:
+            self.warn_func = disc_fill_warn_func
+        else:
+            # need to check warn_func somehow (using introspection?)
+            self.warn_func = warn_func
+        if err_func is None:
+            self.err_func = disc_fill_err_func
+        else:
+            # need to check err_func somehow (using introspection?)
+            self.err_func = err_func
+
+        # remember relevant disc stats at start
+        bufs = dict((stat.name, LogarithmicBuffer(5)) for stat in stats)
+        now = dt.now()
+        for stat in stats:
+            bufs[stat.name].add((now, stat.available))
+
+    def run(self):
+        while True:
+            sleep(self.interval)
+            stats = get_disc_stats()
+            now = dt.now()
+            for stat in stats:
+                buf = None
+                try:
+                    buf = bufs[stat.name]
+                except KeyError:
+                    # new file system -- need to create a new buffer
+                    buf = LogarithmicBuffer(5)
+
+                buf.add((now, stat.available))
+
+                raise NotImplementedError('estimate time until disc full')
+                raise NotImplementedError('warn or err if too soon')
+
+
+@contextlib.contextmanager
+def disc_fill_checked(path):
+    """ run test function while separate thread watches disc space """
+
+    # todo: add args and forward
+    DiscFillCheckerThread().start()
+
+    try:
+        yield
+    finally:
+        pass # wait for other to stop or so? or stop it?