c183bd8749
Class usage came from the unittest framework and it was always redundant after migration to the pytest. This commit removes classes from files containing tests to make them more readable and understandable.
108 lines
2.6 KiB
Python
108 lines
2.6 KiB
Python
import re
|
|
import subprocess
|
|
import time
|
|
|
|
import pytest
|
|
from unit.applications.lang.python import ApplicationPython
|
|
|
|
prerequisites = {'modules': {'python': 'any'}}
|
|
|
|
client = ApplicationPython()
|
|
|
|
PATTERN_ROUTER = 'unit: router'
|
|
PATTERN_CONTROLLER = 'unit: controller'
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_method_fixture(temp_dir):
|
|
client.app_name = f'app-{temp_dir.split("/")[-1]}'
|
|
|
|
client.load('empty', client.app_name)
|
|
|
|
assert 'success' in client.conf(
|
|
'1', f'applications/{client.app_name}/processes'
|
|
)
|
|
|
|
|
|
def pid_by_name(name, ppid):
|
|
output = subprocess.check_output(['ps', 'ax', '-O', 'ppid']).decode()
|
|
m = re.search(fr'\s*(\d+)\s*{ppid}.*{name}', output)
|
|
return None if m is None else m.group(1)
|
|
|
|
|
|
def kill_pids(*pids):
|
|
subprocess.call(['kill', '-9', *pids])
|
|
|
|
|
|
def wait_for_process(process, unit_pid):
|
|
for _ in range(50):
|
|
found = pid_by_name(process, unit_pid)
|
|
|
|
if found is not None:
|
|
break
|
|
|
|
time.sleep(0.1)
|
|
|
|
return found
|
|
|
|
|
|
def find_proc(name, ppid, ps_output):
|
|
return re.findall(fr'{ppid}.*{name}', ps_output)
|
|
|
|
|
|
def smoke_test(unit_pid):
|
|
for _ in range(10):
|
|
r = client.conf('1', f'applications/{client.app_name}/processes')
|
|
|
|
if 'success' in r:
|
|
break
|
|
|
|
time.sleep(0.1)
|
|
|
|
assert 'success' in r
|
|
assert client.get()['status'] == 200
|
|
|
|
# Check if the only one router, controller,
|
|
# and application processes running.
|
|
|
|
out = subprocess.check_output(['ps', 'ax', '-O', 'ppid']).decode()
|
|
assert len(find_proc(PATTERN_ROUTER, unit_pid, out)) == 1
|
|
assert len(find_proc(PATTERN_CONTROLLER, unit_pid, out)) == 1
|
|
assert len(find_proc(client.app_name, unit_pid, out)) == 1
|
|
|
|
|
|
def test_respawn_router(skip_alert, unit_pid, skip_fds_check):
|
|
skip_fds_check(router=True)
|
|
pid = pid_by_name(PATTERN_ROUTER, unit_pid)
|
|
|
|
kill_pids(pid)
|
|
skip_alert(fr'process {pid} exited on signal 9')
|
|
|
|
assert wait_for_process(PATTERN_ROUTER, unit_pid) is not None
|
|
|
|
smoke_test(unit_pid)
|
|
|
|
|
|
def test_respawn_controller(skip_alert, unit_pid, skip_fds_check):
|
|
skip_fds_check(controller=True)
|
|
pid = pid_by_name(PATTERN_CONTROLLER, unit_pid)
|
|
|
|
kill_pids(pid)
|
|
skip_alert(fr'process {pid} exited on signal 9')
|
|
|
|
assert wait_for_process(PATTERN_CONTROLLER, unit_pid) is not None
|
|
|
|
assert client.get()['status'] == 200
|
|
|
|
smoke_test(unit_pid)
|
|
|
|
|
|
def test_respawn_application(skip_alert, unit_pid):
|
|
pid = pid_by_name(client.app_name, unit_pid)
|
|
|
|
kill_pids(pid)
|
|
skip_alert(fr'process {pid} exited on signal 9')
|
|
|
|
assert wait_for_process(client.app_name, unit_pid) is not None
|
|
|
|
smoke_test(unit_pid)
|