diff options
author | Andrei Zeliankou <zelenkov@nginx.com> | 2023-06-14 18:20:09 +0100 |
---|---|---|
committer | Andrei Zeliankou <zelenkov@nginx.com> | 2023-06-14 18:20:09 +0100 |
commit | c183bd8749a19477390f8cb77efe5f6d223f0905 (patch) | |
tree | 4e821e9cb07be9a86bf2d442acb3ea6740ba5a99 /test/test_respawn.py | |
parent | c6d05191a069ac150cc8eb2bece75cf79c0a465a (diff) | |
download | unit-c183bd8749a19477390f8cb77efe5f6d223f0905.tar.gz unit-c183bd8749a19477390f8cb77efe5f6d223f0905.tar.bz2 |
Tests: get rid of classes in test files.
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.
Diffstat (limited to '')
-rw-r--r-- | test/test_respawn.py | 133 |
1 files changed, 70 insertions, 63 deletions
diff --git a/test/test_respawn.py b/test/test_respawn.py index 84bddbaa..dc465cda 100644 --- a/test/test_respawn.py +++ b/test/test_respawn.py @@ -3,99 +3,106 @@ import subprocess import time import pytest -from unit.applications.lang.python import TestApplicationPython +from unit.applications.lang.python import ApplicationPython prerequisites = {'modules': {'python': 'any'}} +client = ApplicationPython() -class TestRespawn(TestApplicationPython): - PATTERN_ROUTER = 'unit: router' - PATTERN_CONTROLLER = 'unit: controller' +PATTERN_ROUTER = 'unit: router' +PATTERN_CONTROLLER = 'unit: controller' - @pytest.fixture(autouse=True) - def setup_method_fixture(self, temp_dir): - self.app_name = f'app-{temp_dir.split("/")[-1]}' - self.load('empty', self.app_name) +@pytest.fixture(autouse=True) +def setup_method_fixture(temp_dir): + client.app_name = f'app-{temp_dir.split("/")[-1]}' - assert 'success' in self.conf( - '1', f'applications/{self.app_name}/processes' - ) + client.load('empty', client.app_name) - def pid_by_name(self, 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) + assert 'success' in client.conf( + '1', f'applications/{client.app_name}/processes' + ) - def kill_pids(self, *pids): - subprocess.call(['kill', '-9', *pids]) - def wait_for_process(self, process, unit_pid): - for _ in range(50): - found = self.pid_by_name(process, unit_pid) +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) - if found is not None: - break - time.sleep(0.1) +def kill_pids(*pids): + subprocess.call(['kill', '-9', *pids]) - return found - def find_proc(self, name, ppid, ps_output): - return re.findall(fr'{ppid}.*{name}', ps_output) +def wait_for_process(process, unit_pid): + for _ in range(50): + found = pid_by_name(process, unit_pid) - def smoke_test(self, unit_pid): - for _ in range(10): - r = self.conf('1', f'applications/{self.app_name}/processes') + if found is not None: + break - if 'success' in r: - break + time.sleep(0.1) - time.sleep(0.1) + return found - assert 'success' in r - assert self.get()['status'] == 200 - # Check if the only one router, controller, - # and application processes running. +def find_proc(name, ppid, ps_output): + return re.findall(fr'{ppid}.*{name}', ps_output) - out = subprocess.check_output(['ps', 'ax', '-O', 'ppid']).decode() - assert len(self.find_proc(self.PATTERN_ROUTER, unit_pid, out)) == 1 - assert len(self.find_proc(self.PATTERN_CONTROLLER, unit_pid, out)) == 1 - assert len(self.find_proc(self.app_name, unit_pid, out)) == 1 - def test_respawn_router(self, skip_alert, unit_pid, skip_fds_check): - skip_fds_check(router=True) - pid = self.pid_by_name(self.PATTERN_ROUTER, unit_pid) +def smoke_test(unit_pid): + for _ in range(10): + r = client.conf('1', f'applications/{client.app_name}/processes') - self.kill_pids(pid) - skip_alert(fr'process {pid} exited on signal 9') + if 'success' in r: + break - assert self.wait_for_process(self.PATTERN_ROUTER, unit_pid) is not None + time.sleep(0.1) - self.smoke_test(unit_pid) + assert 'success' in r + assert client.get()['status'] == 200 - def test_respawn_controller(self, skip_alert, unit_pid, skip_fds_check): - skip_fds_check(controller=True) - pid = self.pid_by_name(self.PATTERN_CONTROLLER, unit_pid) + # Check if the only one router, controller, + # and application processes running. - self.kill_pids(pid) - skip_alert(fr'process {pid} exited on signal 9') + 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 - assert ( - self.wait_for_process(self.PATTERN_CONTROLLER, unit_pid) is not None - ) - assert self.get()['status'] == 200 +def test_respawn_router(skip_alert, unit_pid, skip_fds_check): + skip_fds_check(router=True) + pid = pid_by_name(PATTERN_ROUTER, unit_pid) - self.smoke_test(unit_pid) + kill_pids(pid) + skip_alert(fr'process {pid} exited on signal 9') - def test_respawn_application(self, skip_alert, unit_pid): - pid = self.pid_by_name(self.app_name, unit_pid) + assert wait_for_process(PATTERN_ROUTER, unit_pid) is not None - self.kill_pids(pid) - skip_alert(fr'process {pid} exited on signal 9') + smoke_test(unit_pid) - assert self.wait_for_process(self.app_name, unit_pid) is not None - self.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) |