1#!/usr/bin/python2
2
3# Copyright 2016 The Chromium OS Authors. All rights reserved.
4# Use of this source code is governed by a BSD-style license that can be
5# found in the LICENSE file.
6"""Unit tests for the process_hot_functions module."""
7
8from process_hot_functions import HotFunctionsProcessor, ParseArguments
9
10import mock
11import os
12import shutil
13import tempfile
14import unittest
15
16
17class ParseArgumentsTest(unittest.TestCase):
18  """Test class for command line argument parsing."""
19
20  def __init__(self, *args, **kwargs):
21    super(ParseArgumentsTest, self).__init__(*args, **kwargs)
22
23  def testParseArguments(self):
24    arguments = \
25      ['-p', 'dummy_pprof', '-c', 'dummy_common', '-e', 'dummy_extra', '-w',
26       'dummy_cwp']
27    options = ParseArguments(arguments)
28
29    self.assertEqual(options.pprof_path, 'dummy_pprof')
30    self.assertEqual(options.cwp_hot_functions_file, 'dummy_cwp')
31    self.assertEqual(options.common_functions_path, 'dummy_common')
32    self.assertEqual(options.extra_cwp_functions_file, 'dummy_extra')
33
34  @mock.patch('sys.exit')
35  def testDeathParseArguments(self, sys_exit_method):
36    self.assertFalse(sys_exit_method.called)
37    ParseArguments([])
38    self.assertTrue(sys_exit_method.called)
39    self.assertNotEqual(sys_exit_method.return_value, 0)
40
41
42class HotFunctionsProcessorTest(unittest.TestCase):
43  """Test class for HotFunctionsProcessor class."""
44
45  def __init__(self, *args, **kwargs):
46    super(HotFunctionsProcessorTest, self).__init__(*args, **kwargs)
47    self._pprof_path = 'testdata/input/pprof'
48    self._cwp_functions_file = 'testdata/input/cwp_functions_file.csv'
49    self._cwp_functions_file_parsing = \
50      'testdata/input/parse_cwp_statistics.csv'
51    self._common_functions_path = ''
52    self._expected_common_functions_path = 'testdata/expected/pprof_common'
53    self._extra_cwp_functions_file = ''
54    self._cwp_function_groups_file = 'testdata/input/cwp_function_groups'
55    self._cwp_function_groups_statistics_file = 'dummy'
56    self._cwp_function_groups_file_prefix = 'dummy'
57
58  def _CreateHotFunctionsProcessor(self,
59                                   extra_cwp_functions_file,
60                                   cwp_function_groups_file=None,
61                                   cwp_function_groups_statistics_file=None,
62                                   cwp_function_groups_file_prefix=None):
63    return HotFunctionsProcessor(self._pprof_path, self._cwp_functions_file,
64                                 self._common_functions_path,
65                                 extra_cwp_functions_file,
66                                 cwp_function_groups_file,
67                                 cwp_function_groups_statistics_file,
68                                 cwp_function_groups_file_prefix)
69
70  def checkFileContents(self, file_name, expected_content_lines):
71    with open(file_name, 'r') as input_file:
72      result_content_lines = input_file.readlines()
73    self.assertListEqual(expected_content_lines, result_content_lines)
74
75  @mock.patch.object(HotFunctionsProcessor, 'ExtractCommonFunctions')
76  @mock.patch.object(HotFunctionsProcessor, 'ExtractExtraFunctions')
77  @mock.patch.object(HotFunctionsProcessor, 'GroupExtraFunctions')
78  def testProcessHotFunctionsNoGroupping(self, group_functions_method,
79                                         extra_functions_method,
80                                         common_functions_method):
81    hot_functions_processor = self._CreateHotFunctionsProcessor(
82        self._extra_cwp_functions_file)
83
84    hot_functions_processor.ProcessHotFunctions()
85
86    self.assertTrue(common_functions_method.called)
87    self.assertTrue(extra_functions_method.called)
88    self.assertEqual(common_functions_method.call_count, 1)
89    self.assertEqual(extra_functions_method.call_count, 1)
90    self.assertFalse(group_functions_method.called)
91
92  @mock.patch.object(HotFunctionsProcessor, 'ExtractCommonFunctions')
93  @mock.patch.object(HotFunctionsProcessor, 'ExtractExtraFunctions')
94  @mock.patch.object(HotFunctionsProcessor, 'GroupExtraFunctions')
95  def testProcessHotFunctionsGroupping(self, group_functions_method,
96                                       extra_functions_method,
97                                       common_functions_method):
98    hot_functions_processor = self._CreateHotFunctionsProcessor(
99        self._extra_cwp_functions_file, self._cwp_function_groups_file,
100        self._cwp_function_groups_statistics_file,
101        self._cwp_function_groups_file_prefix)
102
103    hot_functions_processor.ProcessHotFunctions()
104
105    self.assertTrue(common_functions_method.called)
106    self.assertTrue(extra_functions_method.called)
107    self.assertEqual(common_functions_method.call_count, 1)
108    self.assertEqual(extra_functions_method.call_count, 1)
109    self.assertTrue(group_functions_method.called)
110    self.assertEqual(group_functions_method.call_count, 1)
111
112  def testParseCWPStatistics(self):
113    cwp_statistics = {'dummy_method1,dummy_file1': ('dummy_object1,1', 0),
114                      'dummy_method2,dummy_file2': ('dummy_object2,2', 0),
115                      'dummy_method3,dummy_file3': ('dummy_object3,3', 0),
116                      'dummy_method4,dummy_file4': ('dummy_object4,4', 0)}
117    hot_functions_processor = self._CreateHotFunctionsProcessor(
118        self._extra_cwp_functions_file)
119    result = hot_functions_processor.ParseCWPStatistics(
120        self._cwp_functions_file_parsing)
121
122    self.assertDictEqual(result, cwp_statistics)
123
124  def testExtractCommonFunctions(self):
125    hot_functions_processor = self._CreateHotFunctionsProcessor(
126        self._extra_cwp_functions_file)
127    common_functions_path = tempfile.mkdtemp()
128    hot_functions_processor.ExtractCommonFunctions(self._pprof_path,
129                                                   common_functions_path,
130                                                   self._cwp_functions_file)
131    expected_files = \
132      [os.path.join(self._expected_common_functions_path, expected_file)
133       for expected_file in os.listdir(self._expected_common_functions_path)]
134    result_files = \
135      [os.path.join(common_functions_path, result_file)
136       for result_file in os.listdir(common_functions_path)]
137
138    expected_files.sort()
139    result_files.sort()
140
141    for expected_file_name, result_file_name in \
142      zip(expected_files, result_files):
143      with open(expected_file_name) as expected_file:
144        expected_output_lines = expected_file.readlines()
145        self.checkFileContents(result_file_name, expected_output_lines)
146    shutil.rmtree(common_functions_path)
147
148  def testExtractExtraFunctions(self):
149    cwp_statistics = {'dummy_method1,dummy_file1': ('dummy_object1,1', 0),
150                      'dummy_method2,dummy_file2': ('dummy_object2,2', 1),
151                      'dummy_method3,dummy_file3': ('dummy_object3,3', 1),
152                      'dummy_method4,dummy_file4': ('dummy_object4,4', 0)}
153    expected_output_lines = ['function,file,dso,inclusive_count\n',
154                             'dummy_method4,dummy_file4,dummy_object4,4\n',
155                             'dummy_method1,dummy_file1,dummy_object1,1']
156    temp_file, temp_filename = tempfile.mkstemp()
157    os.close(temp_file)
158    hot_functions_processor = self._CreateHotFunctionsProcessor(temp_filename)
159
160    hot_functions_processor.ExtractExtraFunctions(cwp_statistics, temp_filename)
161    self.checkFileContents(temp_filename, expected_output_lines)
162    os.remove(temp_filename)
163
164  def testParseFunctionGroups(self):
165    cwp_function_groups_lines = ['group1 /a\n', 'group2 /b\n', 'group3 /c\n',
166                                 'group4 /d\n']
167    expected_output = [('group1', '/a', 0, []), ('group2', '/b', 0, []),
168                       ('group3', '/c', 0, []), ('group4', '/d', 0, [])]
169    result = HotFunctionsProcessor.ParseFunctionGroups(
170        cwp_function_groups_lines)
171    self.assertListEqual(expected_output, result)
172
173  def testGroupExtraFunctions(self):
174    cwp_statistics = {'dummy_method1,/a/b': ('dummy_object1,1', 1),
175                      'dummy_method2,/c/d': ('dummy_object2,2', 0),
176                      'dummy_method3,/a/b': ('dummy_object3,3', 0),
177                      'dummy_method4,/c/d': ('dummy_object4,4', 1),
178                      'dummy_method5,/a/b': ('dummy_object5,5', 0),
179                      'dummy_method6,/e': ('dummy_object6,6', 0),
180                      'dummy_method7,/c/d': ('dummy_object7,7', 0),
181                      'dummy_method8,/e': ('dummy_object8,8', 0)}
182    cwp_groups_statistics_file, \
183        cwp_groups_statistics_filename = tempfile.mkstemp()
184
185    os.close(cwp_groups_statistics_file)
186
187    cwp_groups_file_path = tempfile.mkdtemp()
188    cwp_groups_file_prefix = os.path.join(cwp_groups_file_path, 'dummy')
189    hot_functions_processor = self._CreateHotFunctionsProcessor(
190        self._extra_cwp_functions_file)
191
192    hot_functions_processor.GroupExtraFunctions(cwp_statistics,
193                                                cwp_groups_file_prefix,
194                                                self._cwp_function_groups_file,
195                                                cwp_groups_statistics_filename)
196
197    expected_group_ab_lines = ['function,file,dso,inclusive_count\n',
198                               'dummy_method5,/a/b,dummy_object5,5\n',
199                               'dummy_method3,/a/b,dummy_object3,3']
200    expected_group_cd_lines = ['function,file,dso,inclusive_count\n',
201                               'dummy_method7,/c/d,dummy_object7,7\n',
202                               'dummy_method2,/c/d,dummy_object2,2']
203    expected_group_e_lines = ['function,file,dso,inclusive_count\n',
204                              'dummy_method8,/e,dummy_object8,8\n',
205                              'dummy_method6,/e,dummy_object6,6']
206    expected_group_statistics_lines = ['group,shared_path,inclusive_count\n',
207                                       'e,/e,14\n', 'cd,/c/d,9\n', 'ab,/a/b,8']
208
209    self.checkFileContents('%sab' % (cwp_groups_file_prefix,),
210                           expected_group_ab_lines)
211    self.checkFileContents('%scd' % (cwp_groups_file_prefix,),
212                           expected_group_cd_lines)
213    self.checkFileContents('%se' % (cwp_groups_file_prefix,),
214                           expected_group_e_lines)
215    self.checkFileContents(cwp_groups_statistics_filename,
216                           expected_group_statistics_lines)
217
218    shutil.rmtree(cwp_groups_file_path)
219    os.remove(cwp_groups_statistics_filename)
220
221
222if __name__ == '__main__':
223  unittest.main()
224