src/tools/memory_inspector/memory_inspector/data/file_storage_unittest.py

   1 # Copyright 2014 The Chromium Authors. All rights reserved.
   2 # Use of this source code is governed by a BSD-style license that can be
   3 # found in the LICENSE file.
   4
   5 """This unittest covers both file_storage and serialization modules."""
   6
   7 import os
   8 import tempfile
   9 import time
  10 import unittest
  11
  12 from memory_inspector.core import memory_map
  13 from memory_inspector.core import native_heap
  14 from memory_inspector.core import stacktrace
  15 from memory_inspector.core import symbol
  16 from memory_inspector.data import file_storage
  17
  18
  19 class FileStorageTest(unittest.TestCase):
  20   def setUp(self):
  21     self._storage_path = tempfile.mkdtemp()
  22     self._storage = file_storage.Storage(self._storage_path)
  23
  24   def tearDown(self):
  25     os.removedirs(self._storage_path)
  26
  27   def testSettings(self):
  28     settings_1 = { 'foo' : 1, 'bar' : 2 }
  29     settings_2 = { 'foo' : 1, 'bar' : 2 }
  30     self._storage.StoreSettings('one', settings_1)
  31     self._storage.StoreSettings('two', settings_2)
  32     self._DeepCompare(settings_1, self._storage.LoadSettings('one'))
  33     self._DeepCompare(settings_2, self._storage.LoadSettings('two'))
  34     self._storage.StoreSettings('one', {})
  35     self._storage.StoreSettings('two', {})
  36
  37   def testArchives(self):
  38     self._storage.OpenArchive('foo', create=True)
  39     self._storage.OpenArchive('bar', create=True)
  40     self._storage.OpenArchive('baz', create=True)
  41     self._storage.DeleteArchive('bar')
  42     self.assertTrue('foo' in self._storage.ListArchives())
  43     self.assertFalse('bar' in self._storage.ListArchives())
  44     self.assertTrue('baz' in self._storage.ListArchives())
  45     self._storage.DeleteArchive('foo')
  46     self._storage.DeleteArchive('baz')
  47
  48   def testSnapshots(self):
  49     archive = self._storage.OpenArchive('snapshots', create=True)
  50     t1 = archive.StartNewSnapshot()
  51     archive.StoreMemMaps(memory_map.Map())
  52     time.sleep(0.01) # Max snapshot resolution is in the order of usecs.
  53     t2 = archive.StartNewSnapshot()
  54     archive.StoreMemMaps(memory_map.Map())
  55     archive.StoreNativeHeap(native_heap.NativeHeap())
  56     self.assertIn(t1, archive.ListSnapshots())
  57     self.assertIn(t2, archive.ListSnapshots())
  58     self.assertTrue(archive.HasMemMaps(t1))
  59     self.assertFalse(archive.HasNativeHeap(t1))
  60     self.assertTrue(archive.HasMemMaps(t2))
  61     self.assertTrue(archive.HasNativeHeap(t2))
  62     self._storage.DeleteArchive('snapshots')
  63
  64   def testMmap(self):
  65     archive = self._storage.OpenArchive('mmap', create=True)
  66     timestamp = archive.StartNewSnapshot()
  67     mmap = memory_map.Map()
  68     map_entry1 = memory_map.MapEntry(4096, 8191, 'rw--', '/foo', 0)
  69     map_entry2 = memory_map.MapEntry(65536, 81919, 'rw--', '/bar', 4096)
  70     map_entry2.resident_pages = [5]
  71     mmap.Add(map_entry1)
  72     mmap.Add(map_entry2)
  73     archive.StoreMemMaps(mmap)
  74     mmap_deser = archive.LoadMemMaps(timestamp)
  75     self._DeepCompare(mmap, mmap_deser)
  76     self._storage.DeleteArchive('mmap')
  77
  78   def testNativeHeap(self):
  79     archive = self._storage.OpenArchive('nheap', create=True)
  80     timestamp = archive.StartNewSnapshot()
  81     nh = native_heap.NativeHeap()
  82     for i in xrange(1, 4):
  83       stack_trace = stacktrace.Stacktrace()
  84       frame = nh.GetStackFrame(i * 10 + 1)
  85       frame.SetExecFileInfo('foo.so', 1)
  86       stack_trace.Add(frame)
  87       frame = nh.GetStackFrame(i * 10 + 2)
  88       frame.SetExecFileInfo('bar.so', 2)
  89       stack_trace.Add(frame)
  90       nh.Add(native_heap.Allocation(size=i * 10,
  91                                     stack_trace=stack_trace,
  92                                     start=i * 20,
  93                                     flags=i * 30))
  94     archive.StoreNativeHeap(nh)
  95     nh_deser = archive.LoadNativeHeap(timestamp)
  96     self._DeepCompare(nh, nh_deser)
  97     self._storage.DeleteArchive('nheap')
  98
  99   def testSymbols(self):
 100     archive = self._storage.OpenArchive('symbols', create=True)
 101     symbols = symbol.Symbols()
 102     # Symbol db is global per archive, no need to StartNewSnapshot.
 103     symbols.Add('foo.so', 1, symbol.Symbol('sym1', 'file1.c', 11))
 104     symbols.Add('bar.so', 2, symbol.Symbol('sym2', 'file2.c', 12))
 105     sym3 = symbol.Symbol('sym3', 'file2.c', 13)
 106     sym3.AddSourceLineInfo('outer_file.c', 23)
 107     symbols.Add('baz.so', 3, sym3)
 108     archive.StoreSymbols(symbols)
 109     symbols_deser = archive.LoadSymbols()
 110     self._DeepCompare(symbols, symbols_deser)
 111     self._storage.DeleteArchive('symbols')
 112
 113   def _DeepCompare(self, a, b, prefix=''):
 114     """Recursively compares two objects (original and deserialized)."""
 115
 116     self.assertEqual(a is None, b is None)
 117     if a is None:
 118       return
 119
 120     _BASICTYPES = (long, int, basestring, float)
 121     if isinstance(a, _BASICTYPES) and isinstance(b, _BASICTYPES):
 122       return self.assertEqual(a, b, prefix)
 123
 124     self.assertEqual(type(a), type(b), prefix + ' type (%s vs %s' % (
 125         type(a), type(b)))
 126
 127     if isinstance(a, list):
 128       self.assertEqual(len(a), len(b), prefix + ' len (%d vs %d)' % (
 129           len(a), len(b)))
 130       for i in range(len(a)):
 131         self._DeepCompare(a[i], b[i], prefix + '[%d]' % i)
 132       return
 133
 134     if isinstance(a, dict):
 135       self.assertEqual(a.keys(), b.keys(), prefix + ' keys (%s vs %s)' % (
 136         str(a.keys()), str(b.keys())))
 137       for k in a.iterkeys():
 138         self._DeepCompare(a[k], b[k], prefix + '.' + str(k))
 139       return
 140
 141     return self._DeepCompare(a.__dict__, b.__dict__, prefix)