Changeset 12581 for main/waeup.kofa/trunk/src/waeup/kofa/browser

Timestamp:

10 Feb 2015, 14:13:17 (10 years ago)

Author:

uli

Message:

Block jobs if system load is too high.

File:

• main/waeup.kofa/trunk/src/waeup/kofa/browser/tests/test_async.py (modified) (4 diffs)

main/waeup.kofa/trunk/src/waeup/kofa/browser/tests/test_async.py

@@ -3 +3 @@
 import tempfile
 import transaction
+from contextlib import contextmanager
 from zc.async.testing import wait_for_result
 from zope.component import getUtility
@@ -10 +11 @@
 from waeup.kofa.app import University
 from waeup.kofa.async import AsyncJob, get_job_id
-from waeup.kofa.interfaces import IJobManager
+from waeup.kofa.interfaces import IJobManager, IKofaUtils
 from waeup.kofa.testing import FunctionalTestCase, FunctionalLayer
 from waeup.kofa.tests.test_async import FunctionalAsyncTestCase
 
+
 def fake_get_export_jobs_status(user_id=None):
     return [('completed', 'Completed', 'my_exporter'),]
 
+
 def dummy_func():
     return 42
+
+
+@contextmanager
+def expensive_actions():
+    """A contextmanager to work with a modified `expensive_acrtions_allowed()`
+
+    This method is part of local IKofaUtils.
+    """
+    utils = getUtility(IKofaUtils)
+    old_func = utils.expensive_actions_allowed
+    try:
+        yield utils
+    finally:
+        # restore function after use
+        utils.expensive_actions_allowed = old_func
+
 
 class AsyncBrowserTests(FunctionalAsyncTestCase):
@@ -92 +111 @@
         return
 
+
 class JobOverviewTests(AsyncBrowserTests):
     # Tests for the job overview page of universities
@@ -106 +126 @@
         manager.put(job, site=self.getRootFolder())
         return job
+
+    def test_job_high_load(self):
+        # under high load, jobs won't start (None is returned then)
+        manager = getUtility(IJobManager)
+        job = AsyncJob(dummy_func)
+        site = self.getRootFolder()
+        with expensive_actions() as utils:
+            utils.expensive_actions_allowed = lambda : False
+            result = manager.put(job, site=self.getRootFolder())
+        self.assertTrue(result is None)
 
     def test_job_get(self):