1
# Copyright (C) 2007, 2009, 2010, 2011 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Tests for profiling data collection."""
24
from bzrlib import errors, tests
25
from bzrlib.tests import (
30
_TXT_HEADER = " CallCount Recursive Total(ms) " + \
31
"Inline(ms) module:lineno(function)\n"
35
"A simple routine to profile."
36
result = sorted(['abc', 'def', 'ghi'])
40
"Collect and return some dummy profile data."
41
from bzrlib.lsprof import profile
42
ret, stats = profile(_junk_callable)
46
class TestStatsSave(tests.TestCaseInTempDir):
48
_test_needs_features = [features.lsprof_feature]
51
super(tests.TestCaseInTempDir, self).setUp()
52
self.stats = _collect_stats()
54
def _tempfile(self, ext):
56
return bzrlib.osutils.pathjoin(dir, "tmp_profile_data." + ext)
58
def test_stats_save_to_txt(self):
59
f = self._tempfile("txt")
61
lines = open(f).readlines()
62
self.assertEqual(lines[0], _TXT_HEADER)
64
def test_stats_save_to_callgrind(self):
65
f = self._tempfile("callgrind")
67
lines = open(f).readlines()
68
self.assertEqual(lines[0], "events: Ticks\n")
69
f = bzrlib.osutils.pathjoin(self.test_dir, "callgrind.out.foo")
71
lines = open(f).readlines()
72
self.assertEqual(lines[0], "events: Ticks\n")
73
# Test explicit format nommination
74
f2 = self._tempfile("txt")
75
self.stats.save(f2, format="callgrind")
76
lines2 = open(f2).readlines()
77
self.assertEqual(lines2[0], "events: Ticks\n")
79
def test_stats_save_to_pickle(self):
80
f = self._tempfile("pkl")
82
data1 = cPickle.load(open(f))
83
self.assertEqual(type(data1), bzrlib.lsprof.Stats)
86
class TestBzrProfiler(tests.TestCase):
88
_test_needs_features = [features.lsprof_feature]
90
def test_start_call_stuff_stop(self):
91
profiler = bzrlib.lsprof.BzrProfiler()
98
stats = profiler.stop()
100
lines = [str(data) for data in stats.data]
101
lines = [line for line in lines if 'a_function' in line]
102
self.assertLength(1, lines)
104
def test_block_0(self):
105
# When profiler_block is 0, reentrant profile requests fail.
106
self.overrideAttr(bzrlib.lsprof.BzrProfiler, 'profiler_block', 0)
109
profiler = bzrlib.lsprof.BzrProfiler()
110
self.assertRaises(errors.BzrError, profiler.start)
111
inner_calls.append(True)
112
bzrlib.lsprof.profile(inner)
113
self.assertLength(1, inner_calls)
115
def test_block_1(self):
116
# When profiler_block is 1, concurrent profiles serialise.
117
# This is tested by manually acquiring the profiler lock, then
118
# starting a thread that tries to profile, and releasing the lock.
119
# We know due to test_block_0 that two profiles at once hit the lock,
120
# so while this isn't perfect (we'd want a callback on the lock being
121
# entered to allow lockstep evaluation of the actions), its good enough
122
# to be confident regressions would be caught. Alternatively, if this
123
# is flakey, a fake Lock object can be used to trace the calls made.
126
calls.append('profiled')
128
bzrlib.lsprof.profile(profiled)
129
calls.append('after_profiled')
130
thread = threading.Thread(target=do_profile)
131
bzrlib.lsprof.BzrProfiler.profiler_lock.acquire()
136
bzrlib.lsprof.BzrProfiler.profiler_lock.release()
139
self.assertLength(2, calls)