1#!/usr/bin/env python 2# Copyright 2014 The Chromium Authors. All rights reserved. 3# Use of this source code is governed by a BSD-style license that can be 4# found in the LICENSE file. 5 6import functools 7import logging 8import os 9import sys 10import unittest 11 12sys.path.insert(0, os.path.dirname(__file__)) 13import elf_symbolizer 14import mock_addr2line 15 16 17_MOCK_A2L_PATH = os.path.join(os.path.dirname(mock_addr2line.__file__), 18 'mock_addr2line') 19_INCOMPLETE_MOCK_ADDR = 1024 * 1024 20_UNKNOWN_MOCK_ADDR = 2 * 1024 * 1024 21_INLINE_MOCK_ADDR = 3 * 1024 * 1024 22 23 24class ELFSymbolizerTest(unittest.TestCase): 25 def setUp(self): 26 self._callback = functools.partial( 27 ELFSymbolizerTest._SymbolizeCallback, self) 28 self._resolved_addresses = set() 29 # Mute warnings, we expect them due to the crash/hang tests. 30 logging.getLogger().setLevel(logging.ERROR) 31 32 def testParallelism1(self): 33 self._RunTest(max_concurrent_jobs=1, num_symbols=100) 34 35 def testParallelism4(self): 36 self._RunTest(max_concurrent_jobs=4, num_symbols=100) 37 38 def testParallelism8(self): 39 self._RunTest(max_concurrent_jobs=8, num_symbols=100) 40 41 def testCrash(self): 42 os.environ['MOCK_A2L_CRASH_EVERY'] = '99' 43 self._RunTest(max_concurrent_jobs=1, num_symbols=100) 44 os.environ['MOCK_A2L_CRASH_EVERY'] = '0' 45 46 def testHang(self): 47 os.environ['MOCK_A2L_HANG_EVERY'] = '99' 48 self._RunTest(max_concurrent_jobs=1, num_symbols=100) 49 os.environ['MOCK_A2L_HANG_EVERY'] = '0' 50 51 def testInlines(self): 52 """Stimulate the inline processing logic.""" 53 symbolizer = elf_symbolizer.ELFSymbolizer( 54 elf_file_path='/path/doesnt/matter/mock_lib1.so', 55 addr2line_path=_MOCK_A2L_PATH, 56 callback=self._callback, 57 inlines=True, 58 max_concurrent_jobs=4) 59 60 for addr in xrange(1000): 61 exp_inline = False 62 exp_unknown = False 63 64 # First 100 addresses with inlines. 65 if addr < 100: 66 addr += _INLINE_MOCK_ADDR 67 exp_inline = True 68 69 # Followed by 100 without inlines. 70 elif addr < 200: 71 pass 72 73 # Followed by 100 interleaved inlines and not inlines. 74 elif addr < 300: 75 if addr & 1: 76 addr += _INLINE_MOCK_ADDR 77 exp_inline = True 78 79 # Followed by 100 interleaved inlines and unknonwn. 80 elif addr < 400: 81 if addr & 1: 82 addr += _INLINE_MOCK_ADDR 83 exp_inline = True 84 else: 85 addr += _UNKNOWN_MOCK_ADDR 86 exp_unknown = True 87 88 exp_name = 'mock_sym_for_addr_%d' % addr if not exp_unknown else None 89 exp_source_path = 'mock_src/mock_lib1.so.c' if not exp_unknown else None 90 exp_source_line = addr if not exp_unknown else None 91 cb_arg = (addr, exp_name, exp_source_path, exp_source_line, exp_inline) 92 symbolizer.SymbolizeAsync(addr, cb_arg) 93 94 symbolizer.Join() 95 96 def testIncompleteSyminfo(self): 97 """Stimulate the symbol-not-resolved logic.""" 98 symbolizer = elf_symbolizer.ELFSymbolizer( 99 elf_file_path='/path/doesnt/matter/mock_lib1.so', 100 addr2line_path=_MOCK_A2L_PATH, 101 callback=self._callback, 102 max_concurrent_jobs=1) 103 104 # Test symbols with valid name but incomplete path. 105 addr = _INCOMPLETE_MOCK_ADDR 106 exp_name = 'mock_sym_for_addr_%d' % addr 107 exp_source_path = None 108 exp_source_line = None 109 cb_arg = (addr, exp_name, exp_source_path, exp_source_line, False) 110 symbolizer.SymbolizeAsync(addr, cb_arg) 111 112 # Test symbols with no name or sym info. 113 addr = _UNKNOWN_MOCK_ADDR 114 exp_name = None 115 exp_source_path = None 116 exp_source_line = None 117 cb_arg = (addr, exp_name, exp_source_path, exp_source_line, False) 118 symbolizer.SymbolizeAsync(addr, cb_arg) 119 120 symbolizer.Join() 121 122 def _RunTest(self, max_concurrent_jobs, num_symbols): 123 symbolizer = elf_symbolizer.ELFSymbolizer( 124 elf_file_path='/path/doesnt/matter/mock_lib1.so', 125 addr2line_path=_MOCK_A2L_PATH, 126 callback=self._callback, 127 max_concurrent_jobs=max_concurrent_jobs, 128 addr2line_timeout=0.5) 129 130 for addr in xrange(num_symbols): 131 exp_name = 'mock_sym_for_addr_%d' % addr 132 exp_source_path = 'mock_src/mock_lib1.so.c' 133 exp_source_line = addr 134 cb_arg = (addr, exp_name, exp_source_path, exp_source_line, False) 135 symbolizer.SymbolizeAsync(addr, cb_arg) 136 137 symbolizer.Join() 138 139 # Check that all the expected callbacks have been received. 140 for addr in xrange(num_symbols): 141 self.assertIn(addr, self._resolved_addresses) 142 self._resolved_addresses.remove(addr) 143 144 # Check for unexpected callbacks. 145 self.assertEqual(len(self._resolved_addresses), 0) 146 147 def _SymbolizeCallback(self, sym_info, cb_arg): 148 self.assertTrue(isinstance(sym_info, elf_symbolizer.ELFSymbolInfo)) 149 self.assertTrue(isinstance(cb_arg, tuple)) 150 self.assertEqual(len(cb_arg), 5) 151 152 # Unpack expectations from the callback extra argument. 153 (addr, exp_name, exp_source_path, exp_source_line, exp_inlines) = cb_arg 154 if exp_name is None: 155 self.assertIsNone(sym_info.name) 156 else: 157 self.assertTrue(sym_info.name.startswith(exp_name)) 158 self.assertEqual(sym_info.source_path, exp_source_path) 159 self.assertEqual(sym_info.source_line, exp_source_line) 160 161 if exp_inlines: 162 self.assertEqual(sym_info.name, exp_name + '_inner') 163 self.assertEqual(sym_info.inlined_by.name, exp_name + '_middle') 164 self.assertEqual(sym_info.inlined_by.inlined_by.name, 165 exp_name + '_outer') 166 167 # Check against duplicate callbacks. 168 self.assertNotIn(addr, self._resolved_addresses) 169 self._resolved_addresses.add(addr) 170 171 172if __name__ == '__main__': 173 unittest.main()