Merged with the default branch.1.31.0-1

1 files changed, 72 insertions, 64 deletions

diff --git a/test/test_respawn.py b/test/test_respawn.py
index 3d3dfac3..dc465cda 100644
--- a/test/test_respawn.py
+++ b/test/test_respawn.py
@@ -2,99 +2,107 @@ import re
 import subprocess
 import time
 
-from unit.applications.lang.python import TestApplicationPython
-from unit.option import option
+import pytest
+from unit.applications.lang.python import ApplicationPython
 
+prerequisites = {'modules': {'python': 'any'}}
 
-class TestRespawn(TestApplicationPython):
-    prerequisites = {'modules': {'python': 'any'}}
+client = ApplicationPython()
 
-    PATTERN_ROUTER = 'unit: router'
-    PATTERN_CONTROLLER = 'unit: controller'
+PATTERN_ROUTER = 'unit: router'
+PATTERN_CONTROLLER = 'unit: controller'
 
-    def setup_method(self):
-        self.app_name = f'app-{option.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 i 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)