1#!/usr/bin/env python3 2# Copyright 2021 The Pigweed Authors 3# 4# Licensed under the Apache License, Version 2.0 (the "License"); you may not 5# use this file except in compliance with the License. You may obtain a copy of 6# the License at 7# 8# https://www.apache.org/licenses/LICENSE-2.0 9# 10# Unless required by applicable law or agreed to in writing, software 11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 13# License for the specific language governing permissions and limitations under 14# the License. 15"""Tests for the thread analyzer.""" 16 17import unittest 18from pw_thread.thread_analyzer import ThreadInfo, ThreadSnapshotAnalyzer 19from pw_thread_protos import thread_pb2 20 21 22class ThreadInfoTest(unittest.TestCase): 23 """Tests that the ThreadInfo class produces expected results.""" 24 def test_empty_thread(self): 25 thread_info = ThreadInfo(thread_pb2.Thread()) 26 expected = '\n'.join( 27 ('Est CPU usage: unknown', 'Stack info', 28 ' Current usage: 0x???????? - 0x???????? (size unknown)', 29 ' Est peak usage: size unknown', 30 ' Stack limits: 0x???????? - 0x???????? (size unknown)')) 31 self.assertFalse(thread_info.has_stack_size_limit()) 32 self.assertFalse(thread_info.has_stack_used()) 33 self.assertEqual(expected, str(thread_info)) 34 35 def test_thread_with_cpu_usage(self): 36 thread = thread_pb2.Thread() 37 thread.cpu_usage_hundredths = 1234 38 thread_info = ThreadInfo(thread) 39 40 expected = '\n'.join( 41 ('Est CPU usage: 12.34%', 'Stack info', 42 ' Current usage: 0x???????? - 0x???????? (size unknown)', 43 ' Est peak usage: size unknown', 44 ' Stack limits: 0x???????? - 0x???????? (size unknown)')) 45 self.assertFalse(thread_info.has_stack_size_limit()) 46 self.assertFalse(thread_info.has_stack_used()) 47 self.assertEqual(expected, str(thread_info)) 48 49 def test_thread_with_stack_pointer(self): 50 thread = thread_pb2.Thread() 51 thread.stack_pointer = 0x5AC6A86C 52 thread_info = ThreadInfo(thread) 53 54 expected = '\n'.join( 55 ('Est CPU usage: unknown', 'Stack info', 56 ' Current usage: 0x???????? - 0x5ac6a86c (size unknown)', 57 ' Est peak usage: size unknown', 58 ' Stack limits: 0x???????? - 0x???????? (size unknown)')) 59 self.assertFalse(thread_info.has_stack_size_limit()) 60 self.assertFalse(thread_info.has_stack_used()) 61 self.assertEqual(expected, str(thread_info)) 62 63 def test_thread_with_stack_usage(self): 64 thread = thread_pb2.Thread() 65 thread.stack_start_pointer = 0x5AC6B86C 66 thread.stack_pointer = 0x5AC6A86C 67 thread_info = ThreadInfo(thread) 68 69 expected = '\n'.join( 70 ('Est CPU usage: unknown', 'Stack info', 71 ' Current usage: 0x5ac6b86c - 0x5ac6a86c (4096 bytes)', 72 ' Est peak usage: size unknown', 73 ' Stack limits: 0x5ac6b86c - 0x???????? (size unknown)')) 74 self.assertFalse(thread_info.has_stack_size_limit()) 75 self.assertTrue(thread_info.has_stack_used()) 76 self.assertEqual(expected, str(thread_info)) 77 78 def test_thread_with_all_stack_info(self): 79 thread = thread_pb2.Thread() 80 thread.stack_start_pointer = 0x5AC6B86C 81 thread.stack_end_pointer = 0x5AC6986C 82 thread.stack_pointer = 0x5AC6A86C 83 thread_info = ThreadInfo(thread) 84 85 expected = '\n'.join( 86 ('Est CPU usage: unknown', 'Stack info', 87 ' Current usage: 0x5ac6b86c - 0x5ac6a86c (4096 bytes, 50.00%)', 88 ' Est peak usage: size unknown', 89 ' Stack limits: 0x5ac6b86c - 0x5ac6986c (8192 bytes)')) 90 self.assertTrue(thread_info.has_stack_size_limit()) 91 self.assertTrue(thread_info.has_stack_used()) 92 self.assertEqual(expected, str(thread_info)) 93 94 95class ThreadSnapshotAnalyzerTest(unittest.TestCase): 96 """Tests that the ThreadSnapshotAnalyzer class produces expected results.""" 97 def test_no_threads(self): 98 analyzer = ThreadSnapshotAnalyzer(thread_pb2.SnapshotThreadInfo()) 99 self.assertEqual('', str(analyzer)) 100 101 def test_one_empty_thread(self): 102 snapshot = thread_pb2.SnapshotThreadInfo() 103 snapshot.threads.append(thread_pb2.Thread()) 104 expected = '\n'.join(( 105 'Thread State', 106 ' 1 thread running.', 107 '', 108 'Thread (UNKNOWN): [unnamed thread]', 109 'Est CPU usage: unknown', 110 'Stack info', 111 ' Current usage: 0x???????? - 0x???????? (size unknown)', 112 ' Est peak usage: size unknown', 113 ' Stack limits: 0x???????? - 0x???????? (size unknown)', 114 '', 115 )) 116 analyzer = ThreadSnapshotAnalyzer(snapshot) 117 self.assertEqual(analyzer.active_thread(), None) 118 self.assertEqual(str(ThreadSnapshotAnalyzer(snapshot)), expected) 119 120 def test_two_threads(self): 121 """Ensures multiple threads are printed correctly.""" 122 snapshot = thread_pb2.SnapshotThreadInfo() 123 124 temp_thread = thread_pb2.Thread() 125 temp_thread.name = 'Idle'.encode() 126 temp_thread.state = thread_pb2.ThreadState.Enum.READY 127 temp_thread.stack_start_pointer = 0x2001ac00 128 temp_thread.stack_end_pointer = 0x2001aa00 129 temp_thread.stack_pointer = 0x2001ab0c 130 temp_thread.stack_pointer_est_peak = 0x2001aa00 131 snapshot.threads.append(temp_thread) 132 133 temp_thread = thread_pb2.Thread() 134 temp_thread.name = 'Alice'.encode() 135 temp_thread.stack_start_pointer = 0x2001b000 136 temp_thread.stack_pointer = 0x2001ae20 137 temp_thread.state = thread_pb2.ThreadState.Enum.BLOCKED 138 snapshot.threads.append(temp_thread) 139 140 expected = '\n'.join(( 141 'Thread State', 142 ' 2 threads running.', 143 '', 144 'Thread (READY): Idle', 145 'Est CPU usage: unknown', 146 'Stack info', 147 ' Current usage: 0x2001ac00 - 0x2001ab0c (244 bytes, 47.66%)', 148 ' Est peak usage: 512 bytes, 100.00%', 149 ' Stack limits: 0x2001ac00 - 0x2001aa00 (512 bytes)', 150 '', 151 'Thread (BLOCKED): Alice', 152 'Est CPU usage: unknown', 153 'Stack info', 154 ' Current usage: 0x2001b000 - 0x2001ae20 (480 bytes)', 155 ' Est peak usage: size unknown', 156 ' Stack limits: 0x2001b000 - 0x???????? (size unknown)', 157 '', 158 )) 159 analyzer = ThreadSnapshotAnalyzer(snapshot) 160 self.assertEqual(analyzer.active_thread(), None) 161 self.assertEqual(str(ThreadSnapshotAnalyzer(snapshot)), expected) 162 163 def test_interrupts_with_thread(self): 164 """Ensures interrupts are properly reported as active.""" 165 snapshot = thread_pb2.SnapshotThreadInfo() 166 167 temp_thread = thread_pb2.Thread() 168 temp_thread.name = 'Idle'.encode() 169 temp_thread.state = thread_pb2.ThreadState.Enum.READY 170 temp_thread.stack_start_pointer = 0x2001ac00 171 temp_thread.stack_end_pointer = 0x2001aa00 172 temp_thread.stack_pointer = 0x2001ab0c 173 temp_thread.stack_pointer_est_peak = 0x2001aa00 174 snapshot.threads.append(temp_thread) 175 176 temp_thread = thread_pb2.Thread() 177 temp_thread.name = 'Main/Handler'.encode() 178 temp_thread.stack_start_pointer = 0x2001b000 179 temp_thread.stack_pointer = 0x2001ae20 180 temp_thread.state = thread_pb2.ThreadState.Enum.INTERRUPT_HANDLER 181 snapshot.threads.append(temp_thread) 182 183 expected = '\n'.join(( 184 'Thread State', 185 ' 2 threads running, Main/Handler active at the time of capture.', 186 ' ~~~~~~~~~~~~', 187 '', 188 # Ensure the active thread is moved to the top of the list. 189 'Thread (INTERRUPT_HANDLER): Main/Handler <-- [ACTIVE]', 190 'Est CPU usage: unknown', 191 'Stack info', 192 ' Current usage: 0x2001b000 - 0x2001ae20 (480 bytes)', 193 ' Est peak usage: size unknown', 194 ' Stack limits: 0x2001b000 - 0x???????? (size unknown)', 195 '', 196 'Thread (READY): Idle', 197 'Est CPU usage: unknown', 198 'Stack info', 199 ' Current usage: 0x2001ac00 - 0x2001ab0c (244 bytes, 47.66%)', 200 ' Est peak usage: 512 bytes, 100.00%', 201 ' Stack limits: 0x2001ac00 - 0x2001aa00 (512 bytes)', 202 '', 203 )) 204 analyzer = ThreadSnapshotAnalyzer(snapshot) 205 self.assertEqual(analyzer.active_thread(), temp_thread) 206 self.assertEqual(str(ThreadSnapshotAnalyzer(snapshot)), expected) 207 208 def test_active_thread(self): 209 """Ensures the 'active' thread is highlighted.""" 210 snapshot = thread_pb2.SnapshotThreadInfo() 211 212 temp_thread = thread_pb2.Thread() 213 temp_thread.name = 'Idle'.encode() 214 temp_thread.state = thread_pb2.ThreadState.Enum.READY 215 temp_thread.stack_start_pointer = 0x2001ac00 216 temp_thread.stack_end_pointer = 0x2001aa00 217 temp_thread.stack_pointer = 0x2001ab0c 218 temp_thread.stack_pointer_est_peak = 0x2001ac00 + 0x100 219 snapshot.threads.append(temp_thread) 220 221 temp_thread = thread_pb2.Thread() 222 temp_thread.name = 'Main/Handler'.encode() 223 temp_thread.active = True 224 temp_thread.stack_start_pointer = 0x2001b000 225 temp_thread.stack_pointer = 0x2001ae20 226 temp_thread.stack_pointer_est_peak = 0x2001b000 + 0x200 227 temp_thread.state = thread_pb2.ThreadState.Enum.INTERRUPT_HANDLER 228 snapshot.threads.append(temp_thread) 229 230 expected = '\n'.join(( 231 'Thread State', 232 ' 2 threads running, Main/Handler active at the time of capture.', 233 ' ~~~~~~~~~~~~', 234 '', 235 # Ensure the active thread is moved to the top of the list. 236 'Thread (INTERRUPT_HANDLER): Main/Handler <-- [ACTIVE]', 237 'Est CPU usage: unknown', 238 'Stack info', 239 ' Current usage: 0x2001b000 - 0x2001ae20 (480 bytes)', 240 ' Est peak usage: 512 bytes', 241 ' Stack limits: 0x2001b000 - 0x???????? (size unknown)', 242 '', 243 'Thread (READY): Idle', 244 'Est CPU usage: unknown', 245 'Stack info', 246 ' Current usage: 0x2001ac00 - 0x2001ab0c (244 bytes, 47.66%)', 247 ' Est peak usage: 256 bytes, 50.00%', 248 ' Stack limits: 0x2001ac00 - 0x2001aa00 (512 bytes)', 249 '', 250 )) 251 analyzer = ThreadSnapshotAnalyzer(snapshot) 252 253 # Ensure the active thread is found. 254 self.assertEqual(analyzer.active_thread(), temp_thread) 255 self.assertEqual(str(ThreadSnapshotAnalyzer(snapshot)), expected) 256 257 258if __name__ == '__main__': 259 unittest.main() 260