Remove psutil usages in utils_test. (#783)
diff --git a/tests/mobly/utils_test.py b/tests/mobly/utils_test.py
index c8908d1..61a8941 100755
--- a/tests/mobly/utils_test.py
+++ b/tests/mobly/utils_test.py
@@ -17,6 +17,7 @@
import os
import platform
import shutil
+import signal
import socket
import subprocess
import sys
@@ -33,12 +34,30 @@
from tests.lib import mock_instrumentation_test
from tests.lib import multiple_subclasses_module
import mock
-import psutil
MOCK_AVAILABLE_PORT = 5
ADB_MODULE_PACKAGE_NAME = 'mobly.controllers.android_device_lib.adb'
+def _is_process_running(pid):
+ """Whether the process with given PID is running."""
+ if os.name == 'nt':
+ return str(pid) in subprocess.check_output([
+ 'tasklist',
+ '/fi',
+ f'PID eq {pid}',
+ ]).decode()
+
+ try:
+ # os.kill throws OSError if the process with PID pid is not running.
+ # signal.SIG_DFL is one of two standard signal handling options, it will
+ # simply perform the default function for the signal.
+ os.kill(pid, signal.SIG_DFL)
+ except OSError:
+ return False
+ return True
+
+
class UtilsTest(unittest.TestCase):
"""Unit tests for the implementation of everything under mobly.utils."""
@@ -140,9 +159,9 @@
def test_start_standing_subproc(self):
try:
- p = utils.start_standing_subprocess(self.sleep_cmd(0.01))
- p1 = psutil.Process(p.pid)
- self.assertTrue(p1.is_running())
+ p = utils.start_standing_subprocess(self.sleep_cmd(4))
+ self.assertTrue(_is_process_running(p.pid))
+ os.kill(p.pid, signal.SIGTERM)
finally:
p.stdout.close()
p.stderr.close()
@@ -178,16 +197,14 @@
def test_stop_standing_subproc(self):
p = utils.start_standing_subprocess(self.sleep_cmd(4))
- p1 = psutil.Process(p.pid)
utils.stop_standing_subprocess(p)
- self.assertFalse(p1.is_running())
+ self.assertFalse(_is_process_running(p.pid))
- def test_stop_standing_subproc_wihtout_pipe(self):
+ def test_stop_standing_subproc_without_pipe(self):
p = subprocess.Popen(self.sleep_cmd(4))
self.assertIsNone(p.stdout)
- p1 = psutil.Process(p.pid)
utils.stop_standing_subprocess(p)
- self.assertFalse(p1.is_running())
+ self.assertFalse(_is_process_running(p.pid))
@unittest.skipIf(sys.version_info >= (3, 4) and sys.version_info < (3, 5),
'Python 3.4 does not support `None` max_workers.')
