Tests: run_process() and stop_processes() introduced.

This commit is contained in:
Andrei Zeliankou
2019-11-14 17:15:20 +03:00
parent b5e3e22a46
commit efb461c0e1

View File

@@ -185,6 +185,8 @@ class TestUnit(unittest.TestCase):
if self._started: if self._started:
self._stop() self._stop()
self.stop_processes()
def _run(self): def _run(self):
self.unitd = self.pardir + '/build/unitd' self.unitd = self.pardir + '/build/unitd'
@@ -287,6 +289,26 @@ class TestUnit(unittest.TestCase):
if found: if found:
print('skipped.') print('skipped.')
def run_process(self, target):
if not hasattr(self, '_processes'):
self._processes = []
process = Process(target=target)
process.start()
self._processes.append(process)
def stop_processes(self):
if not hasattr(self, '_processes'):
return
for process in self._processes:
process.terminate()
process.join(timeout=5)
if process.is_alive():
self.fail('Fail to stop process')
def waitforfiles(self, *files): def waitforfiles(self, *files):
for i in range(50): for i in range(50):
wait = False wait = False