1# Copyright 2016 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. 4import json 5import sys 6import time 7import unittest 8 9from telemetry import decorators 10from telemetry.internal.platform.tracing_agent import cpu_tracing_agent 11from telemetry.internal.platform import tracing_agent 12from telemetry.internal.platform import linux_platform_backend 13from telemetry.internal.platform import mac_platform_backend 14from telemetry.internal.platform import win_platform_backend 15from telemetry.timeline import tracing_config 16from tracing.trace_data import trace_data 17 18 19SNAPSHOT_KEYS = ['pid', 'ppid', 'name', 'pCpu', 'pMem'] 20TRACE_EVENT_KEYS = ['name', 'tid', 'pid', 'ph', 'args', 'local', 'id', 'ts'] 21 22 23class FakeAndroidPlatformBackend(object): 24 def __init__(self): 25 self.device = 'fake_device' 26 27 def GetOSName(self): 28 return 'android' 29 30 31class CpuTracingAgentTest(unittest.TestCase): 32 33 def setUp(self): 34 self._config = tracing_config.TracingConfig() 35 self._config.enable_cpu_trace = True 36 if sys.platform.startswith('win'): 37 self._desktop_backend = win_platform_backend.WinPlatformBackend() 38 elif sys.platform.startswith('darwin'): 39 self._desktop_backend = mac_platform_backend.MacPlatformBackend() 40 else: 41 self._desktop_backend = linux_platform_backend.LinuxPlatformBackend() 42 self._agent = cpu_tracing_agent.CpuTracingAgent(self._desktop_backend) 43 44 @decorators.Enabled('linux', 'mac', 'win') 45 def testInit(self): 46 self.assertTrue(isinstance(self._agent, 47 tracing_agent.TracingAgent)) 48 self.assertFalse(self._agent._snapshots) 49 self.assertFalse(self._agent._snapshot_ongoing) 50 51 @decorators.Enabled('linux', 'mac', 'win') 52 def testIsSupported(self): 53 self.assertTrue(cpu_tracing_agent.CpuTracingAgent.IsSupported( 54 self._desktop_backend)) 55 self.assertFalse(cpu_tracing_agent.CpuTracingAgent.IsSupported( 56 FakeAndroidPlatformBackend())) 57 58 @decorators.Enabled('linux', 'mac', 'win') 59 def testStartAgentTracing(self): 60 self.assertFalse(self._agent._snapshot_ongoing) 61 self.assertFalse(self._agent._snapshots) 62 self.assertTrue(self._agent.StartAgentTracing(self._config, 0)) 63 self.assertTrue(self._agent._snapshot_ongoing) 64 time.sleep(2) 65 self.assertTrue(self._agent._snapshots) 66 self._agent.StopAgentTracing() 67 68 @decorators.Enabled('linux', 'mac', 'win') 69 def testStartAgentTracingNotEnabled(self): 70 self._config.enable_cpu_trace = False 71 self.assertFalse(self._agent._snapshot_ongoing) 72 self.assertFalse(self._agent.StartAgentTracing(self._config, 0)) 73 self.assertFalse(self._agent._snapshot_ongoing) 74 self.assertFalse(self._agent._snapshots) 75 time.sleep(2) 76 self.assertFalse(self._agent._snapshots) 77 78 @decorators.Enabled('linux', 'mac', 'win') 79 def testStopAgentTracingBeforeStart(self): 80 self.assertRaises(AssertionError, self._agent.StopAgentTracing) 81 82 @decorators.Enabled('linux', 'mac', 'win') 83 def testStopAgentTracing(self): 84 self._agent.StartAgentTracing(self._config, 0) 85 self._agent.StopAgentTracing() 86 self.assertFalse(self._agent._snapshot_ongoing) 87 88 @decorators.Enabled('linux', 'mac', 'win') 89 def testCollectAgentTraceDataBeforeStop(self): 90 self._agent.StartAgentTracing(self._config, 0) 91 self.assertRaises(AssertionError, self._agent.CollectAgentTraceData, 92 trace_data.TraceDataBuilder()) 93 self._agent.StopAgentTracing() 94 95 @decorators.Enabled('linux', 'mac', 'win') 96 def testCollectAgentTraceData(self): 97 builder = trace_data.TraceDataBuilder() 98 self._agent.StartAgentTracing(self._config, 0) 99 self._agent.StopAgentTracing() 100 self._agent.CollectAgentTraceData(builder) 101 self.assertFalse(self._agent._snapshot_ongoing) 102 builder = builder.AsData() 103 self.assertTrue(builder.HasTracesFor(trace_data.CPU_TRACE_DATA)) 104 105 @decorators.Enabled('linux', 'mac', 'win') 106 def testCollectAgentTraceDataFormat(self): 107 builder = trace_data.TraceDataBuilder() 108 self._agent.StartAgentTracing(self._config, 0) 109 time.sleep(2) 110 self._agent.StopAgentTracing() 111 self._agent.CollectAgentTraceData(builder) 112 builder = builder.AsData() 113 data = json.loads(builder.GetTracesFor(trace_data.CPU_TRACE_DATA)[0]) 114 self.assertTrue(data) 115 self.assertEquals(set(data[0].keys()), set(TRACE_EVENT_KEYS)) 116 self.assertEquals(set(data[0]['args']['snapshot'].keys()), 117 set(['processes'])) 118 self.assertTrue(data[0]['args']['snapshot']['processes']) 119 self.assertEquals(set(data[0]['args']['snapshot']['processes'][0].keys()), 120 set(SNAPSHOT_KEYS)) 121 122 @decorators.Enabled('linux', 'mac', 'win') 123 def testContainsRealProcesses(self): 124 builder = trace_data.TraceDataBuilder() 125 self._agent.StartAgentTracing(self._config, 0) 126 time.sleep(2) 127 self._agent.StopAgentTracing() 128 self._agent.CollectAgentTraceData(builder) 129 builder = builder.AsData() 130 data = json.loads(builder.GetTracesFor(trace_data.CPU_TRACE_DATA)[0]) 131 self.assertTrue(data) 132 for snapshot in data: 133 found_unittest_process = False 134 processes = snapshot['args']['snapshot']['processes'] 135 for process in processes: 136 if 'run_tests' in process['name']: 137 found_unittest_process = True 138 139 self.assertTrue(found_unittest_process) 140 141 @decorators.Enabled('win') 142 def testWindowsCanHandleProcessesWithSpaces(self): 143 proc_collector = cpu_tracing_agent.WindowsProcessCollector() 144 proc_collector.Init() 145 proc = proc_collector._ParseProcessString( 146 '0 1 Multi Word Process 50 75') 147 self.assertEquals(proc['ppid'], 0) 148 self.assertEquals(proc['pid'], 1) 149 self.assertEquals(proc['name'], 'Multi Word Process') 150 self.assertEquals(proc['pCpu'], 50) 151