1import re 2import subprocess 3import time 4 5from conftest import skip_alert 6from unit.applications.lang.python import TestApplicationPython 7 8 9class TestRespawn(TestApplicationPython): 10 prerequisites = {'modules': {'python': 'any'}} 11 12 PATTERN_ROUTER = 'unit: router' 13 PATTERN_CONTROLLER = 'unit: controller' 14 15 def setup_method(self): 16 super().setup_method() 17 18 self.app_name = "app-" + self.temp_dir.split('/')[-1] 19 20 self.load('empty', self.app_name) 21 22 assert 'success' in self.conf( 23 '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(r'\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 assert 'success' in self.conf( 48 '1', 'applications/' + self.app_name + '/processes' 49 ) 50 assert self.get()['status'] == 200 51 52 # Check if the only one router, controller, 53 # and application processes running. 54 55 output = subprocess.check_output(['ps', 'ax']).decode() 56 assert len(re.findall(self.PATTERN_ROUTER, output)) == 1 57 assert len(re.findall(self.PATTERN_CONTROLLER, output)) == 1 58 assert len(re.findall(self.app_name, output)) == 1 59 60 def test_respawn_router(self): 61 pid = self.pid_by_name(self.PATTERN_ROUTER) 62 63 self.kill_pids(pid) 64 skip_alert(r'process %s exited on signal 9' % pid) 65 66 assert self.wait_for_process(self.PATTERN_ROUTER) is not None 67 68 self.smoke_test() 69 70 def test_respawn_controller(self): 71 pid = self.pid_by_name(self.PATTERN_CONTROLLER) 72 73 self.kill_pids(pid) 74 skip_alert(r'process %s exited on signal 9' % pid) 75 76 assert self.wait_for_process(self.PATTERN_CONTROLLER) is not None 77 78 assert self.get()['status'] == 200 79 80 self.smoke_test() 81 82 def test_respawn_application(self): 83 pid = self.pid_by_name(self.app_name) 84 85 self.kill_pids(pid) 86 skip_alert(r'process %s exited on signal 9' % pid) 87 88 assert self.wait_for_process(self.app_name) is not None 89 90 self.smoke_test() 91