Browse Source

make test self-sufficient

komu 9 years ago
parent
commit
1b2e6d61c9
1 changed files with 21 additions and 0 deletions
  1. 21 0
      celery/tests/worker/test_autoscale.py

+ 21 - 0
celery/tests/worker/test_autoscale.py

@@ -192,3 +192,24 @@ class test_Autoscaler(AppCase):
             sys.stderr = p
         _exit.assert_called_with(1)
         self.assertTrue(stderr.write.call_count)
+
+    def test_no_negative_scale(self):
+        total_num_processes = []
+        worker = Mock(name='worker')
+        x = autoscale.Autoscaler(self.pool, 10, 3, worker=worker)
+        x.body() #the body func scales up or down
+
+        for i in range(35):
+            state.reserved_requests.add(i)
+            x.body()
+            total_num_processes.append(self.pool.num_processes)
+
+        for i in range(35):
+            state.reserved_requests.remove(i)
+            x.body()
+            total_num_processes.append(self.pool.num_processes)
+
+        self. assertTrue(all(i <= x.min_concurrency for i in total_num_processes)
+                        )
+        self. assertTrue(all(i <= x.max_concurrency for i in total_num_processes)
+                        )