sdk/platform-tools/systrace/catapult/systrace/systrace/agents/atrace_agent_unittest.py

   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.
   6
   7 import contextlib
   8 import os
   9 import unittest
  10
  11 from systrace import systrace
  12 from systrace.agents import atrace_agent
  13
  14 DEVICE_SERIAL = 'AG8404EC0444AGC'
  15 ATRACE_ARGS = ['atrace', '-z', '-t', '10', '-b', '4096']
  16 CATEGORIES = ['sched', 'gfx', 'view', 'wm']
  17 ADB_SHELL = ['adb', '-s', DEVICE_SERIAL, 'shell']
  18
  19 SYSTRACE_CMD = ['./systrace.py', '--time', '10', '-o', 'out.html', '-e',
  20                 DEVICE_SERIAL] + CATEGORIES
  21 TRACE_CMD = (ADB_SHELL + ATRACE_ARGS + CATEGORIES)
  22
  23 SYSTRACE_LIST_CATEGORIES_CMD = ['./systrace.py', '-e', DEVICE_SERIAL, '-l']
  24 TRACE_LIST_CATEGORIES_CMD = (ADB_SHELL + ['atrace', '--list_categories'])
  25
  26 LEGACY_ATRACE_ARGS = ['atrace', '-z', '-t', '10', '-b', '4096', '-s']
  27 LEGACY_TRACE_CMD = (ADB_SHELL + LEGACY_ATRACE_ARGS)
  28
  29 STOP_FIX_UPS = ['atrace', '--no-fix-threads', '--no-fix-tgids']
  30
  31
  32 SYSTRACE_BOOT_CMD = (['./systrace.py', '--boot', '-e', DEVICE_SERIAL] +
  33                      CATEGORIES)
  34 TRACE_BOOT_CMD = (ADB_SHELL +
  35                   ['atrace', '--async_stop', '&&', 'setprop',
  36                    'persist.debug.atrace.boottrace', '0', '&&',
  37                    'rm', '/data/misc/boottrace/categories'])
  38
  39 TEST_DIR = os.path.join(os.path.dirname(__file__), os.pardir, 'test_data')
  40 ATRACE_DATA = os.path.join(TEST_DIR, 'atrace_data')
  41 ATRACE_DATA_RAW = os.path.join(TEST_DIR, 'atrace_data_raw')
  42 ATRACE_DATA_RAW_FROM_FILE = os.path.join(TEST_DIR, 'atrace_data_raw_from_file')
  43 ATRACE_DATA_STRIPPED = os.path.join(TEST_DIR, 'atrace_data_stripped')
  44 ATRACE_DATA_THREAD_FIXED = os.path.join(TEST_DIR, 'atrace_data_thread_fixed')
  45 ATRACE_DATA_WITH_THREAD_LIST = os.path.join(TEST_DIR,
  46                                             'atrace_data_with_thread_list')
  47 ATRACE_THREAD_NAMES = os.path.join(TEST_DIR, 'atrace_thread_names')
  48 ATRACE_THREAD_LIST = os.path.join(TEST_DIR, 'atrace_ps_dump')
  49 ATRACE_EXTRACTED_THREADS = os.path.join(TEST_DIR, 'atrace_extracted_threads')
  50 ATRACE_PROCFS_DUMP = os.path.join(TEST_DIR, 'atrace_procfs_dump')
  51 ATRACE_EXTRACTED_TGIDS = os.path.join(TEST_DIR, 'atrace_extracted_tgids')
  52 ATRACE_MISSING_TGIDS = os.path.join(TEST_DIR, 'atrace_missing_tgids')
  53 ATRACE_FIXED_TGIDS = os.path.join(TEST_DIR, 'atrace_fixed_tgids')
  54
  55
  56 class AtraceAgentTest(unittest.TestCase):
  57
  58   def test_construct_trace_command(self):
  59     options, categories = systrace.parse_options(SYSTRACE_CMD)
  60     agent = atrace_agent.AtraceAgent(options, categories)
  61     tracer_args = agent._construct_trace_command()
  62     self.assertEqual(' '.join(TRACE_CMD), ' '.join(tracer_args))
  63     self.assertEqual(True, agent.expect_trace())
  64
  65   def test_extract_thread_list(self):
  66     with contextlib.nested(open(ATRACE_EXTRACTED_THREADS, 'r'),
  67                            open(ATRACE_THREAD_LIST)) as (f1, f2):
  68
  69       atrace_result = f1.read()
  70       ps_dump = f2.read()
  71
  72       thread_names = atrace_agent.extract_thread_list(ps_dump)
  73       self.assertEqual(atrace_result, 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_STRIPPED, 'r'),
 100                            open(ATRACE_DATA_RAW, 'r')) as (f1, f2):
 101       atrace_data = f1.read()
 102       atrace_data_raw = f2.read()
 103
 104       options, categories = systrace.parse_options(STOP_FIX_UPS)
 105       agent = atrace_agent.AtraceAgent(options, categories)
 106       trace_data = agent._preprocess_trace_data(atrace_data_raw)
 107
 108       self.assertEqual(atrace_data, trace_data)
 109
 110   def test_list_categories(self):
 111     options, categories = systrace.parse_options(SYSTRACE_LIST_CATEGORIES_CMD)
 112     agent = atrace_agent.AtraceAgent(options, categories)
 113     tracer_args = agent._construct_trace_command()
 114     self.assertEqual(' '.join(TRACE_LIST_CATEGORIES_CMD), ' '.join(tracer_args))
 115     self.assertEqual(False, agent.expect_trace())
 116
 117   def test_construct_trace_from_file_command(self):
 118     options, categories = systrace.parse_options(['systrace.py',
 119         '--from-file', ATRACE_DATA_RAW_FROM_FILE])
 120     agent = atrace_agent.try_create_agent(options, categories)
 121     tracer_args = agent._construct_trace_command()
 122     self.assertEqual(' '.join(['cat', ATRACE_DATA_RAW_FROM_FILE]),
 123             ' '.join(tracer_args))
 124     self.assertEqual(True, agent.expect_trace())
 125
 126   def test_extract_tgids(self):
 127     with contextlib.nested(open(ATRACE_PROCFS_DUMP, 'r'),
 128                            open(ATRACE_EXTRACTED_TGIDS, 'r')) as (f1, f2):
 129
 130       atrace_procfs_dump = f1.read()
 131       atrace_procfs_extracted = f2.read()
 132
 133       tgids = eval(atrace_procfs_extracted)
 134       result = atrace_agent.extract_tgids(atrace_procfs_dump)
 135
 136       self.assertEqual(result, tgids)
 137
 138   def test_fix_missing_tgids(self):
 139     with contextlib.nested(open(ATRACE_EXTRACTED_TGIDS, 'r'),
 140                            open(ATRACE_MISSING_TGIDS, 'r'),
 141                            open(ATRACE_FIXED_TGIDS, 'r')) as (f1, f2, f3):
 142
 143       atrace_data = f2.read()
 144       tgid_map = eval(f1.read())
 145       fixed = f3.read()
 146
 147       res = atrace_agent.fix_missing_tgids(atrace_data, tgid_map)
 148       self.assertEqual(res, fixed)
 149
 150
 151 class AtraceLegacyAgentTest(unittest.TestCase):
 152
 153   def test_construct_trace_command(self):
 154     options, categories = systrace.parse_options(SYSTRACE_CMD)
 155     agent = atrace_agent.AtraceLegacyAgent(options, categories)
 156     tracer_args = agent._construct_trace_command()
 157     self.assertEqual(' '.join(LEGACY_TRACE_CMD), ' '.join(tracer_args))
 158     self.assertEqual(True, agent.expect_trace())
 159
 160
 161 class BootAgentTest(unittest.TestCase):
 162
 163   def test_boot(self):
 164     options, categories = systrace.parse_options(SYSTRACE_BOOT_CMD)
 165     agent = atrace_agent.BootAgent(options, categories)
 166     tracer_args = agent._construct_trace_command()
 167     self.assertEqual(' '.join(TRACE_BOOT_CMD), ' '.join(tracer_args))
 168     self.assertEqual(True, agent.expect_trace())