5from unit.applications.lang.python import TestApplicationPython 6 7 8class TestRespawn(TestApplicationPython): 9 prerequisites = {'modules': {'python': 'any'}} 10 11 PATTERN_ROUTER = 'unit: router' 12 PATTERN_CONTROLLER = 'unit: controller' 13 14 def setUp(self): 15 super().setUp() 16 17 self.app_name = "app-" + self.testdir.split('/')[-1] 18 19 self.load('empty', self.app_name) 20 21 self.assertIn( 22 'success', 23 self.conf('1', 'applications/' + self.app_name + '/processes') 24 ) 25 26 def pid_by_name(self, name): 27 output = subprocess.check_output(['ps', 'ax']).decode() 28 m = re.search('\s*(\d+).*' + name, output) 29 return m if m is None else m.group(1) 30 31 def kill_pids(self, *pids): 32 subprocess.call(['kill', '-9'] + list(pids)) 33 34 def wait_for_process(self, process): 35 for i in range(50): 36 found = self.pid_by_name(process) 37 38 if found is not None: 39 break 40 41 time.sleep(0.1) 42 43 return found 44 45 def smoke_test(self): 46 for _ in range(5): 47 self.assertIn( 48 'success', 49 self.conf('1', 'applications/' + self.app_name + '/processes') 50 ) 51 self.assertEqual(self.get()['status'], 200) 52 53 # Check if the only one router, controller, 54 # and application processes running. 55 56 output = subprocess.check_output(['ps', 'ax']).decode() 57 self.assertEqual(len(re.findall(self.PATTERN_ROUTER, output)), 1) 58 self.assertEqual(len(re.findall(self.PATTERN_CONTROLLER, output)), 1) 59 self.assertEqual(len(re.findall(self.app_name, output)), 1) 60 61 def test_respawn_router(self): 62 pid = self.pid_by_name(self.PATTERN_ROUTER) 63 64 self.kill_pids(pid) 65 self.skip_alerts.append(r'process %s exited on signal 9' % pid) 66 67 self.assertIsNotNone(self.wait_for_process(self.PATTERN_ROUTER)) 68 69 self.smoke_test() 70 71 def test_respawn_controller(self): 72 pid = self.pid_by_name(self.PATTERN_CONTROLLER) 73 74 self.kill_pids(pid) 75 self.skip_alerts.append(r'process %s exited on signal 9' % pid) 76 77 self.assertIsNotNone(self.wait_for_process(self.PATTERN_CONTROLLER)) 78 79 self.assertEqual(self.get()['status'], 200) 80 81 self.smoke_test() 82 83 def test_respawn_application(self): 84 pid = self.pid_by_name(self.app_name) 85 86 self.kill_pids(pid) 87 self.skip_alerts.append(r'process %s exited on signal 9' % pid) 88 89 self.assertIsNotNone(self.wait_for_process(self.app_name)) 90 91 self.smoke_test() 92 93 94if __name__ == '__main__': 95 TestRespawn.main()
| 5from unit.applications.lang.python import TestApplicationPython 6 7 8class TestRespawn(TestApplicationPython): 9 prerequisites = {'modules': {'python': 'any'}} 10 11 PATTERN_ROUTER = 'unit: router' 12 PATTERN_CONTROLLER = 'unit: controller' 13 14 def setUp(self): 15 super().setUp() 16 17 self.app_name = "app-" + self.testdir.split('/')[-1] 18 19 self.load('empty', self.app_name) 20 21 self.assertIn( 22 'success', 23 self.conf('1', 'applications/' + self.app_name + '/processes') 24 ) 25 26 def pid_by_name(self, name): 27 output = subprocess.check_output(['ps', 'ax']).decode() 28 m = re.search('\s*(\d+).*' + name, output) 29 return m if m is None else m.group(1) 30 31 def kill_pids(self, *pids): 32 subprocess.call(['kill', '-9'] + list(pids)) 33 34 def wait_for_process(self, process): 35 for i in range(50): 36 found = self.pid_by_name(process) 37 38 if found is not None: 39 break 40 41 time.sleep(0.1) 42 43 return found 44 45 def smoke_test(self): 46 for _ in range(5): 47 self.assertIn( 48 'success', 49 self.conf('1', 'applications/' + self.app_name + '/processes') 50 ) 51 self.assertEqual(self.get()['status'], 200) 52 53 # Check if the only one router, controller, 54 # and application processes running. 55 56 output = subprocess.check_output(['ps', 'ax']).decode() 57 self.assertEqual(len(re.findall(self.PATTERN_ROUTER, output)), 1) 58 self.assertEqual(len(re.findall(self.PATTERN_CONTROLLER, output)), 1) 59 self.assertEqual(len(re.findall(self.app_name, output)), 1) 60 61 def test_respawn_router(self): 62 pid = self.pid_by_name(self.PATTERN_ROUTER) 63 64 self.kill_pids(pid) 65 self.skip_alerts.append(r'process %s exited on signal 9' % pid) 66 67 self.assertIsNotNone(self.wait_for_process(self.PATTERN_ROUTER)) 68 69 self.smoke_test() 70 71 def test_respawn_controller(self): 72 pid = self.pid_by_name(self.PATTERN_CONTROLLER) 73 74 self.kill_pids(pid) 75 self.skip_alerts.append(r'process %s exited on signal 9' % pid) 76 77 self.assertIsNotNone(self.wait_for_process(self.PATTERN_CONTROLLER)) 78 79 self.assertEqual(self.get()['status'], 200) 80 81 self.smoke_test() 82 83 def test_respawn_application(self): 84 pid = self.pid_by_name(self.app_name) 85 86 self.kill_pids(pid) 87 self.skip_alerts.append(r'process %s exited on signal 9' % pid) 88 89 self.assertIsNotNone(self.wait_for_process(self.app_name)) 90 91 self.smoke_test() 92 93 94if __name__ == '__main__': 95 TestRespawn.main()
|