summaryrefslogtreecommitdiffhomepage
path: root/test/test_respawn.py
diff options
context:
space:
mode:
authorAndrei Zeliankou <zelenkov@nginx.com>2023-06-14 18:20:09 +0100
committerAndrei Zeliankou <zelenkov@nginx.com>2023-06-14 18:20:09 +0100
commitc183bd8749a19477390f8cb77efe5f6d223f0905 (patch)
tree4e821e9cb07be9a86bf2d442acb3ea6740ba5a99 /test/test_respawn.py
parentc6d05191a069ac150cc8eb2bece75cf79c0a465a (diff)
downloadunit-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 'test/test_respawn.py')
-rw-r--r--test/test_respawn.py133
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)