summaryrefslogtreecommitdiffhomepage
path: root/test/test_python_procman.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_python_procman.py')
-rw-r--r--test/test_python_procman.py137
1 files changed, 57 insertions, 80 deletions
diff --git a/test/test_python_procman.py b/test/test_python_procman.py
index c327ab14..8eccae3e 100644
--- a/test/test_python_procman.py
+++ b/test/test_python_procman.py
@@ -1,7 +1,8 @@
import re
import subprocess
import time
-import unittest
+
+import pytest
from unit.applications.lang.python import TestApplicationPython
@@ -9,10 +10,10 @@ from unit.applications.lang.python import TestApplicationPython
class TestPythonProcman(TestApplicationPython):
prerequisites = {'modules': {'python': 'any'}}
- def setUp(self):
- super().setUp()
+ def setup_method(self):
+ super().setup_method()
- self.app_name = "app-" + self.testdir.split('/')[-1]
+ self.app_name = "app-" + self.temp_dir.split('/')[-1]
self.app_proc = 'applications/' + self.app_name + '/processes'
self.load('empty', self.app_name)
@@ -23,7 +24,7 @@ class TestPythonProcman(TestApplicationPython):
pids = set()
for m in re.findall('.*' + self.app_name, output.decode()):
- pids.add(re.search('^\s*(\d+)', m).group(1))
+ pids.add(re.search(r'^\s*(\d+)', m).group(1))
return pids
@@ -31,35 +32,35 @@ class TestPythonProcman(TestApplicationPython):
if path is None:
path = self.app_proc
- self.assertIn('success', self.conf(conf, path), 'configure processes')
+ assert 'success' in self.conf(conf, path), 'configure processes'
- @unittest.skip('not yet')
+ @pytest.mark.skip('not yet')
def test_python_processes_idle_timeout_zero(self):
self.conf_proc({"spare": 0, "max": 2, "idle_timeout": 0})
self.get()
- self.assertEqual(len(self.pids_for_process()), 0, 'idle timeout 0')
+ assert len(self.pids_for_process()) == 0, 'idle timeout 0'
def test_python_prefork(self):
self.conf_proc('2')
pids = self.pids_for_process()
- self.assertEqual(len(pids), 2, 'prefork 2')
+ assert len(pids) == 2, 'prefork 2'
self.get()
- self.assertSetEqual(self.pids_for_process(), pids, 'prefork still 2')
+ assert self.pids_for_process() == pids, 'prefork still 2'
self.conf_proc('4')
pids = self.pids_for_process()
- self.assertEqual(len(pids), 4, 'prefork 4')
+ assert len(pids) == 4, 'prefork 4'
self.get()
- self.assertSetEqual(self.pids_for_process(), pids, 'prefork still 4')
+ assert self.pids_for_process() == pids, 'prefork still 4'
self.stop_all()
- @unittest.skip('not yet')
+ @pytest.mark.skip('not yet')
def test_python_prefork_same_processes(self):
self.conf_proc('2')
pids = self.pids_for_process()
@@ -67,25 +68,23 @@ class TestPythonProcman(TestApplicationPython):
self.conf_proc('4')
pids_new = self.pids_for_process()
- self.assertTrue(pids.issubset(pids_new), 'prefork same processes')
+ assert pids.issubset(pids_new), 'prefork same processes'
def test_python_ondemand(self):
self.conf_proc({"spare": 0, "max": 8, "idle_timeout": 1})
- self.assertEqual(len(self.pids_for_process()), 0, 'on-demand 0')
+ assert len(self.pids_for_process()) == 0, 'on-demand 0'
self.get()
pids = self.pids_for_process()
- self.assertEqual(len(pids), 1, 'on-demand 1')
+ assert len(pids) == 1, 'on-demand 1'
self.get()
- self.assertSetEqual(self.pids_for_process(), pids, 'on-demand still 1')
+ assert self.pids_for_process() == pids, 'on-demand still 1'
time.sleep(1)
- self.assertEqual(
- len(self.pids_for_process()), 0, 'on-demand stop idle'
- )
+ assert len(self.pids_for_process()) == 0, 'on-demand stop idle'
self.stop_all()
@@ -93,27 +92,25 @@ class TestPythonProcman(TestApplicationPython):
self.conf_proc({"spare": 2, "max": 8, "idle_timeout": 1})
pids = self.pids_for_process()
- self.assertEqual(len(pids), 2, 'updown 2')
+ assert len(pids) == 2, 'updown 2'
self.get()
pids_new = self.pids_for_process()
- self.assertEqual(len(pids_new), 3, 'updown 3')
- self.assertTrue(pids.issubset(pids_new), 'updown 3 only 1 new')
+ assert len(pids_new) == 3, 'updown 3'
+ assert pids.issubset(pids_new), 'updown 3 only 1 new'
self.get()
- self.assertSetEqual(
- self.pids_for_process(), pids_new, 'updown still 3'
- )
+ assert self.pids_for_process() == pids_new, 'updown still 3'
time.sleep(1)
pids = self.pids_for_process()
- self.assertEqual(len(pids), 2, 'updown stop idle')
+ assert len(pids) == 2, 'updown stop idle'
self.get()
pids_new = self.pids_for_process()
- self.assertEqual(len(pids_new), 3, 'updown again 3')
- self.assertTrue(pids.issubset(pids_new), 'updown again 3 only 1 new')
+ assert len(pids_new) == 3, 'updown again 3'
+ assert pids.issubset(pids_new), 'updown again 3 only 1 new'
self.stop_all()
@@ -121,20 +118,20 @@ class TestPythonProcman(TestApplicationPython):
self.conf_proc({"spare": 2, "max": 6, "idle_timeout": 1})
pids = self.pids_for_process()
- self.assertEqual(len(pids), 2, 'reconf 2')
+ assert len(pids) == 2, 'reconf 2'
self.get()
pids_new = self.pids_for_process()
- self.assertEqual(len(pids_new), 3, 'reconf 3')
- self.assertTrue(pids.issubset(pids_new), 'reconf 3 only 1 new')
+ assert len(pids_new) == 3, 'reconf 3'
+ assert pids.issubset(pids_new), 'reconf 3 only 1 new'
self.conf_proc('6', self.app_proc + '/spare')
pids = self.pids_for_process()
- self.assertEqual(len(pids), 6, 'reconf 6')
+ assert len(pids) == 6, 'reconf 6'
self.get()
- self.assertSetEqual(self.pids_for_process(), pids, 'reconf still 6')
+ assert self.pids_for_process() == pids, 'reconf still 6'
self.stop_all()
@@ -143,7 +140,7 @@ class TestPythonProcman(TestApplicationPython):
self.get()
pids = self.pids_for_process()
- self.assertEqual(len(pids), 1, 'idle timeout 1')
+ assert len(pids) == 1, 'idle timeout 1'
time.sleep(1)
@@ -152,14 +149,12 @@ class TestPythonProcman(TestApplicationPython):
time.sleep(1)
pids_new = self.pids_for_process()
- self.assertEqual(len(pids_new), 1, 'idle timeout still 1')
- self.assertSetEqual(
- self.pids_for_process(), pids, 'idle timeout still 1 same pid'
- )
+ assert len(pids_new) == 1, 'idle timeout still 1'
+ assert self.pids_for_process() == pids, 'idle timeout still 1 same pid'
time.sleep(1)
- self.assertEqual(len(self.pids_for_process()), 0, 'idle timed out')
+ assert len(self.pids_for_process()) == 0, 'idle timed out'
def test_python_processes_connection_keepalive(self):
self.conf_proc({"spare": 0, "max": 6, "idle_timeout": 2})
@@ -169,15 +164,11 @@ class TestPythonProcman(TestApplicationPython):
start=True,
read_timeout=1,
)
- self.assertEqual(
- len(self.pids_for_process()), 1, 'keepalive connection 1'
- )
+ assert len(self.pids_for_process()) == 1, 'keepalive connection 1'
time.sleep(2)
- self.assertEqual(
- len(self.pids_for_process()), 0, 'keepalive connection 0'
- )
+ assert len(self.pids_for_process()) == 0, 'keepalive connection 0'
sock.close()
@@ -185,43 +176,29 @@ class TestPythonProcman(TestApplicationPython):
self.conf_proc('1')
path = '/' + self.app_proc
- self.assertIn('error', self.conf_get(path + '/max'))
- self.assertIn('error', self.conf_get(path + '/spare'))
- self.assertIn('error', self.conf_get(path + '/idle_timeout'))
+ assert 'error' in self.conf_get(path + '/max')
+ assert 'error' in self.conf_get(path + '/spare')
+ assert 'error' in self.conf_get(path + '/idle_timeout')
def test_python_processes_invalid(self):
- self.assertIn(
- 'error', self.conf({"spare": -1}, self.app_proc), 'negative spare',
- )
- self.assertIn(
- 'error', self.conf({"max": -1}, self.app_proc), 'negative max',
- )
- self.assertIn(
- 'error',
- self.conf({"idle_timeout": -1}, self.app_proc),
- 'negative idle_timeout',
- )
- self.assertIn(
- 'error',
- self.conf({"spare": 2}, self.app_proc),
- 'spare gt max default',
- )
- self.assertIn(
- 'error',
- self.conf({"spare": 2, "max": 1}, self.app_proc),
- 'spare gt max',
- )
- self.assertIn(
- 'error',
- self.conf({"spare": 0, "max": 0}, self.app_proc),
- 'max zero',
- )
+ assert 'error' in self.conf(
+ {"spare": -1}, self.app_proc
+ ), 'negative spare'
+ assert 'error' in self.conf({"max": -1}, self.app_proc), 'negative max'
+ assert 'error' in self.conf(
+ {"idle_timeout": -1}, self.app_proc
+ ), 'negative idle_timeout'
+ assert 'error' in self.conf(
+ {"spare": 2}, self.app_proc
+ ), 'spare gt max default'
+ assert 'error' in self.conf(
+ {"spare": 2, "max": 1}, self.app_proc
+ ), 'spare gt max'
+ assert 'error' in self.conf(
+ {"spare": 0, "max": 0}, self.app_proc
+ ), 'max zero'
def stop_all(self):
self.conf({"listeners": {}, "applications": {}})
- self.assertEqual(len(self.pids_for_process()), 0, 'stop all')
-
-
-if __name__ == '__main__':
- TestPythonProcman.main()
+ assert len(self.pids_for_process()) == 0, 'stop all'