1import re 2import subprocess 3import time 4 5import pytest 6 7from unit.applications.lang.python import ApplicationPython 8 9prerequisites = {'modules': {'python': 'any'}} 10 11client = ApplicationPython() 12 13PATTERN_ROUTER = 'unit: router' 14PATTERN_CONTROLLER = 'unit: controller' 15 16 17@pytest.fixture(autouse=True) 18def setup_method_fixture(temp_dir): 19 client.app_name = f'app-{temp_dir.split("/")[-1]}' 20 21 client.load('empty', client.app_name) 22 23 assert 'success' in client.conf( 24 '1', f'applications/{client.app_name}/processes' 25 ) 26 27 28def pid_by_name(name, ppid): 29 output = subprocess.check_output(['ps', 'ax', '-O', 'ppid']).decode() 30 m = re.search(fr'\s*(\d+)\s*{ppid}.*{name}', output) 31 return None if m is None else m.group(1) 32 33 34def kill_pids(*pids): 35 subprocess.call(['kill', '-9', *pids]) 36 37 38def wait_for_process(process, unit_pid): 39 for _ in range(50): 40 found = pid_by_name(process, unit_pid) 41 42 if found is not None: 43 break 44 45 time.sleep(0.1) 46 47 return found 48 49 50def find_proc(name, ppid, ps_output): 51 return re.findall(fr'{ppid}.*{name}', ps_output) 52 53 54def smoke_test(unit_pid): 55 for _ in range(10): 56 r = client.conf('1', f'applications/{client.app_name}/processes') 57 58 if 'success' in r: 59 break 60 61 time.sleep(0.1) 62 63 assert 'success' in r 64 assert client.get()['status'] == 200 65 66 # Check if the only one router, controller, 67 # and application processes running. 68 69 out = subprocess.check_output(['ps', 'ax', '-O', 'ppid']).decode() 70 assert len(find_proc(PATTERN_ROUTER, unit_pid, out)) == 1 71 assert len(find_proc(PATTERN_CONTROLLER, unit_pid, out)) == 1 72 assert len(find_proc(client.app_name, unit_pid, out)) == 1 73 74 75def test_respawn_router(skip_alert, unit_pid, skip_fds_check): 76 skip_fds_check(router=True) 77 pid = pid_by_name(PATTERN_ROUTER, unit_pid) 78 79 kill_pids(pid) 80 skip_alert(fr'process {pid} exited on signal 9') 81 82 assert wait_for_process(PATTERN_ROUTER, unit_pid) is not None 83 84 smoke_test(unit_pid) 85 86 87def test_respawn_controller(skip_alert, unit_pid, skip_fds_check): 88 skip_fds_check(controller=True) 89 pid = pid_by_name(PATTERN_CONTROLLER, unit_pid) 90 91 kill_pids(pid) 92 skip_alert(fr'process {pid} exited on signal 9') 93 94 assert wait_for_process(PATTERN_CONTROLLER, unit_pid) is not None 95 96 assert client.get()['status'] == 200 97 98 smoke_test(unit_pid) 99 100 101def test_respawn_application(skip_alert, unit_pid): 102 pid = pid_by_name(client.app_name, unit_pid) 103 104 kill_pids(pid) 105 skip_alert(fr'process {pid} exited on signal 9') 106 107 assert wait_for_process(client.app_name, unit_pid) is not None 108 109 smoke_test(unit_pid) 110