# Copyright (c) 2013 The Chromium OS Authors. All rights reserved.
|
# Use of this source code is governed by a BSD-style license that can be
|
# found in the LICENSE file.
|
|
import logging
|
import unittest
|
import re
|
import csv
|
import common
|
import os
|
|
from itertools import imap
|
from autotest_lib.server.cros import resource_monitor
|
from autotest_lib.server.hosts import abstract_ssh
|
from autotest_lib.server import utils
|
|
class HostMock(abstract_ssh.AbstractSSHHost):
|
"""Mocks a host object."""
|
|
TOP_PID = '1234'
|
|
def _initialize(self, test_env):
|
self.top_is_running = False
|
|
# Keep track of whether the top raw output file exists on the system,
|
# and if it does, where it is.
|
self.top_output_file_path = None
|
|
# Keep track of whether the raw top output file is currently being
|
# written to by top.
|
self.top_output_file_is_open = False
|
self.test_env = test_env
|
|
|
def get_file(self, src, dest):
|
pass
|
|
|
def called_unsupported_command(self, command):
|
"""Raises assertion error when called.
|
|
@param command string the unsupported command called.
|
|
"""
|
raise AssertionError(
|
"ResourceMonitor called unsupported command %s" % command)
|
|
|
def _process_top(self, cmd_args, cmd_line):
|
"""Process top command.
|
|
@param cmd_args string_list of command line args.
|
@param cmd_line string the command to run.
|
|
"""
|
self.test_env.assertFalse(self.top_is_running,
|
msg="Top must not already be running.")
|
self.test_env.assertFalse(self.top_output_file_is_open,
|
msg="The top output file should not be being written "
|
"to before top is started")
|
self.test_env.assertIsNone(self.top_output_file_path,
|
msg="The top output file should not exist"
|
"before top is started")
|
try:
|
self.redirect_index = cmd_args.index(">")
|
self.top_output_file_path = cmd_args[self.redirect_index + 1]
|
except ValueError, IndexError:
|
self.called_unsupported_command(cmd_line)
|
|
self.top_is_running = True
|
self.top_output_file_is_open = True
|
|
return HostMock.TOP_PID
|
|
|
def _process_kill(self, cmd_args, cmd_line):
|
"""Process kill command.
|
|
@param cmd_args string_list of command line args.
|
@param cmd_line string the command to run.
|
|
"""
|
try:
|
if cmd_args[1].startswith('-'):
|
pid_to_kill = cmd_args[2]
|
else:
|
pid_to_kill = cmd_args[1]
|
except IndexError:
|
self.called_unsupported_command(cmd_line)
|
|
self.test_env.assertEqual(pid_to_kill, HostMock.TOP_PID,
|
msg="Wrong pid (%r) killed . Top pid is %r." % (pid_to_kill,
|
HostMock.TOP_PID))
|
self.test_env.assertTrue(self.top_is_running,
|
msg="Top must be running before we try to kill it")
|
|
self.top_is_running = False
|
self.top_output_file_is_open = False
|
|
|
def _process_rm(self, cmd_args, cmd_line):
|
"""Process rm command.
|
|
@param cmd_args string list list of command line args.
|
@param cmd_line string the command to run.
|
|
"""
|
try:
|
if cmd_args[1].startswith('-'):
|
file_to_rm = cmd_args[2]
|
else:
|
file_to_rm = cmd_args[1]
|
except IndexError:
|
self.called_unsupported_command(cmd_line)
|
|
self.test_env.assertEqual(file_to_rm, self.top_output_file_path,
|
msg="Tried to remove file that is not the top output file.")
|
self.test_env.assertFalse(self.top_output_file_is_open,
|
msg="Tried to remove top output file while top is still "
|
"writing to it.")
|
self.test_env.assertFalse(self.top_is_running,
|
msg="Top was still running when we tried to remove"
|
"the top output file.")
|
self.test_env.assertIsNotNone(self.top_output_file_path)
|
|
self.top_output_file_path = None
|
|
|
def _run_single_cmd(self, cmd_line, *args, **kwargs):
|
"""Run a single command on host.
|
|
@param cmd_line command to run on the host.
|
|
"""
|
# Make the input a little nicer.
|
cmd_line = cmd_line.strip()
|
cmd_line = re.sub(">", " > ", cmd_line)
|
|
cmd_args = re.split("\s+", cmd_line)
|
self.test_env.assertTrue(len(cmd_args) >= 1)
|
command = cmd_args[0]
|
if (command == "top"):
|
return self._process_top(cmd_args, cmd_line)
|
elif (command == "kill"):
|
return self._process_kill(cmd_args, cmd_line)
|
elif(command == "rm"):
|
return self._process_rm(cmd_args, cmd_line)
|
else:
|
logging.warning("Called unemulated command %r", cmd_line)
|
return None
|
|
|
def run(self, cmd_line, *args, **kwargs):
|
"""Run command(s) on host.
|
|
@param cmd_line command to run on the host.
|
@return CmdResult object.
|
|
"""
|
cmds = re.split("&&", cmd_line)
|
for cmd in cmds:
|
self._run_single_cmd(cmd)
|
return utils.CmdResult(exit_status=0)
|
|
|
def run_background(self, cmd_line, *args, **kwargs):
|
"""Run command in background on host.
|
|
@param cmd_line command to run on the host.
|
|
"""
|
return self._run_single_cmd(cmd_line, args, kwargs)
|
|
|
def is_monitoring(self):
|
"""Return true iff host is currently running top and writing output
|
to a file.
|
"""
|
return self.top_is_running and self.top_output_file_is_open and (
|
self.top_output_file_path is not None)
|
|
|
def monitoring_stopped(self):
|
"""Return true iff host is not running top and all top output files are
|
closed.
|
"""
|
return not self.is_monitoring()
|
|
|
class ResourceMonitorTest(unittest.TestCase):
|
"""Tests the non-trivial functionality of ResourceMonitor."""
|
|
def setUp(self):
|
self.topoutfile = '/tmp/resourcemonitorunittest-1234'
|
self.monitor_period = 1
|
self.rm_conf = resource_monitor.ResourceMonitorConfig(
|
monitor_period=self.monitor_period,
|
rawresult_output_filename=self.topoutfile)
|
self.host = HostMock(self)
|
|
|
def test_normal_operation(self):
|
"""Checks that normal (i.e. no exceptions, etc.) execution works."""
|
self.assertFalse(self.host.is_monitoring())
|
with resource_monitor.ResourceMonitor(self.host, self.rm_conf) as rm:
|
self.assertFalse(self.host.is_monitoring())
|
for i in range(3):
|
rm.start()
|
self.assertTrue(self.host.is_monitoring())
|
rm.stop()
|
self.assertTrue(self.host.monitoring_stopped())
|
self.assertTrue(self.host.monitoring_stopped())
|
|
|
def test_forgot_to_stop_monitor(self):
|
"""Checks that resource monitor is cleaned up even if user forgets to
|
explicitly stop it.
|
"""
|
self.assertFalse(self.host.is_monitoring())
|
with resource_monitor.ResourceMonitor(self.host, self.rm_conf) as rm:
|
self.assertFalse(self.host.is_monitoring())
|
rm.start()
|
self.assertTrue(self.host.is_monitoring())
|
self.assertTrue(self.host.monitoring_stopped())
|
|
|
def test_unexpected_interruption_while_monitoring(self):
|
"""Checks that monitor is cleaned up upon unexpected interrupt."""
|
self.assertFalse(self.host.is_monitoring())
|
|
with resource_monitor.ResourceMonitor(self.host, self.rm_conf) as rm:
|
self.assertFalse(self.host.is_monitoring())
|
rm.start()
|
self.assertTrue(self.host.is_monitoring())
|
raise KeyboardInterrupt
|
|
self.assertTrue(self.host.monitoring_stopped())
|
|
|
class ResourceMonitorResultTest(unittest.TestCase):
|
"""Functional tests for ResourceMonitorParsedResult."""
|
|
def setUp(self):
|
self._res_dir = os.path.join(
|
os.path.dirname(os.path.realpath(__file__)),
|
'res_resource_monitor')
|
|
|
def run_with_test_data(self, testdata_file, testans_file):
|
"""Parses testdata_file with the parses, and checks that results
|
are the same as those in testans_file.
|
|
@param testdata_file string filename containing top output to test.
|
@param testans_file string filename containing answers to the test.
|
|
"""
|
parsed_results = resource_monitor.ResourceMonitorParsedResult(
|
testdata_file)
|
with open(testans_file, "rb") as testans:
|
csvreader = csv.reader(testans)
|
columns = csvreader.next()
|
self.assertEqual(list(columns),
|
resource_monitor.ResourceMonitorParsedResult._columns)
|
utils_over_time = []
|
for util_val in imap(
|
resource_monitor.
|
ResourceMonitorParsedResult.UtilValues._make,
|
csvreader):
|
utils_over_time.append(util_val)
|
self.assertEqual(utils_over_time, parsed_results._utils_over_time)
|
|
|
def test_full_data(self):
|
"""General test with many possible changes to input."""
|
self.run_with_test_data(
|
os.path.join(self._res_dir, 'top_test_data.txt'),
|
os.path.join(self._res_dir, 'top_test_data_ans.csv'))
|
|
|
def test_whitespace_ridden(self):
|
"""Tests resilience to arbitrary whitespace characters between fields"""
|
self.run_with_test_data(
|
os.path.join(self._res_dir, 'top_whitespace_ridden.txt'),
|
os.path.join(self._res_dir, 'top_whitespace_ridden_ans.csv'))
|
|
|
def test_field_order_changed(self):
|
"""Tests resilience to changes in the order of fields
|
(for e.g, if the Mem free/used fields change orders in the input).
|
"""
|
self.run_with_test_data(
|
os.path.join(self._res_dir, 'top_field_order_changed.txt'),
|
os.path.join(self._res_dir, 'top_field_order_changed_ans.csv'))
|
|
|
if __name__ == '__main__':
|
unittest.main()
|
