src/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 unittest

from pylib.symbols import elf_symbolizer
from pylib.symbols import mock_addr2line


_MOCK_A2L_PATH = os.path.join(os.path.dirname(mock_addr2line.__file__),
                              'mock_addr2line')
_INCOMPLETE_MOCK_ADDR = 1024 * 1024
_UNKNOWN_MOCK_ADDR = 2 * 1024 * 1024
_INLINE_MOCK_ADDR = 3 * 1024 * 1024


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 testParallelism1(self):
    self._RunTest(max_concurrent_jobs=1, num_symbols=100)

  def testParallelism4(self):
    self._RunTest(max_concurrent_jobs=4, num_symbols=100)

  def testParallelism8(self):
    self._RunTest(max_concurrent_jobs=8, num_symbols=100)

  def testCrash(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 testHang(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 testInlines(self):
    """Stimulate the inline processing logic."""
    symbolizer = elf_symbolizer.ELFSymbolizer(
        elf_file_path='/path/doesnt/matter/mock_lib1.so',
        addr2line_path=_MOCK_A2L_PATH,
        callback=self._callback,
        inlines=True,
        max_concurrent_jobs=4)

    for addr in xrange(1000):
      exp_inline = False
      exp_unknown = False

      # First 100 addresses with inlines.
      if addr < 100:
        addr += _INLINE_MOCK_ADDR
        exp_inline = True

      # Followed by 100 without inlines.
      elif addr < 200:
        pass

      # Followed by 100 interleaved inlines and not inlines.
      elif addr < 300:
        if addr & 1:
          addr += _INLINE_MOCK_ADDR
          exp_inline = True

      # Followed by 100 interleaved inlines and unknonwn.
      elif addr < 400:
        if addr & 1:
          addr += _INLINE_MOCK_ADDR
          exp_inline = True
        else:
          addr += _UNKNOWN_MOCK_ADDR
          exp_unknown = True

      exp_name = 'mock_sym_for_addr_%d' % addr if not exp_unknown else None
      exp_source_path = 'mock_src/mock_lib1.so.c' if not exp_unknown else None
      exp_source_line = addr if not exp_unknown else None
      cb_arg = (addr, exp_name, exp_source_path, exp_source_line, exp_inline)
      symbolizer.SymbolizeAsync(addr, cb_arg)

    symbolizer.Join()

  def testIncompleteSyminfo(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 = _INCOMPLETE_MOCK_ADDR
    exp_name = 'mock_sym_for_addr_%d' % addr
    exp_source_path = None
    exp_source_line = None
    cb_arg = (addr, exp_name, exp_source_path, exp_source_line, False)
    symbolizer.SymbolizeAsync(addr, cb_arg)

    # Test symbols with no name or sym info.
    addr = _UNKNOWN_MOCK_ADDR
    exp_name = None
    exp_source_path = None
    exp_source_line = None
    cb_arg = (addr, exp_name, exp_source_path, exp_source_line, False)
    symbolizer.SymbolizeAsync(addr, cb_arg)

    symbolizer.Join()

  def testWaitForIdle(self):
    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 = _INCOMPLETE_MOCK_ADDR
    exp_name = 'mock_sym_for_addr_%d' % addr
    exp_source_path = None
    exp_source_line = None
    cb_arg = (addr, exp_name, exp_source_path, exp_source_line, False)
    symbolizer.SymbolizeAsync(addr, cb_arg)
    symbolizer.WaitForIdle()

    # Test symbols with no name or sym info.
    addr = _UNKNOWN_MOCK_ADDR
    exp_name = None
    exp_source_path = None
    exp_source_line = None
    cb_arg = (addr, exp_name, exp_source_path, exp_source_line, False)
    symbolizer.SymbolizeAsync(addr, cb_arg)
    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_name = 'mock_sym_for_addr_%d' % addr
      exp_source_path = 'mock_src/mock_lib1.so.c'
      exp_source_line = addr
      cb_arg = (addr, exp_name, exp_source_path, exp_source_line, False)
      symbolizer.SymbolizeAsync(addr, cb_arg)

    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_arg):
    self.assertTrue(isinstance(sym_info, elf_symbolizer.ELFSymbolInfo))
    self.assertTrue(isinstance(cb_arg, tuple))
    self.assertEqual(len(cb_arg), 5)

    # Unpack expectations from the callback extra argument.
    (addr, exp_name, exp_source_path, exp_source_line, exp_inlines) = cb_arg
    if exp_name is None:
      self.assertIsNone(sym_info.name)
    else:
      self.assertTrue(sym_info.name.startswith(exp_name))
    self.assertEqual(sym_info.source_path, exp_source_path)
    self.assertEqual(sym_info.source_line, exp_source_line)

    if exp_inlines:
      self.assertEqual(sym_info.name, exp_name + '_inner')
      self.assertEqual(sym_info.inlined_by.name, exp_name + '_middle')
      self.assertEqual(sym_info.inlined_by.inlined_by.name,
                       exp_name + '_outer')

    # Check against duplicate callbacks.
    self.assertNotIn(addr, self._resolved_addresses)
    self._resolved_addresses.add(addr)


if __name__ == '__main__':
  unittest.main()