#!/usr/bin/python
|
#pylint: disable-msg=C0111
|
"""Unit Tests for autotest.client.common_lib.test"""
|
|
__author__ = 'gps@google.com (Gregory P. Smith)'
|
|
import json
|
import tempfile
|
import unittest
|
import common
|
from autotest_lib.client.common_lib import test
|
from autotest_lib.client.common_lib.test_utils import mock
|
|
class TestTestCase(unittest.TestCase):
|
class _neutered_base_test(test.base_test):
|
"""A child class of base_test to avoid calling the constructor."""
|
def __init__(self, *args, **kwargs):
|
class MockJob(object):
|
pass
|
class MockProfilerManager(object):
|
def active(self):
|
return False
|
def present(self):
|
return True
|
self.job = MockJob()
|
self.job.default_profile_only = False
|
self.job.profilers = MockProfilerManager()
|
self.job.test_retry = 0
|
self.job.fast = False
|
self._new_keyval = False
|
self.iteration = 0
|
self.tagged_testname = 'neutered_base_test'
|
self.before_iteration_hooks = []
|
self.after_iteration_hooks = []
|
|
|
def setUp(self):
|
self.god = mock.mock_god()
|
self.test = self._neutered_base_test()
|
|
|
def tearDown(self):
|
self.god.unstub_all()
|
|
|
|
class Test_base_test_execute(TestTestCase):
|
# Test the various behaviors of the base_test.execute() method.
|
def setUp(self):
|
TestTestCase.setUp(self)
|
self.god.stub_function(self.test, 'run_once_profiling')
|
self.god.stub_function(self.test, 'postprocess')
|
self.god.stub_function(self.test, 'process_failed_constraints')
|
|
|
def test_call_run_once(self):
|
# setup
|
self.god.stub_function(self.test, 'drop_caches_between_iterations')
|
self.god.stub_function(self.test, 'run_once')
|
self.god.stub_function(self.test, 'postprocess_iteration')
|
self.god.stub_function(self.test, 'analyze_perf_constraints')
|
before_hook = self.god.create_mock_function('before_hook')
|
after_hook = self.god.create_mock_function('after_hook')
|
self.test.register_before_iteration_hook(before_hook)
|
self.test.register_after_iteration_hook(after_hook)
|
|
# tests the test._call_run_once implementation
|
self.test.drop_caches_between_iterations.expect_call()
|
before_hook.expect_call(self.test)
|
self.test.run_once.expect_call(1, 2, arg='val')
|
self.test.postprocess_iteration.expect_call()
|
self.test.analyze_perf_constraints.expect_call([])
|
after_hook.expect_call(self.test)
|
self.test._call_run_once([], False, None, (1, 2), {'arg': 'val'})
|
self.god.check_playback()
|
|
|
def test_call_run_once_with_exception(self):
|
# setup
|
self.god.stub_function(self.test, 'drop_caches_between_iterations')
|
self.god.stub_function(self.test, 'run_once')
|
before_hook = self.god.create_mock_function('before_hook')
|
after_hook = self.god.create_mock_function('after_hook')
|
self.test.register_before_iteration_hook(before_hook)
|
self.test.register_after_iteration_hook(after_hook)
|
error = Exception('fail')
|
|
# tests the test._call_run_once implementation
|
self.test.drop_caches_between_iterations.expect_call()
|
before_hook.expect_call(self.test)
|
self.test.run_once.expect_call(1, 2, arg='val').and_raises(error)
|
after_hook.expect_call(self.test)
|
try:
|
self.test._call_run_once([], False, None, (1, 2), {'arg': 'val'})
|
except:
|
pass
|
self.god.check_playback()
|
|
|
def _expect_call_run_once(self):
|
self.test._call_run_once.expect_call((), False, None, (), {})
|
|
|
def test_execute_test_length(self):
|
# test that test_length overrides iterations and works.
|
self.god.stub_function(self.test, '_call_run_once')
|
|
self._expect_call_run_once()
|
self._expect_call_run_once()
|
self._expect_call_run_once()
|
self.test.run_once_profiling.expect_call(None)
|
self.test.postprocess.expect_call()
|
self.test.process_failed_constraints.expect_call()
|
|
fake_time = iter(xrange(4)).next
|
self.test.execute(iterations=1, test_length=3, _get_time=fake_time)
|
self.god.check_playback()
|
|
|
def test_execute_iterations(self):
|
# test that iterations works.
|
self.god.stub_function(self.test, '_call_run_once')
|
|
iterations = 2
|
for _ in range(iterations):
|
self._expect_call_run_once()
|
self.test.run_once_profiling.expect_call(None)
|
self.test.postprocess.expect_call()
|
self.test.process_failed_constraints.expect_call()
|
|
self.test.execute(iterations=iterations)
|
self.god.check_playback()
|
|
|
def _mock_calls_for_execute_no_iterations(self):
|
self.test.run_once_profiling.expect_call(None)
|
self.test.postprocess.expect_call()
|
self.test.process_failed_constraints.expect_call()
|
|
|
def test_execute_iteration_zero(self):
|
# test that iterations=0 works.
|
self._mock_calls_for_execute_no_iterations()
|
|
self.test.execute(iterations=0)
|
self.god.check_playback()
|
|
|
def test_execute_profile_only(self):
|
# test that profile_only=True works.
|
self.god.stub_function(self.test, 'drop_caches_between_iterations')
|
self.test.drop_caches_between_iterations.expect_call()
|
self.test.run_once_profiling.expect_call(None)
|
self.test.drop_caches_between_iterations.expect_call()
|
self.test.run_once_profiling.expect_call(None)
|
self.test.postprocess.expect_call()
|
self.test.process_failed_constraints.expect_call()
|
self.test.execute(profile_only=True, iterations=2)
|
self.god.check_playback()
|
|
|
def test_execute_default_profile_only(self):
|
# test that profile_only=True works.
|
self.god.stub_function(self.test, 'drop_caches_between_iterations')
|
for _ in xrange(3):
|
self.test.drop_caches_between_iterations.expect_call()
|
self.test.run_once_profiling.expect_call(None)
|
self.test.postprocess.expect_call()
|
self.test.process_failed_constraints.expect_call()
|
self.test.job.default_profile_only = True
|
self.test.execute(iterations=3)
|
self.god.check_playback()
|
|
|
def test_execute_postprocess_profiled_false(self):
|
# test that postprocess_profiled_run=False works
|
self.god.stub_function(self.test, '_call_run_once')
|
|
self.test._call_run_once.expect_call((), False, False, (), {})
|
self.test.run_once_profiling.expect_call(False)
|
self.test.postprocess.expect_call()
|
self.test.process_failed_constraints.expect_call()
|
|
self.test.execute(postprocess_profiled_run=False, iterations=1)
|
self.god.check_playback()
|
|
|
def test_execute_postprocess_profiled_true(self):
|
# test that postprocess_profiled_run=True works
|
self.god.stub_function(self.test, '_call_run_once')
|
|
self.test._call_run_once.expect_call((), False, True, (), {})
|
self.test.run_once_profiling.expect_call(True)
|
self.test.postprocess.expect_call()
|
self.test.process_failed_constraints.expect_call()
|
|
self.test.execute(postprocess_profiled_run=True, iterations=1)
|
self.god.check_playback()
|
|
|
def test_output_single_perf_value(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
|
self.test.output_perf_value("Test", 1, units="ms", higher_is_better=True)
|
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {"Test": {"summary": {"units": "ms", "type": "scalar",
|
"value": 1, "improvement_direction": "up"}}}
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
|
def test_output_single_perf_value_twice(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
|
self.test.output_perf_value("Test", 1, units="ms", higher_is_better=True)
|
self.test.output_perf_value("Test", 2, units="ms", higher_is_better=True)
|
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {"Test": {"summary": {"units": "ms",
|
"type": "list_of_scalar_values", "values": [1, 2],
|
"improvement_direction": "up"}}}
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
|
def test_output_single_perf_value_three_times(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
|
self.test.output_perf_value("Test", 1, units="ms",
|
higher_is_better=True)
|
self.test.output_perf_value("Test", 2, units="ms", higher_is_better=True)
|
self.test.output_perf_value("Test", 3, units="ms", higher_is_better=True)
|
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {"Test": {"summary": {"units": "ms",
|
"type": "list_of_scalar_values", "values": [1, 2, 3],
|
"improvement_direction": "up"}}}
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
|
def test_output_list_perf_value(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
|
self.test.output_perf_value("Test", [1, 2, 3], units="ms",
|
higher_is_better=False)
|
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {"Test": {"summary": {"units": "ms",
|
"type": "list_of_scalar_values", "values": [1, 2, 3],
|
"improvement_direction": "down"}}}
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
|
def test_output_single_then_list_perf_value(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
self.test.output_perf_value("Test", 1, units="ms",
|
higher_is_better=False)
|
self.test.output_perf_value("Test", [4, 3, 2], units="ms",
|
higher_is_better=False)
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {"Test": {"summary": {"units": "ms",
|
"type": "list_of_scalar_values",
|
"values": [1, 4, 3, 2],
|
"improvement_direction": "down"}}}
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
|
def test_output_list_then_list_perf_value(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
self.test.output_perf_value("Test", [1, 2, 3], units="ms",
|
higher_is_better=False)
|
self.test.output_perf_value("Test", [4, 3, 2], units="ms",
|
higher_is_better=False)
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {"Test": {"summary": {"units": "ms",
|
"type": "list_of_scalar_values",
|
"values": [1, 2, 3, 4, 3, 2],
|
"improvement_direction": "down"}}}
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
|
def test_output_single_perf_value_input_string(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
|
self.test.output_perf_value("Test", u'-0.34', units="ms",
|
higher_is_better=True)
|
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {"Test": {"summary": {"units": "ms", "type": "scalar",
|
"value": -0.34, "improvement_direction": "up"}}}
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
|
def test_output_single_perf_value_input_list_of_string(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
|
self.test.output_perf_value("Test", [0, u'-0.34', 1], units="ms",
|
higher_is_better=True)
|
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {"Test": {"summary": {"units": "ms",
|
"type": "list_of_scalar_values",
|
"values": [0, -0.34, 1],
|
"improvement_direction": "up"}}}
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
def test_output_list_then_replace_list_perf_value(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
self.test.output_perf_value("Test", [1, 2, 3], units="ms",
|
higher_is_better=False)
|
self.test.output_perf_value("Test", [4, 5, 6], units="ms",
|
higher_is_better=False,
|
replace_existing_values=True)
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {"Test": {"summary": {"units": "ms",
|
"type": "list_of_scalar_values",
|
"values": [4, 5, 6],
|
"improvement_direction": "down"}}}
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
def test_output_single_then_replace_list_perf_value(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
self.test.output_perf_value("Test", 3, units="ms",
|
higher_is_better=False)
|
self.test.output_perf_value("Test", [4, 5, 6], units="ms",
|
higher_is_better=False,
|
replace_existing_values=True)
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {"Test": {"summary": {"units": "ms",
|
"type": "list_of_scalar_values",
|
"values": [4, 5, 6],
|
"improvement_direction": "down"}}}
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
def test_output_list_then_replace_single_perf_value(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
self.test.output_perf_value("Test", [1,2,3], units="ms",
|
higher_is_better=False)
|
self.test.output_perf_value("Test", 4, units="ms",
|
higher_is_better=False,
|
replace_existing_values=True)
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {"Test": {"summary": {"units": "ms",
|
"type": "scalar",
|
"value": 4,
|
"improvement_direction": "down"}}}
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
def test_output_single_then_replace_single_perf_value(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
self.test.output_perf_value("Test", 1, units="ms",
|
higher_is_better=False)
|
self.test.output_perf_value("Test", 2, units="ms",
|
higher_is_better=False,
|
replace_existing_values=True)
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {"Test": {"summary": {"units": "ms",
|
"type": "scalar",
|
"value": 2,
|
"improvement_direction": "down"}}}
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
def test_output_perf_then_replace_certain_perf_value(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
self.test.output_perf_value("Test1", 1, units="ms",
|
higher_is_better=False)
|
self.test.output_perf_value("Test2", 2, units="ms",
|
higher_is_better=False)
|
self.test.output_perf_value("Test3", 3, units="ms",
|
higher_is_better=False)
|
self.test.output_perf_value("Test2", -1, units="ms",
|
higher_is_better=False,
|
replace_existing_values=True)
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {"Test1": {"summary":
|
{"units": "ms",
|
"type": "scalar",
|
"value": 1,
|
"improvement_direction": "down"}},
|
"Test2": {"summary":
|
{"units": "ms",
|
"type": "scalar",
|
"value": -1,
|
"improvement_direction": "down"}},
|
"Test3": {"summary":
|
{"units": "ms",
|
"type": "scalar",
|
"value": 3,
|
"improvement_direction": "down"}}}
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
def test_chart_supplied(self):
|
self.test.resultsdir = tempfile.mkdtemp()
|
|
test_data = [("tcp_tx", "ch006_mode11B_none", "BT_connected_but_not_streaming", 0),
|
("tcp_tx", "ch006_mode11B_none", "BT_streaming_audiofile", 5),
|
("tcp_tx", "ch006_mode11B_none", "BT_disconnected_again", 0),
|
("tcp_rx", "ch006_mode11B_none", "BT_connected_but_not_streaming", 0),
|
("tcp_rx", "ch006_mode11B_none", "BT_streaming_audiofile", 8),
|
("tcp_rx", "ch006_mode11B_none", "BT_disconnected_again", 0),
|
("udp_tx", "ch006_mode11B_none", "BT_connected_but_not_streaming", 0),
|
("udp_tx", "ch006_mode11B_none", "BT_streaming_audiofile", 6),
|
("udp_tx", "ch006_mode11B_none", "BT_disconnected_again", 0),
|
("udp_rx", "ch006_mode11B_none", "BT_connected_but_not_streaming", 0),
|
("udp_rx", "ch006_mode11B_none", "BT_streaming_audiofile", 8),
|
("udp_rx", "ch006_mode11B_none", "BT_streaming_audiofile", 9),
|
("udp_rx", "ch006_mode11B_none", "BT_disconnected_again", 0)]
|
|
|
for (config_tag, ap_config_tag, bt_tag, drop) in test_data:
|
self.test.output_perf_value(config_tag + '_' + bt_tag + '_drop',
|
drop, units='percent_drop',
|
higher_is_better=False,
|
graph=ap_config_tag + '_drop')
|
f = open(self.test.resultsdir + "/results-chart.json")
|
expected_result = {
|
"ch006_mode11B_none_drop": {
|
"udp_tx_BT_streaming_audiofile_drop": {
|
"units": "percent_drop",
|
"type": "scalar",
|
"value": 6.0,
|
"improvement_direction": "down"
|
},
|
"udp_rx_BT_disconnected_again_drop": {
|
"units": "percent_drop",
|
"type": "scalar",
|
"value": 0.0,
|
"improvement_direction": "down"
|
},
|
"tcp_tx_BT_disconnected_again_drop": {
|
"units": "percent_drop",
|
"type": "scalar",
|
"value": 0.0,
|
"improvement_direction": "down"
|
},
|
"tcp_rx_BT_streaming_audiofile_drop": {
|
"units": "percent_drop",
|
"type": "scalar",
|
"value": 8.0,
|
"improvement_direction": "down"
|
},
|
"udp_tx_BT_connected_but_not_streaming_drop": {
|
"units": "percent_drop",
|
"type": "scalar",
|
"value": 0.0,
|
"improvement_direction": "down"
|
},
|
"tcp_tx_BT_connected_but_not_streaming_drop": {
|
"units": "percent_drop",
|
"type": "scalar",
|
"value": 0.0,
|
"improvement_direction": "down"
|
},
|
"udp_tx_BT_disconnected_again_drop": {
|
"units": "percent_drop",
|
"type": "scalar",
|
"value": 0.0,
|
"improvement_direction": "down"
|
},
|
"tcp_tx_BT_streaming_audiofile_drop": {
|
"units": "percent_drop",
|
"type": "scalar",
|
"value": 5.0,
|
"improvement_direction": "down"
|
},
|
"tcp_rx_BT_connected_but_not_streaming_drop": {
|
"units": "percent_drop",
|
"type": "scalar",
|
"value": 0.0,
|
"improvement_direction": "down"
|
},
|
"udp_rx_BT_connected_but_not_streaming_drop": {
|
"units": "percent_drop",
|
"type": "scalar",
|
"value": 0.0,
|
"improvement_direction": "down"
|
},
|
"udp_rx_BT_streaming_audiofile_drop": {
|
"units": "percent_drop",
|
"type": "list_of_scalar_values",
|
"values": [
|
8.0,
|
9.0
|
],
|
"improvement_direction": "down"
|
},
|
"tcp_rx_BT_disconnected_again_drop": {
|
"units": "percent_drop",
|
"type": "scalar",
|
"value": 0.0,
|
"improvement_direction": "down"
|
}
|
}
|
}
|
self.maxDiff = None
|
self.assertDictEqual(expected_result, json.loads(f.read()))
|
|
if __name__ == '__main__':
|
unittest.main()
|
