mirrors/poky
1
0
Fork 0
mirror of https://git.yoctoproject.org/poky synced 2026-04-21 21:32:12 +02:00
Code Issues Packages Projects Releases Wiki Activity

pybootchartgui: Import pybootchartgui 0.14.5

Browse Source 
This update the pybootchartgui code to the latest release from its new
location at "https://github.com/mmeeks/bootchart". This only imports
the relevant parts, and not all of bootchart2.

(From OE-Core rev: 6f1568e54a7808b2ab568618fc5bb244249579f1)

Signed-off-by: Peter Kjellerstedt <peter.kjellerstedt@axis.com>
Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org>
This commit is contained in:
Peter Kjellerstedt
2013-11-15 18:08:50 +01:00
committed by Richard Purdie
parent a52c9b9a76
commit d0e513d437
16 changed files with 2514 additions and 1414 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
scripts/pybootchartgui/pybootchartgui/tests/parser_test.py
View File

@@ -4,90 +4,102 @@ import unittest
sys.path.insert(0, os.getcwd())
import parsing
import pybootchartgui.parsing as parsing
import pybootchartgui.main as main
debug = False
def floatEq(f1, f2):
return math.fabs(f1-f2) < 0.00001
bootchart_dir = os.path.join(os.path.dirname(sys.argv[0]), '../../examples/1/')
parser = main._mk_options_parser()
options, args = parser.parse_args(['--q', bootchart_dir])
writer = main._mk_writer(options)
class TestBCParser(unittest.TestCase):
def setUp(self):
self.name = "My first unittest"
self.rootdir = '../examples/1'
self.rootdir = bootchart_dir
def mk_fname(self,f):
return os.path.join(self.rootdir, f)
def testParseHeader(self):
state = parsing.parse_file(parsing.ParserState(), self.mk_fname('header'))
trace = parsing.Trace(writer, args, options)
state = parsing.parse_file(writer, trace, self.mk_fname('header'))
self.assertEqual(6, len(state.headers))
self.assertEqual(2, parsing.get_num_cpus(state.headers))
def test_parseTimedBlocks(self):
state = parsing.parse_file(parsing.ParserState(), self.mk_fname('proc_diskstats.log'))
trace = parsing.Trace(writer, args, options)
state = parsing.parse_file(writer, trace, self.mk_fname('proc_diskstats.log'))
self.assertEqual(141, len(state.disk_stats))
def testParseProcPsLog(self):
state = parsing.parse_file(parsing.ParserState(), self.mk_fname('proc_ps.log'))
trace = parsing.Trace(writer, args, options)
state = parsing.parse_file(writer, trace, self.mk_fname('proc_ps.log'))
samples = state.ps_stats
processes = samples.process_list
sorted_processes = sorted(processes, key=lambda p: p.pid )
for index, line in enumerate(open(self.mk_fname('extract2.proc_ps.log'))):
processes = samples.process_map
sorted_processes = [processes[k] for k in sorted(processes.keys())]
ps_data = open(self.mk_fname('extract2.proc_ps.log'))
for index, line in enumerate(ps_data):
tokens = line.split();
process = sorted_processes[index]
if debug:
print tokens[0:4]
print process.pid, process.cmd, process.ppid, len(process.samples)
print '-------------------'
self.assertEqual(tokens[0], str(process.pid))
if debug:
print(tokens[0:4])
print(process.pid / 1000, process.cmd, process.ppid, len(process.samples))
print('-------------------')
self.assertEqual(tokens[0], str(process.pid // 1000))
self.assertEqual(tokens[1], str(process.cmd))
self.assertEqual(tokens[2], str(process.ppid))
self.assertEqual(tokens[2], str(process.ppid // 1000))
self.assertEqual(tokens[3], str(len(process.samples)))
ps_data.close()
def testparseProcDiskStatLog(self):
state_with_headers = parsing.parse_file(parsing.ParserState(), self.mk_fname('header'))
trace = parsing.Trace(writer, args, options)
state_with_headers = parsing.parse_file(writer, trace, self.mk_fname('header'))
state_with_headers.headers['system.cpu'] = 'xxx (2)'
samples = parsing.parse_file(state_with_headers, self.mk_fname('proc_diskstats.log')).disk_stats
samples = parsing.parse_file(writer, state_with_headers, self.mk_fname('proc_diskstats.log')).disk_stats
self.assertEqual(141, len(samples))
for index, line in enumerate(open(self.mk_fname('extract.proc_diskstats.log'))):
diskstats_data = open(self.mk_fname('extract.proc_diskstats.log'))
for index, line in enumerate(diskstats_data):
tokens = line.split('\t')
sample = samples[index]
if debug:
print line.rstrip(),
print sample
print '-------------------'
print(line.rstrip())
print(sample)
print('-------------------')
self.assertEqual(tokens[0], str(sample.time))
self.assert_(floatEq(float(tokens[1]), sample.read))
self.assert_(floatEq(float(tokens[2]), sample.write))
self.assert_(floatEq(float(tokens[3]), sample.util))
diskstats_data.close()
def testparseProcStatLog(self):
samples = parsing.parse_file(parsing.ParserState(), self.mk_fname('proc_stat.log')).cpu_stats
trace = parsing.Trace(writer, args, options)
samples = parsing.parse_file(writer, trace, self.mk_fname('proc_stat.log')).cpu_stats
self.assertEqual(141, len(samples))
for index, line in enumerate(open(self.mk_fname('extract.proc_stat.log'))):
stat_data = open(self.mk_fname('extract.proc_stat.log'))
for index, line in enumerate(stat_data):
tokens = line.split('\t')
sample = samples[index]
if debug:
print line.rstrip()
print sample
print '-------------------'
print(line.rstrip())
print(sample)
print('-------------------')
self.assert_(floatEq(float(tokens[0]), sample.time))
self.assert_(floatEq(float(tokens[1]), sample.user))
self.assert_(floatEq(float(tokens[2]), sample.sys))
self.assert_(floatEq(float(tokens[3]), sample.io))
def testParseLogDir(self):
res = parsing.parse([self.rootdir], False)
self.assertEqual(4, len(res))
stat_data.close()
if __name__ == '__main__':
unittest.main()

30
scripts/pybootchartgui/pybootchartgui/tests/process_tree_test.py
View File

@@ -4,16 +4,28 @@ import unittest
sys.path.insert(0, os.getcwd())
import parsing
import process_tree
import pybootchartgui.parsing as parsing
import pybootchartgui.process_tree as process_tree
import pybootchartgui.main as main
if sys.version_info >= (3, 0):
long = int
class TestProcessTree(unittest.TestCase):
def setUp(self):
self.name = "Process tree unittest"
self.rootdir = '../examples/1'
self.ps_stats = parsing.parse_file(parsing.ParserState(), self.mk_fname('proc_ps.log')).ps_stats
self.processtree = process_tree.ProcessTree(self.ps_stats, None, False, for_testing = True)
self.name = "Process tree unittest"
self.rootdir = os.path.join(os.path.dirname(sys.argv[0]), '../../examples/1/')
parser = main._mk_options_parser()
options, args = parser.parse_args(['--q', self.rootdir])
writer = main._mk_writer(options)
trace = parsing.Trace(writer, args, options)
parsing.parse_file(writer, trace, self.mk_fname('proc_ps.log'))
trace.compile(writer)
self.processtree = process_tree.ProcessTree(writer, None, trace.ps_stats, \
trace.ps_stats.sample_period, None, options.prune, None, None, False, for_testing = True)
def mk_fname(self,f):
return os.path.join(self.rootdir, f)
@@ -26,14 +38,16 @@ class TestProcessTree(unittest.TestCase):
return flattened
def checkAgainstJavaExtract(self, filename, process_tree):
for expected, actual in zip(open(filename), self.flatten(process_tree)):
test_data = open(filename)
for expected, actual in zip(test_data, self.flatten(process_tree)):
tokens = expected.split('\t')
self.assertEqual(int(tokens[0]), actual.pid)
self.assertEqual(int(tokens[0]), actual.pid // 1000)
self.assertEqual(tokens[1], actual.cmd)
self.assertEqual(long(tokens[2]), 10 * actual.start_time)
self.assert_(long(tokens[3]) - 10 * actual.duration < 5, "duration")
self.assertEqual(int(tokens[4]), len(actual.child_list))
self.assertEqual(int(tokens[5]), len(actual.samples))
test_data.close()
def testBuild(self):
process_tree = self.processtree.process_tree