1#!/usr/bin/env python
2
3# Copyright (c) 2015 The Chromium Authors. All rights reserved.
4# Use of this source code is governed by a BSD-style license that can be
5# found in the LICENSE file.
6import contextlib
7import unittest
8
9import agents.atrace_agent as atrace_agent
10import systrace
11import util
12
13DEVICE_SERIAL = 'AG8404EC0444AGC'
14LIST_TMP_ARGS = ['ls', '/data/local/tmp']
15ATRACE_ARGS = ['atrace', '-z', '-t', '10']
16CATEGORIES = ['sched', 'gfx', 'view', 'wm']
17ADB_SHELL = ['adb', '-s', DEVICE_SERIAL, 'shell']
18
19SYSTRACE_CMD = ['./systrace.py', '--time', '10', '-o', 'out.html', '-e',
20                DEVICE_SERIAL] + CATEGORIES
21TRACE_CMD = (ADB_SHELL + ATRACE_ARGS + CATEGORIES +
22             [';', 'ps', '-t'])
23
24SYSTRACE_LIST_CATEGORIES_CMD = ['./systrace.py', '-e', DEVICE_SERIAL, '-l']
25TRACE_LIST_CATEGORIES_CMD = (ADB_SHELL + ['atrace', '--list_categories'])
26
27LEGACY_ATRACE_ARGS = ['atrace', '-z', '-t', '10', '-s']
28LEGACY_TRACE_CMD = (ADB_SHELL + LEGACY_ATRACE_ARGS +
29             [';', 'ps', '-t'])
30
31TEST_DIR = 'test_data/'
32ATRACE_DATA = TEST_DIR + 'atrace_data'
33ATRACE_DATA_RAW = TEST_DIR + 'atrace_data_raw'
34ATRACE_DATA_STRIPPED = TEST_DIR + 'atrace_data_stripped'
35ATRACE_DATA_THREAD_FIXED = TEST_DIR + 'atrace_data_thread_fixed'
36ATRACE_DATA_WITH_THREAD_LIST = TEST_DIR + 'atrace_data_with_thread_list'
37ATRACE_THREAD_NAMES = TEST_DIR + 'atrace_thread_names'
38
39
40class UtilUnitTest(unittest.TestCase):
41  def test_construct_adb_shell_command(self):
42    command = util.construct_adb_shell_command(LIST_TMP_ARGS, None)
43    self.assertEqual(' '.join(command), 'adb shell ls /data/local/tmp')
44
45    command = util.construct_adb_shell_command(LIST_TMP_ARGS, DEVICE_SERIAL)
46    self.assertEqual(' '.join(command),
47                     'adb -s AG8404EC0444AGC shell ls /data/local/tmp')
48
49    command = util.construct_adb_shell_command(ATRACE_ARGS, DEVICE_SERIAL)
50    self.assertEqual(' '.join(command),
51                     'adb -s AG8404EC0444AGC shell atrace -z -t 10')
52
53
54class AtraceAgentUnitTest(unittest.TestCase):
55  def test_construct_trace_command(self):
56    options, categories = systrace.parse_options(SYSTRACE_CMD)
57    agent = atrace_agent.AtraceAgent(options, categories)
58    tracer_args = agent._construct_trace_command()
59    self.assertEqual(' '.join(TRACE_CMD), ' '.join(tracer_args))
60    self.assertEqual(True, agent.expect_trace())
61
62  def test_extract_thread_list(self):
63    with contextlib.nested(open(ATRACE_DATA_WITH_THREAD_LIST, 'r'),
64                           open(ATRACE_DATA_RAW, 'r'),
65                           open(ATRACE_THREAD_NAMES, 'r')) as (f1, f2, f3):
66      atrace_data_with_thread_list = f1.read()
67      atrace_data_raw = f2.read()
68      atrace_thread_names = f3.read()
69
70      trace_data, thread_names = atrace_agent.extract_thread_list(
71          atrace_data_with_thread_list)
72      self.assertEqual(atrace_data_raw, trace_data)
73      self.assertEqual(atrace_thread_names, str(thread_names))
74
75  def test_strip_and_decompress_trace(self):
76    with contextlib.nested(open(ATRACE_DATA_RAW, 'r'),
77                           open(ATRACE_DATA_STRIPPED, 'r')) as (f1, f2):
78      atrace_data_raw = f1.read()
79      atrace_data_stripped = f2.read()
80
81      trace_data = atrace_agent.strip_and_decompress_trace(atrace_data_raw)
82      self.assertEqual(atrace_data_stripped, trace_data)
83
84  def test_fix_thread_names(self):
85    with contextlib.nested(
86        open(ATRACE_DATA_STRIPPED, 'r'),
87        open(ATRACE_THREAD_NAMES, 'r'),
88        open(ATRACE_DATA_THREAD_FIXED, 'r')) as (f1, f2, f3):
89      atrace_data_stripped = f1.read()
90      atrace_thread_names = f2.read()
91      atrace_data_thread_fixed = f3.read()
92      thread_names = eval(atrace_thread_names)
93
94      trace_data = atrace_agent.fix_thread_names(
95          atrace_data_stripped, thread_names)
96      self.assertEqual(atrace_data_thread_fixed, trace_data)
97
98  def test_preprocess_trace_data(self):
99    with contextlib.nested(open(ATRACE_DATA_WITH_THREAD_LIST, 'r'),
100                           open(ATRACE_DATA, 'r')) as (f1, f2):
101      atrace_data_with_thread_list = f1.read()
102      atrace_data = f2.read()
103
104      options, categories = systrace.parse_options([])
105      agent = atrace_agent.AtraceAgent(options, categories)
106      trace_data = agent._preprocess_trace_data(atrace_data_with_thread_list)
107      self.assertEqual(atrace_data, trace_data)
108
109  def test_list_categories(self):
110    options, categories = systrace.parse_options(SYSTRACE_LIST_CATEGORIES_CMD)
111    agent = atrace_agent.AtraceAgent(options, categories)
112    tracer_args = agent._construct_trace_command()
113    self.assertEqual(' '.join(TRACE_LIST_CATEGORIES_CMD), ' '.join(tracer_args))
114    self.assertEqual(False, agent.expect_trace())
115
116class AtraceLegacyAgentUnitTest(unittest.TestCase):
117  def test_construct_trace_command(self):
118    options, categories = systrace.parse_options(SYSTRACE_CMD)
119    agent = atrace_agent.AtraceLegacyAgent(options, categories)
120    tracer_args = agent._construct_trace_command()
121    self.assertEqual(' '.join(LEGACY_TRACE_CMD), ' '.join(tracer_args))
122    self.assertEqual(True, agent.expect_trace())
123
124if __name__ == '__main__':
125    unittest.main()
126