36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
|
import os
|
||
|
|
import subprocess
|
||
|
|
|
||
|
|
import psutil
|
||
|
|
|
||
|
|
from catfish.utils import processes
|
||
|
|
from tests import BaseTestCase
|
||
|
|
|
||
|
|
|
||
|
|
class RootProcessTestCase(BaseTestCase):
|
||
|
|
def test_root_processes(self):
|
||
|
|
root_process = processes.get_root_process()
|
||
|
|
self.assertIsNone(root_process.parent())
|
||
|
|
all_children = [proc.pid for proc in root_process.children(recursive=True)]
|
||
|
|
self.assertIn(os.getpid(), all_children)
|
||
|
|
|
||
|
|
def test_current_process(self):
|
||
|
|
self.assertEqual(processes.CURRENT_PROCESS.pid, os.getpid())
|
||
|
|
|
||
|
|
|
||
|
|
class TerminateProcessesTestCase(BaseTestCase):
|
||
|
|
def test_kills_lots_of_processes(self):
|
||
|
|
created_processes = []
|
||
|
|
for _ in range(10):
|
||
|
|
created_processes.append(subprocess.Popen("yes", stdout=subprocess.PIPE))
|
||
|
|
|
||
|
|
for proc in created_processes:
|
||
|
|
self.assertIsNone(proc.poll())
|
||
|
|
|
||
|
|
still_alive = processes.terminate_processes(
|
||
|
|
[psutil.Process(proc.pid) for proc in created_processes]
|
||
|
|
)
|
||
|
|
self.assertIsNone(still_alive)
|
||
|
|
for proc in created_processes:
|
||
|
|
self.assertEqual(proc.poll(), 0)
|