(self, subtests)
| 227 | assert len(veryslow) < 5, f"Too many slow frame draw times: {veryslow}" |
| 228 | |
| 229 | def test_cpu_usage(self, subtests): |
| 230 | print("\n------------------------------------------------") |
| 231 | print("------------------ CPU Usage -------------------") |
| 232 | print("------------------------------------------------") |
| 233 | |
| 234 | plogs_by_proc = defaultdict(list) |
| 235 | for pl in self.msgs['procLog']: |
| 236 | for x in pl.procLog.procs: |
| 237 | if len(x.cmdline) > 0: |
| 238 | n = list(x.cmdline)[0] |
| 239 | plogs_by_proc[n].append(x) |
| 240 | |
| 241 | cpu_ok = True |
| 242 | dt = (self.msgs['procLog'][-1].logMonoTime - self.msgs['procLog'][0].logMonoTime) / 1e9 |
| 243 | header = ['process', 'usage', 'expected', 'max allowed', 'test result'] |
| 244 | rows = [] |
| 245 | for proc_name, expected in PROCS.items(): |
| 246 | |
| 247 | error = "" |
| 248 | usage = 0. |
| 249 | x = plogs_by_proc[proc_name] |
| 250 | if len(x) > 2: |
| 251 | cpu_time = cputime_total(x[-1]) - cputime_total(x[0]) |
| 252 | usage = cpu_time / dt * 100. |
| 253 | |
| 254 | max_allowed = max(expected * 1.8, expected + 5.0) |
| 255 | if usage > max_allowed: |
| 256 | error = "❌ USING MORE CPU THAN EXPECTED ❌" |
| 257 | cpu_ok = False |
| 258 | |
| 259 | else: |
| 260 | error = "❌ NO METRICS FOUND ❌" |
| 261 | cpu_ok = False |
| 262 | |
| 263 | rows.append([proc_name, usage, expected, max_allowed, error or "✅"]) |
| 264 | print(tabulate(rows, header, tablefmt="simple_grid", stralign="center", numalign="center", floatfmt=".2f")) |
| 265 | |
| 266 | # Ensure there's no missing procs |
| 267 | all_procs = {p.name for p in self.msgs['managerState'][0].managerState.processes if p.shouldBeRunning} |
| 268 | for p in all_procs: |
| 269 | with subtests.test(proc=p): |
| 270 | assert any(p in pp for pp in PROCS.keys()), f"Expected CPU usage missing for {p}" |
| 271 | |
| 272 | # total CPU check |
| 273 | procs_tot = sum([(max(x) if isinstance(x, tuple) else x) for x in PROCS.values()]) |
| 274 | with subtests.test(name="total CPU"): |
| 275 | assert procs_tot < MAX_TOTAL_CPU, "Total CPU budget exceeded" |
| 276 | print("------------------------------------------------") |
| 277 | print(f"Total allocated CPU usage is {procs_tot}%, budget is {MAX_TOTAL_CPU}%, {MAX_TOTAL_CPU-procs_tot:.1f}% left") |
| 278 | print("------------------------------------------------") |
| 279 | |
| 280 | assert cpu_ok |
| 281 | |
| 282 | def test_memory_usage(self): |
| 283 | print("\n------------------------------------------------") |
nothing calls this directly
no test coverage detected