877 lines
40 KiB
Python
877 lines
40 KiB
Python
""" Unit tests for adb_monitor. """
|
|
# pylint: disable=protected-access,no-self-use,missing-docstring,too-many-public-methods
|
|
|
|
import os
|
|
import unittest
|
|
|
|
from unittest.mock import MagicMock, mock_open, patch
|
|
|
|
import buildscripts.mobile.adb_monitor as adb_monitor
|
|
|
|
ADB_MONITOR = "buildscripts.mobile.adb_monitor"
|
|
|
|
|
|
def ns(module):
|
|
"""Get the namespace."""
|
|
# pylint: disable=invalid-name
|
|
return f"{ADB_MONITOR}.{module}"
|
|
|
|
|
|
class TestParseCommandLine(unittest.TestCase):
|
|
def test_parse_command_line(self):
|
|
options = adb_monitor.parse_command_line().parse_args([])
|
|
self.assertEqual(options.adb_binary, adb_monitor.DEFAULT_ADB_BINARY)
|
|
self.assertEqual(options.python27, adb_monitor.DEFAULT_PYTHON27)
|
|
self.assertEqual(options.sample_interval_ms, adb_monitor.DEFAULT_SAMPLE_INTERVAL_MS)
|
|
self.assertEqual(options.log_level, adb_monitor.DEFAULT_LOG_LEVEL)
|
|
self.assertEqual(options.battery_file, adb_monitor.DEFAULT_BATTERY_FILE)
|
|
self.assertEqual(options.memory_file, adb_monitor.DEFAULT_MEMORY_FILE)
|
|
self.assertEqual(options.cpu_file, adb_monitor.DEFAULT_CPU_FILE)
|
|
self.assertEqual(options.num_samples, adb_monitor.DEFAULT_NUM_SAMPLES)
|
|
self.assertEqual(options.sample_interval_ms, adb_monitor.DEFAULT_SAMPLE_INTERVAL_MS)
|
|
self.assertIsNone(options.collection_time_secs)
|
|
|
|
def test_parse_command_line_partial_args(self):
|
|
adb_binary = "myadb"
|
|
python27 = "mypython2"
|
|
collection_time_secs = 35
|
|
battery_file = "mybattery"
|
|
arg_list = [
|
|
"--adbBinary", adb_binary, "--python27", python27, "--collectionTime",
|
|
str(collection_time_secs), "--batteryFile", battery_file, "--noMemory"
|
|
]
|
|
options = adb_monitor.parse_command_line().parse_args(arg_list)
|
|
self.assertEqual(options.adb_binary, adb_binary)
|
|
self.assertEqual(options.python27, python27)
|
|
self.assertEqual(options.collection_time_secs, collection_time_secs)
|
|
self.assertEqual(options.battery_file, battery_file)
|
|
|
|
self.assertEqual(options.sample_interval_ms, adb_monitor.DEFAULT_SAMPLE_INTERVAL_MS)
|
|
self.assertEqual(options.log_level, adb_monitor.DEFAULT_LOG_LEVEL)
|
|
self.assertIsNone(options.memory_file)
|
|
self.assertEqual(options.cpu_file, adb_monitor.DEFAULT_CPU_FILE)
|
|
self.assertEqual(options.num_samples, adb_monitor.DEFAULT_NUM_SAMPLES)
|
|
self.assertEqual(options.sample_interval_ms, adb_monitor.DEFAULT_SAMPLE_INTERVAL_MS)
|
|
|
|
def test_parse_command_line_no_files(self):
|
|
arg_list = ["--noBattery", "--noCpu", "--noMemory"]
|
|
options = adb_monitor.parse_command_line().parse_args(arg_list)
|
|
self.assertIsNone(options.battery_file)
|
|
self.assertIsNone(options.cpu_file)
|
|
self.assertIsNone(options.memory_file)
|
|
|
|
def test_parse_command_line_no_battery_first(self):
|
|
battery_file = "mybattery"
|
|
arg_list = ["--noBattery", "--batteryFile", battery_file]
|
|
options = adb_monitor.parse_command_line().parse_args(arg_list)
|
|
self.assertEqual(options.battery_file, battery_file)
|
|
|
|
def test_parse_command_line_no_battery_second(self):
|
|
battery_file = "mybattery"
|
|
arg_list = ["--batteryFile", battery_file, "--noBattery"]
|
|
options = adb_monitor.parse_command_line().parse_args(arg_list)
|
|
self.assertIsNone(options.battery_file)
|
|
|
|
|
|
class TestMonitorDevice(unittest.TestCase):
|
|
@patch(ns("fileops"), return_value=False)
|
|
def test_monitor_device(self, mock_fileops):
|
|
"""Basic test monitor device."""
|
|
files_mtime = {"file1": 0, "file2": 0}
|
|
mock_fileops.getmtime.return_value = 10
|
|
mock_adb_control = MagicMock()
|
|
adb_monitor.monitor_device(mock_adb_control, files_mtime)
|
|
mock_adb_control.start.assert_called_once()
|
|
mock_adb_control.wait.assert_called_once()
|
|
self.assertEqual(mock_fileops.getmtime.call_count, len(files_mtime))
|
|
self.assertEqual(mock_fileops.is_empty.call_count, len(files_mtime))
|
|
|
|
@patch(ns("fileops"), return_value=True)
|
|
def test_monitor_device_empty_file(self, mock_fileops):
|
|
files_mtime = {"file1": 0, "file2": 0}
|
|
mock_fileops.getmtime.return_value = 10
|
|
mock_adb_control = MagicMock()
|
|
adb_monitor.monitor_device(mock_adb_control, files_mtime)
|
|
mock_adb_control.start.assert_called_once()
|
|
mock_adb_control.wait.assert_called_once()
|
|
self.assertEqual(mock_fileops.getmtime.call_count, len(files_mtime))
|
|
self.assertEqual(mock_fileops.is_empty.call_count, len(files_mtime))
|
|
|
|
@patch(ns("fileops"), return_value=False)
|
|
def test_monitor_device_earlier_mtime(self, mock_fileops):
|
|
files_mtime = {"file1": 10, "file2": 10}
|
|
mock_fileops.getmtime.return_value = 6
|
|
mock_adb_control = MagicMock()
|
|
adb_monitor.monitor_device(mock_adb_control, files_mtime)
|
|
mock_adb_control.start.assert_called_once()
|
|
mock_adb_control.wait.assert_called_once()
|
|
self.assertEqual(mock_fileops.getmtime.call_count, len(files_mtime))
|
|
mock_fileops.is_empty.assert_not_called()
|
|
|
|
@patch(ns("fileops"), return_value=False)
|
|
def test_monitor_device_no_files(self, mock_fileops):
|
|
files_mtime = {}
|
|
mock_fileops.getmtime.return_value = 10
|
|
mock_adb_control = MagicMock()
|
|
adb_monitor.monitor_device(mock_adb_control, files_mtime)
|
|
mock_adb_control.start.assert_called_once()
|
|
mock_adb_control.wait.assert_called_once()
|
|
mock_fileops.getmtime.assert_not_called()
|
|
mock_fileops.is_empty.assert_not_called()
|
|
|
|
|
|
class TestOutputFilesMtime(unittest.TestCase):
|
|
@patch(ns("fileops.getmtime"))
|
|
def test_output_files_mtime(self, mock_getmtime):
|
|
mtime = 11
|
|
mock_getmtime.return_value = mtime
|
|
files = ["file1", "file2"]
|
|
m_files = adb_monitor.create_files_mtime(files)
|
|
self.assertEqual(len(m_files), len(files))
|
|
for fn in files:
|
|
self.assertEqual(m_files[fn], mtime)
|
|
|
|
@patch(ns("fileops.getmtime"))
|
|
def test_output_files_mtime_no_files(self, mock_getmtime):
|
|
# pylint: disable=unused-argument
|
|
m_files = adb_monitor.create_files_mtime([])
|
|
self.assertEqual(len(m_files), 0)
|
|
|
|
|
|
class TestFindExecutable(unittest.TestCase):
|
|
def test_find_executable(self):
|
|
my_binary = "mybinary"
|
|
with patch(ns("distutils.spawn.find_executable"), side_effect=lambda x: x):
|
|
adb_binary = adb_monitor.find_executable(my_binary)
|
|
self.assertEqual(adb_binary, my_binary)
|
|
|
|
def test_find_executable_not_found(self):
|
|
with patch(ns("distutils.spawn.find_executable"), side_effect=lambda x: x),\
|
|
self.assertRaises(EnvironmentError):
|
|
adb_monitor.find_executable(None)
|
|
|
|
|
|
class TestAdb(unittest.TestCase):
|
|
@patch(ns("os.path.isfile"), return_value=True)
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
def test___init__(self, mock_find_executable, mock_isfile):
|
|
# pylint: disable=unused-argument
|
|
os_path = os.environ["PATH"]
|
|
adb = adb_monitor.Adb()
|
|
self.assertTrue(adb.systrace_script.startswith(os.path.join("systrace", "systrace.py")))
|
|
self.assertEqual(adb.logger, adb_monitor.LOGGER)
|
|
self.assertEqual(adb.python27, adb_monitor.DEFAULT_PYTHON27)
|
|
self.assertEqual(adb.logger, adb_monitor.LOGGER)
|
|
self.assertEqual(os.environ["PATH"], os_path)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
def test___init__adb_binary(self, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
adb_dir = os.path.join("/root", "adb_dir")
|
|
adb_path = os.path.join(adb_dir, "adb")
|
|
mock_os.environ = {"PATH": os.environ["PATH"]}
|
|
mock_os.path.pathsep = os.path.pathsep
|
|
mock_os.path.dirname = lambda x: x
|
|
adb = adb_monitor.Adb(adb_binary=adb_path)
|
|
self.assertTrue(adb.systrace_script.startswith(adb_dir))
|
|
self.assertIn(os.path.pathsep + adb_dir, mock_os.environ["PATH"])
|
|
|
|
@patch(ns("os.environ"))
|
|
@patch(ns("os.path.isfile"), return_value=True)
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
def test___init__python27_binary(self, mock_find_executable, mock_isfile, mock_os_environ):
|
|
# pylint: disable=unused-argument
|
|
python_dir = os.path.join("/root", "python27_dir")
|
|
python27_path = os.path.join(python_dir, "python2")
|
|
adb = adb_monitor.Adb(python27=python27_path)
|
|
self.assertEqual(adb.python27, python27_path)
|
|
|
|
@patch(ns("os.environ"))
|
|
@patch(ns("os.path.isfile"), return_value=False)
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
def test___init__bad_systrace(self, mock_find_executable, mock_isfile, mock_os_environ):
|
|
# pylint: disable=unused-argument
|
|
with self.assertRaises(EnvironmentError):
|
|
adb_monitor.Adb()
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("runcommand"))
|
|
def test_adb_cmd_output(self, mock_runcmd, mock_os):
|
|
# pylint: disable=unused-argument
|
|
adb_result = adb_monitor.Adb.adb_cmd("mycmd")
|
|
self.assertEqual(adb_result, mock_runcmd.RunCommand().execute_with_output())
|
|
self.assertNotEqual(adb_result, mock_runcmd.RunCommand().execute_save_output())
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("runcommand"))
|
|
def test_adb_cmd_output_string(self, mock_runcmd, mock_os):
|
|
# pylint: disable=unused-argument
|
|
adb_result = adb_monitor.Adb.adb_cmd("mycmd", output_string=True)
|
|
self.assertEqual(adb_result, mock_runcmd.RunCommand().execute_with_output())
|
|
self.assertNotEqual(adb_result, mock_runcmd.RunCommand().execute_save_output())
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("runcommand"))
|
|
def test_adb_cmd_save_output(self, mock_runcmd, mock_os):
|
|
# pylint: disable=unused-argument
|
|
adb_result = adb_monitor.Adb.adb_cmd("mycmd", output_file="myfile")
|
|
self.assertEqual(adb_result, mock_runcmd.RunCommand().execute_save_output())
|
|
self.assertNotEqual(adb_result, mock_runcmd.RunCommand().execute_with_output())
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("runcommand"))
|
|
def test_adb_cmd_all_params(self, mock_runcmd, mock_os):
|
|
# pylint: disable=unused-argument
|
|
adb_result = adb_monitor.Adb.adb_cmd("mycmd", output_file="myfile", append_file=True,
|
|
output_string=True)
|
|
self.assertNotEqual(adb_result, mock_runcmd.RunCommand().execute_save_output())
|
|
self.assertEqual(adb_result, mock_runcmd.RunCommand().execute_with_output())
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("runcommand"))
|
|
def test_shell(self, mock_runcmd, mock_os):
|
|
# pylint: disable=unused-argument
|
|
cmd_output = adb_monitor.Adb.shell("mycmd")
|
|
self.assertEqual(cmd_output, mock_runcmd.RunCommand().execute_with_output())
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("runcommand"))
|
|
def test_shell_stripped(self, mock_runcmd, mock_os):
|
|
# pylint: disable=unused-argument
|
|
output = "output from shell"
|
|
mock_runcmd.RunCommand().execute_with_output.return_value = output + "__EXIT__:0\n"
|
|
cmd_output = adb_monitor.Adb.shell("mycmd")
|
|
self.assertEqual(cmd_output, output)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("runcommand"))
|
|
def test_shell_error(self, mock_runcmd, mock_os):
|
|
# pylint: disable=unused-argument
|
|
output = "output from shell"
|
|
mock_runcmd.RunCommand().execute_with_output.return_value = output + "__EXIT__:1\n"
|
|
with self.assertRaises(RuntimeError):
|
|
adb_monitor.Adb.shell("mycmd")
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test_devices(self, mock_runcmd, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
adb = adb_monitor.Adb()
|
|
adb.devices()
|
|
mock_runcmd.RunCommand.assert_called_once_with("adb devices -l", unittest.mock.ANY,
|
|
unittest.mock.ANY)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test_device_available(self, mock_runcmd, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
adb = adb_monitor.Adb()
|
|
adb.device_available()
|
|
mock_runcmd.RunCommand.assert_called_once_with("adb shell uptime", unittest.mock.ANY,
|
|
unittest.mock.ANY)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test_push(self, mock_runcmd, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
files = "myfile"
|
|
remote_dir = "/remotedir"
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
adb = adb_monitor.Adb()
|
|
adb.push(files, remote_dir)
|
|
push_cmd = mock_runcmd.RunCommand.call_args_list[0][0][0]
|
|
self.assertIn(files, push_cmd)
|
|
self.assertIn(remote_dir, push_cmd)
|
|
self.assertNotIn("--sync", push_cmd)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test_push_list(self, mock_runcmd, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
files = ["myfile", "file2"]
|
|
remote_dir = "/remotedir"
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
adb = adb_monitor.Adb()
|
|
adb.push(files, remote_dir)
|
|
push_cmd = mock_runcmd.RunCommand.call_args_list[0][0][0]
|
|
self.assertIn(" ".join(files), push_cmd)
|
|
self.assertIn(remote_dir, push_cmd)
|
|
self.assertNotIn("--sync", push_cmd)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test_push_sync(self, mock_runcmd, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
files = ["myfile", "file2"]
|
|
remote_dir = "/remotedir"
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
adb = adb_monitor.Adb()
|
|
adb.push(files, remote_dir, sync=True)
|
|
push_cmd = mock_runcmd.RunCommand.call_args_list[0][0][0]
|
|
for file_name in files:
|
|
self.assertIn(file_name, push_cmd)
|
|
self.assertIn(remote_dir, push_cmd)
|
|
self.assertIn("--sync", push_cmd)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test_pull(self, mock_runcmd, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
files = "myfile"
|
|
local_dir = "/localdir"
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
adb = adb_monitor.Adb()
|
|
adb.pull(files, local_dir)
|
|
pull_cmd = mock_runcmd.RunCommand.call_args_list[0][0][0]
|
|
self.assertIn(files, pull_cmd)
|
|
self.assertIn(local_dir, pull_cmd)
|
|
self.assertNotIn("--sync", pull_cmd)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test_pull_files(self, mock_runcmd, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
files = ["myfile", "file2"]
|
|
local_dir = "/localdir"
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
adb = adb_monitor.Adb()
|
|
adb.pull(files, local_dir)
|
|
pull_cmd = mock_runcmd.RunCommand.call_args_list[0][0][0]
|
|
for file_name in files:
|
|
self.assertIn(file_name, pull_cmd)
|
|
self.assertIn(local_dir, pull_cmd)
|
|
self.assertNotIn("--sync", pull_cmd)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test__battery_cmd(self, mock_runcmd, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
option = "myopt"
|
|
battery_cmd = "shell dumpsys batterystats " + option
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
adb = adb_monitor.Adb()
|
|
adb._battery_cmd(option)
|
|
mock_runcmd.RunCommand.assert_called_once_with("adb " + battery_cmd, unittest.mock.ANY,
|
|
unittest.mock.ANY)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test__battery_cmd_save_output(self, mock_runcmd, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
option = "myopt"
|
|
battery_cmd = "shell dumpsys batterystats " + option
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
adb = adb_monitor.Adb()
|
|
adb._battery_cmd(option, "myfile")
|
|
mock_runcmd.RunCommand.assert_called_once_with("adb " + battery_cmd, unittest.mock.ANY,
|
|
unittest.mock.ANY)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test_battery(self, mock_runcmd, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
adb = adb_monitor.Adb()
|
|
adb.battery("myfile")
|
|
battery_cmd = mock_runcmd.RunCommand.call_args_list[0][0][0]
|
|
self.assertIn("--checkin", battery_cmd)
|
|
self.assertNotIn("--reset", battery_cmd)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test_battery_reset(self, mock_runcmd, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
adb = adb_monitor.Adb()
|
|
adb.battery("myfile", reset=True, append_file=True)
|
|
battery_cmd1 = mock_runcmd.RunCommand.call_args_list[0][0][0]
|
|
battery_cmd2 = mock_runcmd.RunCommand.call_args_list[1][0][0]
|
|
self.assertIn("--reset", battery_cmd1)
|
|
self.assertIn("--checkin", battery_cmd2)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test_memory(self, mock_runcmd, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
memory_cmd = "shell dumpsys meminfo -c -d"
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
adb = adb_monitor.Adb()
|
|
adb.memory("myfile")
|
|
mock_runcmd.RunCommand.assert_called_once_with("adb " + memory_cmd, unittest.mock.ANY,
|
|
unittest.mock.ANY)
|
|
|
|
@patch(ns("tempfile.NamedTemporaryFile"))
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test_systrace_start(self, mock_runcmd, mock_find_executable, mock_os, mock_tempfile):
|
|
# pylint: disable=unused-argument
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
systrace_script = "systrace.py"
|
|
mock_os.path.join.return_value = systrace_script
|
|
adb = adb_monitor.Adb()
|
|
adb.systrace_start()
|
|
mock_runcmd.RunCommand.assert_called_once_with(output_file=mock_tempfile().name,
|
|
propagate_signals=False)
|
|
self.assertEqual(adb.systrace_script, systrace_script)
|
|
self.assertEqual(mock_runcmd.RunCommand().add_file.call_count, 3)
|
|
mock_runcmd.RunCommand().add_file.assert_any_call(adb_monitor.DEFAULT_PYTHON27)
|
|
mock_runcmd.RunCommand().add_file.assert_any_call(systrace_script)
|
|
self.assertEqual(mock_runcmd.RunCommand().add.call_count, 3)
|
|
mock_runcmd.RunCommand().add.assert_any_call("--json")
|
|
mock_runcmd.RunCommand().add.assert_any_call("-o")
|
|
mock_runcmd.RunCommand().add.assert_any_call("dalvik sched freq idle load")
|
|
mock_runcmd.RunCommand().start_process.assert_called_once()
|
|
|
|
@patch(ns("tempfile.NamedTemporaryFile"))
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
@patch(ns("runcommand"))
|
|
def test_systrace_start_output_file(self, mock_runcmd, mock_find_executable, mock_os,
|
|
mock_tempfile):
|
|
# pylint: disable=unused-argument
|
|
output_file = "myfile"
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
systrace_script = "systrace.py"
|
|
mock_os.path.join.return_value = systrace_script
|
|
adb = adb_monitor.Adb()
|
|
adb.systrace_start(output_file)
|
|
mock_runcmd.RunCommand.assert_called_once_with(output_file=mock_tempfile().name,
|
|
propagate_signals=False)
|
|
self.assertEqual(adb.systrace_script, systrace_script)
|
|
self.assertEqual(mock_runcmd.RunCommand().add_file.call_count, 3)
|
|
mock_runcmd.RunCommand().add_file.assert_any_call(adb_monitor.DEFAULT_PYTHON27)
|
|
mock_runcmd.RunCommand().add_file.assert_any_call(systrace_script)
|
|
mock_runcmd.RunCommand().add_file.assert_any_call(output_file)
|
|
self.assertEqual(mock_runcmd.RunCommand().add.call_count, 3)
|
|
mock_runcmd.RunCommand().add.assert_any_call("--json")
|
|
mock_runcmd.RunCommand().add.assert_any_call("-o")
|
|
mock_runcmd.RunCommand().add.assert_any_call("dalvik sched freq idle load")
|
|
mock_runcmd.RunCommand().start_process.assert_called_once()
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
def test_systrace_stop(self, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
systrace_script = "systrace.py"
|
|
systrace_output = "Systrace: Wrote trace"
|
|
mock_os.path.join.return_value = systrace_script
|
|
adb = adb_monitor.Adb()
|
|
adb._cmd = MagicMock()
|
|
with patch(ADB_MONITOR + ".open", mock_open(read_data=systrace_output)):
|
|
adb.systrace_stop()
|
|
adb._cmd.send_to_process.assert_called_once_with(b"bye")
|
|
mock_os.remove.assert_called_once_with(adb._tempfile)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
def test_systrace_stop_output_file(self, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
systrace_script = "systrace.py"
|
|
systrace_output = "Systrace: Wrote trace"
|
|
output_file = "myfile"
|
|
mock_os.path.join.return_value = systrace_script
|
|
adb = adb_monitor.Adb()
|
|
adb._cmd = MagicMock()
|
|
with patch(ADB_MONITOR + ".open", mock_open(read_data=systrace_output)):
|
|
adb.systrace_stop(output_file=output_file)
|
|
adb._cmd.send_to_process.assert_called_once_with(b"bye")
|
|
mock_os.remove.assert_called_once_with(adb._tempfile)
|
|
|
|
@patch(ns("os"))
|
|
@patch(ns("find_executable"), side_effect=lambda x: x)
|
|
def test_systrace_stop_no_trace(self, mock_find_executable, mock_os):
|
|
# pylint: disable=unused-argument
|
|
mock_os.path.dirname.return_value = "adb_dir"
|
|
systrace_script = "systrace.py"
|
|
systrace_output = "Systrace: did not Write trace"
|
|
output_file = "myfile"
|
|
mock_os.path.join.return_value = systrace_script
|
|
adb = adb_monitor.Adb()
|
|
adb._cmd = MagicMock()
|
|
with patch(ADB_MONITOR + ".open", mock_open(read_data=systrace_output)):
|
|
adb.systrace_stop(output_file=output_file)
|
|
adb._cmd.send_to_process.assert_called_once_with(b"bye")
|
|
self.assertEqual(mock_os.remove.call_count, 2)
|
|
mock_os.remove.assert_any_call(adb._tempfile)
|
|
mock_os.remove.assert_any_call(output_file)
|
|
|
|
|
|
class TestAdbControl(unittest.TestCase):
|
|
def test___init___no_files(self):
|
|
mock_adb = MagicMock()
|
|
with self.assertRaises(ValueError):
|
|
adb_monitor.AdbControl(mock_adb)
|
|
|
|
@patch(ns("fileops.create_empty"))
|
|
def test___init___all_files(self, mock_create_empty):
|
|
mock_adb = MagicMock()
|
|
battery_file = "mybattery"
|
|
memory_file = "mymemory"
|
|
cpu_file = "mycpu"
|
|
adb_control = adb_monitor.AdbControl(mock_adb, battery_file=battery_file,
|
|
memory_file=memory_file, cpu_file=cpu_file)
|
|
self.assertEqual(adb_control.adb, mock_adb)
|
|
self.assertEqual(adb_control.battery_file, battery_file)
|
|
self.assertEqual(adb_control.memory_file, memory_file)
|
|
self.assertEqual(adb_control.cpu_file, cpu_file)
|
|
self.assertEqual(adb_control.num_samples, 0)
|
|
self.assertIsNone(adb_control.collection_time_secs)
|
|
self.assertEqual(adb_control.sample_interval_ms, adb_monitor.DEFAULT_SAMPLE_INTERVAL_MS)
|
|
self.assertEqual(mock_create_empty.call_count, 3)
|
|
mock_create_empty.assert_any_call(battery_file)
|
|
mock_create_empty.assert_any_call(memory_file)
|
|
mock_create_empty.assert_any_call(cpu_file)
|
|
|
|
@patch(ns("fileops.create_empty"))
|
|
def test___init___battery_file(self, mock_create_empty):
|
|
mock_adb = MagicMock()
|
|
battery_file = "mybattery"
|
|
adb_control = adb_monitor.AdbControl(mock_adb, battery_file=battery_file)
|
|
self.assertEqual(adb_control.adb, mock_adb)
|
|
self.assertEqual(adb_control.battery_file, battery_file)
|
|
self.assertIsNone(adb_control.memory_file)
|
|
self.assertIsNone(adb_control.cpu_file)
|
|
self.assertEqual(adb_control.num_samples, 0)
|
|
self.assertIsNone(adb_control.collection_time_secs)
|
|
self.assertEqual(adb_control.sample_interval_ms, adb_monitor.DEFAULT_SAMPLE_INTERVAL_MS)
|
|
mock_create_empty.assert_called_once_with(battery_file)
|
|
|
|
@patch(ns("fileops.create_empty"))
|
|
def test___init___all_params(self, mock_create_empty):
|
|
mock_adb = MagicMock()
|
|
mock_logger = MagicMock()
|
|
battery_file = "mybattery"
|
|
memory_file = "mymemory"
|
|
cpu_file = "mycpu"
|
|
num_samples = 5
|
|
collection_time_secs = 10
|
|
sample_interval_ms = 25
|
|
adb_control = adb_monitor.AdbControl(
|
|
mock_adb, logger=mock_logger, battery_file=battery_file, memory_file=memory_file,
|
|
cpu_file=cpu_file, append_file=True, num_samples=num_samples, collection_time_secs=10,
|
|
sample_interval_ms=sample_interval_ms)
|
|
self.assertEqual(adb_control.adb, mock_adb)
|
|
self.assertEqual(adb_control.battery_file, battery_file)
|
|
self.assertEqual(adb_control.memory_file, memory_file)
|
|
self.assertEqual(adb_control.cpu_file, cpu_file)
|
|
self.assertEqual(adb_control.num_samples, 0)
|
|
self.assertEqual(adb_control.collection_time_secs, collection_time_secs)
|
|
self.assertEqual(adb_control.sample_interval_ms, sample_interval_ms)
|
|
mock_create_empty.assert_not_called()
|
|
|
|
@patch(ns("fileops.create_empty"))
|
|
def test___init___num_samples(self, mock_create_empty):
|
|
# pylint: disable=unused-argument
|
|
mock_adb = MagicMock()
|
|
battery_file = "mybattery"
|
|
num_samples = 5
|
|
adb_control = adb_monitor.AdbControl(mock_adb, battery_file=battery_file,
|
|
num_samples=num_samples)
|
|
self.assertEqual(adb_control.adb, mock_adb)
|
|
self.assertEqual(adb_control.battery_file, battery_file)
|
|
self.assertIsNone(adb_control.memory_file)
|
|
self.assertIsNone(adb_control.cpu_file)
|
|
self.assertEqual(adb_control.num_samples, num_samples)
|
|
self.assertIsNone(adb_control.collection_time_secs)
|
|
self.assertEqual(adb_control.sample_interval_ms, adb_monitor.DEFAULT_SAMPLE_INTERVAL_MS)
|
|
|
|
@patch(ns("AdbContinuousResourceMonitor"))
|
|
@patch(ns("AdbSampleBasedResourceMonitor"))
|
|
def test_start_all(self, mock_sample_monitor, mock_continuous_monitor):
|
|
mock_adb = MagicMock()
|
|
mock_logger = MagicMock()
|
|
battery_file = "mybattery"
|
|
memory_file = "mymemory"
|
|
cpu_file = "mycpu"
|
|
num_samples = 5
|
|
collection_time_secs = 10
|
|
sample_interval_ms = 25
|
|
adb_control = adb_monitor.AdbControl(
|
|
mock_adb, logger=mock_logger, battery_file=battery_file, memory_file=memory_file,
|
|
cpu_file=cpu_file, append_file=True, num_samples=num_samples,
|
|
collection_time_secs=collection_time_secs, sample_interval_ms=sample_interval_ms)
|
|
adb_control.start()
|
|
self.assertEqual(len(adb_control._all_threads), 3)
|
|
self.assertEqual(len(adb_control._sample_based_threads), 2)
|
|
mock_continuous_monitor.assert_called_once_with(
|
|
cpu_file, adb_control._should_stop, mock_adb.systrace_start, mock_adb.systrace_stop)
|
|
mock_continuous_monitor().start.assert_called_once()
|
|
self.assertEqual(mock_sample_monitor.call_count, 2)
|
|
mock_sample_monitor.assert_any_call(battery_file, adb_control._should_stop,
|
|
mock_adb.battery, 0, sample_interval_ms)
|
|
mock_sample_monitor.assert_any_call(memory_file, adb_control._should_stop, mock_adb.memory,
|
|
0, sample_interval_ms)
|
|
self.assertEqual(mock_sample_monitor().start.call_count, 2)
|
|
|
|
@patch(ns("AdbContinuousResourceMonitor"))
|
|
@patch(ns("AdbSampleBasedResourceMonitor"))
|
|
def test_start_cpu(self, mock_sample_monitor, mock_continuous_monitor):
|
|
mock_adb = MagicMock()
|
|
mock_logger = MagicMock()
|
|
cpu_file = "mycpu"
|
|
collection_time_secs = 10
|
|
adb_control = adb_monitor.AdbControl(mock_adb, logger=mock_logger, cpu_file=cpu_file,
|
|
append_file=True,
|
|
collection_time_secs=collection_time_secs)
|
|
adb_control.start()
|
|
self.assertEqual(len(adb_control._all_threads), 1)
|
|
self.assertEqual(len(adb_control._sample_based_threads), 0)
|
|
mock_continuous_monitor.assert_called_once_with(
|
|
cpu_file, adb_control._should_stop, mock_adb.systrace_start, mock_adb.systrace_stop)
|
|
mock_continuous_monitor().start.assert_called_once()
|
|
mock_sample_monitor.assert_not_called()
|
|
|
|
@patch(ns("AdbContinuousResourceMonitor"))
|
|
@patch(ns("AdbSampleBasedResourceMonitor"))
|
|
def test_start_battery(self, mock_sample_monitor, mock_continuous_monitor):
|
|
mock_adb = MagicMock()
|
|
mock_logger = MagicMock()
|
|
battery_file = "mybattery"
|
|
num_samples = 5
|
|
sample_interval_ms = 25
|
|
adb_control = adb_monitor.AdbControl(
|
|
mock_adb, logger=mock_logger, battery_file=battery_file, append_file=True,
|
|
num_samples=num_samples, sample_interval_ms=sample_interval_ms)
|
|
adb_control.start()
|
|
self.assertEqual(len(adb_control._all_threads), 1)
|
|
self.assertEqual(len(adb_control._sample_based_threads), 1)
|
|
mock_continuous_monitor.assert_not_called()
|
|
mock_sample_monitor.assert_called_once_with(battery_file, adb_control._should_stop,
|
|
mock_adb.battery, num_samples,
|
|
sample_interval_ms)
|
|
mock_sample_monitor().start.assert_called_once()
|
|
|
|
def test_stop(self):
|
|
mock_adb = MagicMock()
|
|
mock_logger = MagicMock()
|
|
battery_file = "mybattery"
|
|
num_samples = 5
|
|
sample_interval_ms = 25
|
|
adb_control = adb_monitor.AdbControl(
|
|
mock_adb, logger=mock_logger, battery_file=battery_file, append_file=True,
|
|
num_samples=num_samples, sample_interval_ms=sample_interval_ms)
|
|
adb_control._should_stop = MagicMock()
|
|
adb_control.wait = MagicMock()
|
|
adb_control.stop()
|
|
adb_control._should_stop.set.assert_called_once()
|
|
adb_control.wait.assert_called_once()
|
|
|
|
@patch(ns("AdbContinuousResourceMonitor"))
|
|
@patch(ns("AdbSampleBasedResourceMonitor"))
|
|
def test_wait(self, mock_sample_monitor, mock_continuous_monitor):
|
|
mock_adb = MagicMock()
|
|
mock_logger = MagicMock()
|
|
battery_file = "mybattery"
|
|
memory_file = "mymemory"
|
|
cpu_file = "mycpu"
|
|
num_samples = 5
|
|
collection_time_secs = 10
|
|
sample_interval_ms = 25
|
|
adb_control = adb_monitor.AdbControl(
|
|
mock_adb, logger=mock_logger, battery_file=battery_file, memory_file=memory_file,
|
|
cpu_file=cpu_file, append_file=True, num_samples=num_samples, collection_time_secs=10,
|
|
sample_interval_ms=sample_interval_ms)
|
|
mock_sample_monitor().exception = None
|
|
mock_continuous_monitor().exception = None
|
|
adb_control._should_stop = MagicMock()
|
|
adb_control.start()
|
|
adb_control.wait()
|
|
adb_control._should_stop.wait.assert_called_once_with(collection_time_secs)
|
|
adb_control._should_stop.set.assert_called_once()
|
|
mock_continuous_monitor().join.assert_called_once_with(adb_control._JOIN_TIMEOUT)
|
|
self.assertEqual(mock_sample_monitor().join.call_count, 2)
|
|
mock_sample_monitor().join.assert_called_with(adb_control._JOIN_TIMEOUT)
|
|
|
|
@patch(ns("AdbContinuousResourceMonitor"))
|
|
@patch(ns("AdbSampleBasedResourceMonitor"))
|
|
def test_wait_keyboard_interrupt(self, mock_sample_monitor, mock_continuous_monitor):
|
|
mock_adb = MagicMock()
|
|
mock_logger = MagicMock()
|
|
battery_file = "mybattery"
|
|
memory_file = "mymemory"
|
|
cpu_file = "mycpu"
|
|
num_samples = 5
|
|
collection_time_secs = 10
|
|
sample_interval_ms = 25
|
|
adb_control = adb_monitor.AdbControl(
|
|
mock_adb, logger=mock_logger, battery_file=battery_file, memory_file=memory_file,
|
|
cpu_file=cpu_file, append_file=True, num_samples=num_samples, collection_time_secs=10,
|
|
sample_interval_ms=sample_interval_ms)
|
|
mock_sample_monitor().exception = None
|
|
mock_continuous_monitor().exception = None
|
|
adb_control._should_stop = MagicMock()
|
|
adb_control._should_stop.wait.side_effect = KeyboardInterrupt()
|
|
adb_control.start()
|
|
adb_control.wait()
|
|
adb_control._should_stop.wait.assert_called_once_with(collection_time_secs)
|
|
adb_control._should_stop.set.assert_called_once()
|
|
mock_continuous_monitor().join.assert_called_once_with(adb_control._JOIN_TIMEOUT)
|
|
self.assertEqual(mock_sample_monitor().join.call_count, 2)
|
|
mock_sample_monitor().join.assert_called_with(adb_control._JOIN_TIMEOUT)
|
|
|
|
@patch(ns("AdbSampleBasedResourceMonitor"))
|
|
def test_wait_no_collection_time(self, mock_sample_monitor):
|
|
mock_adb = MagicMock()
|
|
mock_logger = MagicMock()
|
|
battery_file = "mybattery"
|
|
memory_file = "mymemory"
|
|
num_samples = 5
|
|
sample_interval_ms = 25
|
|
adb_control = adb_monitor.AdbControl(
|
|
mock_adb, logger=mock_logger, battery_file=battery_file, memory_file=memory_file,
|
|
append_file=True, num_samples=num_samples, sample_interval_ms=sample_interval_ms)
|
|
mock_sample_monitor().exception = None
|
|
adb_control._should_stop = MagicMock()
|
|
adb_control.start()
|
|
adb_control.wait()
|
|
adb_control._should_stop.wait.assert_not_called()
|
|
adb_control._should_stop.set.assert_called_once()
|
|
self.assertEqual(mock_sample_monitor().join.call_count, 4)
|
|
mock_sample_monitor().join.assert_called_with(adb_control._JOIN_TIMEOUT)
|
|
|
|
@patch(ns("AdbContinuousResourceMonitor"))
|
|
@patch(ns("AdbSampleBasedResourceMonitor"))
|
|
def test_wait_thread_exception(self, mock_sample_monitor, mock_continuous_monitor):
|
|
mock_adb = MagicMock()
|
|
mock_logger = MagicMock()
|
|
battery_file = "mybattery"
|
|
memory_file = "mymemory"
|
|
cpu_file = "mycpu"
|
|
num_samples = 5
|
|
collection_time_secs = 10
|
|
sample_interval_ms = 25
|
|
adb_control = adb_monitor.AdbControl(
|
|
mock_adb, logger=mock_logger, battery_file=battery_file, memory_file=memory_file,
|
|
cpu_file=cpu_file, append_file=True, num_samples=num_samples, collection_time_secs=10,
|
|
sample_interval_ms=sample_interval_ms)
|
|
mock_sample_monitor().exception = RuntimeError("my run time exception")
|
|
mock_continuous_monitor().exception = None
|
|
adb_control._should_stop = MagicMock()
|
|
adb_control.start()
|
|
with self.assertRaises(RuntimeError):
|
|
adb_control.wait()
|
|
adb_control._should_stop.wait.assert_called_once_with(collection_time_secs)
|
|
adb_control._should_stop.set.assert_called_once()
|
|
mock_continuous_monitor().join.assert_called_once_with(adb_control._JOIN_TIMEOUT)
|
|
self.assertEqual(mock_sample_monitor().join.call_count, 2)
|
|
mock_sample_monitor().join.assert_called_with(adb_control._JOIN_TIMEOUT)
|
|
|
|
|
|
class TestAdbResourceMonitor(unittest.TestCase):
|
|
@patch(ns("threading"))
|
|
def test_run(self, mock_threading):
|
|
# pylint: disable=unused-argument
|
|
output_file = "myfile"
|
|
should_stop = MagicMock()
|
|
arm = adb_monitor.AdbResourceMonitor(output_file, should_stop)
|
|
arm._do_monitoring = MagicMock()
|
|
arm.run()
|
|
self.assertEqual(arm._output_file, output_file)
|
|
arm._do_monitoring.assert_called_once()
|
|
self.assertIsNone(arm.exception)
|
|
should_stop.set.assert_not_called()
|
|
|
|
@patch(ns("threading"))
|
|
def test_run_exception(self, mock_threading):
|
|
# pylint: disable=unused-argument
|
|
output_file = "myfile"
|
|
should_stop = MagicMock()
|
|
exception = RuntimeError()
|
|
arm = adb_monitor.AdbResourceMonitor(output_file, should_stop)
|
|
arm._do_monitoring = MagicMock(side_effect=exception)
|
|
arm.run()
|
|
self.assertEqual(arm._output_file, output_file)
|
|
arm._do_monitoring.assert_called_once()
|
|
self.assertEqual(arm.exception, exception)
|
|
should_stop.set.assert_called_once()
|
|
|
|
|
|
class TestAdbSampleBasedResourceMonitor(unittest.TestCase):
|
|
@patch(ns("threading"))
|
|
def test__do_monitoring(self, mock_threading):
|
|
# pylint: disable=unused-argument
|
|
output_file = "myfile"
|
|
mock_should_stop = MagicMock()
|
|
mock_should_stop.is_set.return_value = False
|
|
mock_adb_cmd = MagicMock()
|
|
num_samples = 5
|
|
sample_interval_ms = 50
|
|
arm = adb_monitor.AdbSampleBasedResourceMonitor(output_file, mock_should_stop, mock_adb_cmd,
|
|
num_samples, sample_interval_ms)
|
|
arm._take_sample = MagicMock()
|
|
arm._do_monitoring()
|
|
self.assertEqual(mock_should_stop.is_set.call_count, num_samples + 1)
|
|
self.assertEqual(arm._take_sample.call_count, num_samples)
|
|
self.assertEqual(mock_should_stop.wait.call_count, num_samples - 1)
|
|
mock_should_stop.wait.assert_called_with(sample_interval_ms / 1000.0)
|
|
|
|
def _is_set(self):
|
|
self.num_collected += 1
|
|
return self.num_collected == self.num_samples
|
|
|
|
@patch(ns("threading"))
|
|
def test__do_monitoring_no_samples(self, mock_threading):
|
|
# pylint: disable=unused-argument
|
|
output_file = "myfile"
|
|
mock_should_stop = MagicMock()
|
|
self.num_collected = 0 # pylint: disable=attribute-defined-outside-init
|
|
self.num_samples = 5 # pylint: disable=attribute-defined-outside-init
|
|
mock_should_stop.is_set = self._is_set
|
|
mock_adb_cmd = MagicMock()
|
|
num_samples = 0
|
|
sample_interval_ms = 50
|
|
arm = adb_monitor.AdbSampleBasedResourceMonitor(output_file, mock_should_stop, mock_adb_cmd,
|
|
num_samples, sample_interval_ms)
|
|
arm._take_sample = MagicMock()
|
|
arm._do_monitoring()
|
|
self.assertEqual(arm._take_sample.call_count, self.num_samples - 1)
|
|
self.assertEqual(mock_should_stop.wait.call_count, self.num_samples - 2)
|
|
|
|
@patch(ns("threading"))
|
|
def test__take_sample(self, mock_threading):
|
|
# pylint: disable=unused-argument
|
|
output_file = "myfile"
|
|
mock_should_stop = MagicMock()
|
|
self.num_collected = 0 # pylint: disable=attribute-defined-outside-init
|
|
self.num_samples = 5 # pylint: disable=attribute-defined-outside-init
|
|
mock_should_stop.is_set = self._is_set
|
|
mock_adb_cmd = MagicMock()
|
|
arm = adb_monitor.AdbSampleBasedResourceMonitor(output_file, mock_should_stop, mock_adb_cmd,
|
|
10, 20)
|
|
arm._take_sample(2)
|
|
mock_adb_cmd.assert_called_once_with(output_file=output_file, append_file=True)
|
|
|
|
|
|
class TestAdbContinuousResourceMonitor(unittest.TestCase):
|
|
@patch(ns("threading"))
|
|
def test__do_monitoring(self, mock_threading):
|
|
# pylint: disable=unused-argument
|
|
output_file = "myfile"
|
|
mock_should_stop = MagicMock()
|
|
mock_adb_start_cmd = MagicMock()
|
|
mock_adb_stop_cmd = MagicMock()
|
|
acrm = adb_monitor.AdbContinuousResourceMonitor(output_file, mock_should_stop,
|
|
mock_adb_start_cmd, mock_adb_stop_cmd)
|
|
acrm._do_monitoring()
|
|
mock_should_stop.wait.assert_called_once()
|
|
mock_adb_start_cmd.assert_called_with(output_file=output_file)
|
|
mock_adb_stop_cmd.assert_called_with(output_file=output_file)
|