[Android] Add fast ELF Symbolizer to pylib.

build/android/pylib/symbols/elf_symbolizer_unittest.py

+#!/usr/bin/env python
+# Copyright 2014 The Chromium 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 functools
+import logging
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(__file__))
+import elf_symbolizer
+import mock_addr2line
+
+
+_MOCK_A2L_PATH = os.path.join(os.path.dirname(mock_addr2line.__file__),
+                              'mock_addr2line')
+
+
+class ELFSymbolizerTest(unittest.TestCase):
+  def setUp(self):
+    self._callback = functools.partial(
+        ELFSymbolizerTest._SymbolizeCallback, self)
+    self._resolved_addresses = set()
+    # Mute warnings, we expect them due to the crash/hang tests.
+    logging.getLogger().setLevel(logging.ERROR)
+
+  def test_parallelism_1(self):
+    self._RunTest(max_concurrent_jobs=1, num_symbols=100)
+
+  def test_parallelism_4(self):
+    self._RunTest(max_concurrent_jobs=4, num_symbols=100)
+
+  def test_parallelism_8(self):
+    self._RunTest(max_concurrent_jobs=8, num_symbols=100)
+
+  def test_crash(self):
+    os.environ['MOCK_A2L_CRASH_EVERY'] = '99'
+    self._RunTest(max_concurrent_jobs=1, num_symbols=100)
+    os.environ['MOCK_A2L_CRASH_EVERY'] = '0'
+
+  def test_hang(self):
+    os.environ['MOCK_A2L_HANG_EVERY'] = '99'
+    self._RunTest(max_concurrent_jobs=1, num_symbols=100)
+    os.environ['MOCK_A2L_HANG_EVERY'] = '0'
+
+  def test_incomplete_syminfo(self):
+    """Stimulate the symbol-not-resolved logic."""
+    symbolizer = elf_symbolizer.ELFSymbolizer(
+        elf_file_path='/path/doesnt/matter/mock_lib1.so',
+        addr2line_path=_MOCK_A2L_PATH,
+        callback=self._callback,
+        max_concurrent_jobs=1)
+
+    # Test symbols with valid name but incomplete path.
+    addr = 1024 * 1024  # mock_addr2line treats this addr range specially.
+    exp_sym_name = 'mock_sym_for_addr_%d' % addr
+    exp_source_path = '??'
+    exp_source_line = 0
+    cb_args = (addr, exp_sym_name, exp_source_path, exp_source_line)
+    symbolizer.SymbolizeAsync(addr, cb_args)
+
+    # Test symbols with no name or sym info.
+    addr = 2 * 1024 * 1024  # mock_addr2line treats this addr range specially.
+    exp_sym_name = '??'
+    exp_source_path = '??/mock_lib1.so+0x200000'
+    exp_source_line = 0
+    cb_args = (addr, exp_sym_name, exp_source_path, exp_source_line)
+    symbolizer.SymbolizeAsync(addr, cb_args)
+
+    symbolizer.Join()
+
+  def _RunTest(self, max_concurrent_jobs, num_symbols):
+    symbolizer = elf_symbolizer.ELFSymbolizer(
+        elf_file_path='/path/doesnt/matter/mock_lib1.so',
+        addr2line_path=_MOCK_A2L_PATH,
+        callback=self._callback,
+        max_concurrent_jobs=max_concurrent_jobs,
+        addr2line_timeout=0.5)
+
+    for addr in xrange(num_symbols):
+      exp_sym_name = 'mock_sym_for_addr_%d' % addr
+      exp_source_path = 'mock_src/mock_lib1.so.c'
+      exp_source_line = addr
+      cb_args = (addr, exp_sym_name, exp_source_path, exp_source_line)
+      symbolizer.SymbolizeAsync(addr, cb_args)
+
+    symbolizer.Join()
+
+    # Check that all the expected callbacks have been received.
+    for addr in xrange(num_symbols):
+      self.assertIn(addr, self._resolved_addresses)
+      self._resolved_addresses.remove(addr)
+
+    # Check for unexpected callbacks.
+    self.assertEqual(len(self._resolved_addresses), 0)
+
+  def _SymbolizeCallback(self, sym_info, cb_args):
+    self.assertTrue(isinstance(sym_info, elf_symbolizer.ELFSymbolInfo))
+    self.assertTrue(isinstance(cb_args, tuple))
+    self.assertEqual(len(cb_args), 4)
+
+    # Unpack expectations from the  callback extra argument.
+    (addr, exp_sym_name, exp_source_path, exp_source_line) = cb_args
+    self.assertEqual(sym_info.name, exp_sym_name)
+    self.assertEqual(sym_info.source_path, exp_source_path)
+    self.assertEqual(sym_info.source_line, exp_source_line)
+
+    # Check against duplicate callbacks.
+    self.assertNotIn(addr, self._resolved_addresses)
+    self._resolved_addresses.add(addr)
+
+if __name__ == '__main__':
+  unittest.main()