Tests: changed process counting in test_python_procman.py.

test/test_python_procman.py

@@ -1,5 +1,6 @@
-import os
+import re
 import time
+import subprocess
 import unittest
 import unit
 
@@ -11,192 +12,144 @@ class TestUnitProcman(unit.TestUnitControl):
         u.check_modules('python')
         u.check_version('0.5')
 
-    def count_processes(self):
+    def pids_for_process(self, process=None):
-        n = 0
+        if process is None:
-        for f in os.listdir(self.testdir):
+            process = self.app_name
-            if f.startswith('proctest.'):
-                n += 1
 
-        return n
+        time.sleep(0.2)
 
-    def app_code(self):
+        output = subprocess.check_output(['ps', 'ax'])
-        return """
-import atexit
-import os
 
-fname = "%s.%%d" %% os.getpid()
+        pids = set()
+        for m in re.findall('.*' + process, output.decode()):
+            pids.add(re.search('^\s*(\d+)', m).group(1))
 
-def remove_file():
+        return pids
-    os.remove(fname)
 
-atexit.register(remove_file)
+    def setUp(self):
+        super().setUp()
 
-open(fname, 'w')
+        code, name = """
 
 def application(env, start_response):
     start_response('200', [('Content-Length', '0')])
     return []
 
-""" % (self.testdir + '/proctest')
+""", 'py_app'
 
-
+        self.app_name = "app-" + self.testdir.split('/')[-1]
-    def test_python_prefork(self):
-        code, name = self.app_code(), 'py_app'
 
         self.python_application(name, code)
 
         self.conf({
             "listeners": {
                 "*:7080": {
-                    "application": "app"
+                    "application": self.app_name
                 }
             },
             "applications": {
-                "app": {
+                self.app_name: {
                     "type": "python",
-                    "processes": 2,
+                    "processes": { "spare": 0 },
                     "path": self.testdir + '/' + name,
                     "module": "wsgi"
                 }
             }
         })
 
-        self.assertEqual(self.count_processes(), 2, 'prefork 2')
+    def test_python_prefork(self):
+        self.conf('2', '/applications/' + self.app_name + '/processes')
+
+        pids = self.pids_for_process()
+        self.assertEqual(len(pids), 2, 'prefork 2')
 
         self.get()
-        self.assertEqual(self.count_processes(), 2, 'prefork still 2')
+        self.assertSetEqual(self.pids_for_process(), pids, 'prefork still 2')
 
-        self.conf('4', '/applications/app/processes')
+        self.conf('4', '/applications/' + self.app_name + '/processes')
 
-        time.sleep(0.2)
+        pids = self.pids_for_process()
-        self.assertEqual(self.count_processes(), 4, 'prefork 4')
+        self.assertEqual(len(pids), 4, 'prefork 4')
 
         self.get()
-        self.assertEqual(self.count_processes(), 4, 'prefork still 4')
+        self.assertSetEqual(self.pids_for_process(), pids, 'prefork still 4')
 
         self.stop_all()
 
     def test_python_ondemand(self):
-        code, name = self.app_code(), 'py_app'
-
-        self.python_application(name, code)
-
         self.conf({
-            "listeners": {
+            "spare": 0,
-                "*:7080": {
+            "max": 8,
-                    "application": "app"
+            "idle_timeout": 1
-                }
+        }, '/applications/' + self.app_name + '/processes')
-            },
-            "applications": {
-                "app": {
-                    "type": "python",
-                    "processes": {
-                        "spare": 0,
-                        "max": 8,
-                        "idle_timeout": 2
-                    },
-                    "path": self.testdir + '/' + name,
-                    "module": "wsgi"
-                }
-            }
-        })
 
-        self.assertEqual(self.count_processes(), 0, 'on-demand 0')
+        self.assertEqual(len(self.pids_for_process()), 0, 'on-demand 0')
 
         self.get()
-        self.assertEqual(self.count_processes(), 1, 'on-demand 1')
+        pids = self.pids_for_process()
+        self.assertEqual(len(pids), 1, 'on-demand 1')
 
         self.get()
-        self.assertEqual(self.count_processes(), 1, 'on-demand still 1')
+        self.assertSetEqual(self.pids_for_process(), pids, 'on-demand still 1')
 
-        time.sleep(2.2)
+        time.sleep(1)
-        self.assertEqual(self.count_processes(), 0, 'on-demand stop idle')
+
+        self.assertEqual(len(self.pids_for_process()), 0, 'on-demand stop idle')
 
         self.stop_all()
 
     def test_python_scale_updown(self):
-        code, name = self.app_code(), 'py_app'
-
-        self.python_application(name, code)
-
         self.conf({
-            "listeners": {
+            "spare": 2,
-                "*:7080": {
+            "max": 8,
-                    "application": "app"
+            "idle_timeout": 1
-                }
+        }, '/applications/' + self.app_name + '/processes')
-            },
-            "applications": {
-                "app": {
-                    "type": "python",
-                    "processes": {
-                        "spare": 2,
-                        "max": 8,
-                        "idle_timeout": 2
-                    },
-                    "path": self.testdir + '/' + name,
-                    "module": "wsgi"
-                }
-            }
-        })
 
-        self.assertEqual(self.count_processes(), 2, 'updown idle 2')
+        pids = self.pids_for_process()
+        self.assertEqual(len(pids), 2, 'updown 2')
 
         self.get()
-        time.sleep(0.2)
+        pids_new = self.pids_for_process()
-        self.assertEqual(self.count_processes(), 3, 'updown idle 2, busy 1')
+        self.assertEqual(len(pids_new), 3, 'updown 3')
+        self.assertTrue(pids.issubset(pids_new), 'updown 3 only 1 new')
 
         self.get()
-        time.sleep(0.2)
+        self.assertSetEqual(self.pids_for_process(), pids_new, 'updown still 3')
-        self.assertEqual(self.count_processes(), 3, 'updown still 3')
 
-        time.sleep(2.2)
+        time.sleep(1)
-        self.assertEqual(self.count_processes(), 2, 'updown stop idle')
+
+        pids = self.pids_for_process()
+        self.assertEqual(len(pids), 2, 'updown stop idle')
 
         self.get()
-        time.sleep(0.2)
+        pids_new = self.pids_for_process()
-        self.assertEqual(self.count_processes(), 3, 'updown idle 2, busy 1')
+        self.assertEqual(len(pids_new), 3, 'updown again 3')
+        self.assertTrue(pids.issubset(pids_new), 'updown again 3 only 1 new')
 
         self.stop_all()
 
     def test_python_reconfigure(self):
-        code, name = self.app_code(), 'py_app'
-
-        self.python_application(name, code)
-
         self.conf({
-            "listeners": {
+            "spare": 2,
-                "*:7080": {
+            "max": 6,
-                    "application": "app"
+            "idle_timeout": 1
-                }
+        }, '/applications/' + self.app_name + '/processes')
-            },
-            "applications": {
-                "app": {
-                    "type": "python",
-                    "processes": {
-                        "spare": 2,
-                        "max": 6,
-                        "idle_timeout": 2
-                    },
-                    "path": self.testdir + '/' + name,
-                    "module": "wsgi"
-                }
-            }
-        })
 
-        self.assertEqual(self.count_processes(), 2, 'reconf idle 2')
+        pids = self.pids_for_process()
+        self.assertEqual(len(pids), 2, 'reconf 2')
 
         self.get()
-        time.sleep(0.2)
+        pids_new = self.pids_for_process()
-        self.assertEqual(self.count_processes(), 3, 'reconf idle 2, busy 1')
+        self.assertEqual(len(pids_new), 3, 'reconf 3')
+        self.assertTrue(pids.issubset(pids_new), 'reconf 3 only 1 new')
 
-        self.conf('6', '/applications/app/processes/spare')
+        self.conf('6', '/applications/' + self.app_name + '/processes/spare')
 
-        time.sleep(0.2)
+        pids = self.pids_for_process()
-        self.assertEqual(self.count_processes(), 6, 'reconf idle 6')
+        self.assertEqual(len(pids), 6, 'reconf 6')
 
         self.get()
-        time.sleep(0.2)
+        self.assertSetEqual(self.pids_for_process(), pids, 'reconf still 6')
-        self.assertEqual(self.count_processes(), 6, 'reconf still 6')
 
         self.stop_all()
 
@@ -206,8 +159,7 @@ def application(env, start_response):
             "applications": {}
         })
 
-        time.sleep(0.2)
+        self.assertEqual(len(self.pids_for_process()), 0, 'stop all')
-        self.assertEqual(self.count_processes(), 0, 'stop all')
 
 if __name__ == '__main__':
     unittest.main()