|
8 | 8 |
|
9 | 9 | import os |
10 | 10 | import pstats |
| 11 | +import re |
| 12 | +import subprocess |
11 | 13 | import tempfile |
12 | 14 | import cProfile |
| 15 | +from test.support import os_helper, script_helper |
13 | 16 |
|
14 | 17 | class LazyImportTest(unittest.TestCase): |
15 | 18 | @support.cpython_only |
@@ -155,5 +158,222 @@ def test_SortKey_enum(self): |
155 | 158 | self.assertEqual(SortKey.FILENAME, 'filename') |
156 | 159 | self.assertNotEqual(SortKey.FILENAME, SortKey.CALLS) |
157 | 160 |
|
| 161 | + |
| 162 | +def _profiled_callee(): |
| 163 | + return 1 |
| 164 | + |
| 165 | + |
| 166 | +def _profiled_caller(): |
| 167 | + for _ in range(3): |
| 168 | + _profiled_callee() |
| 169 | + |
| 170 | + |
| 171 | +@support.requires_subprocess() |
| 172 | +class CommandLineTests(unittest.TestCase): |
| 173 | + """Tests for the interactive profile browser (``python -m pstats``).""" |
| 174 | + |
| 175 | + @classmethod |
| 176 | + def setUpClass(cls): |
| 177 | + cls.tmpdir = cls.enterClassContext(os_helper.temp_dir()) |
| 178 | + cls.profile = os.path.join(cls.tmpdir, 'test.prof') |
| 179 | + cProfile.runctx('_profiled_caller()', globals(), {}, |
| 180 | + filename=cls.profile) |
| 181 | + |
| 182 | + def run_browser(self, *args, commands=()): |
| 183 | + """Run ``python -m pstats`` with *args*, feeding *commands* on stdin. |
| 184 | +
|
| 185 | + Return what the browser wrote to stdout. |
| 186 | + """ |
| 187 | + proc = script_helper.spawn_python('-m', 'pstats', *args, |
| 188 | + stderr=subprocess.PIPE) |
| 189 | + stdin = ''.join(f'{command}\n' for command in commands) |
| 190 | + stdout, stderr = proc.communicate(stdin.encode()) |
| 191 | + stdout = stdout.decode('utf-8', 'backslashreplace') |
| 192 | + stderr = stderr.decode('utf-8', 'backslashreplace') |
| 193 | + self.assertEqual(proc.returncode, 0, stderr) |
| 194 | + self.assertEqual(stderr, '') |
| 195 | + return stdout.replace('\r\n', '\n') |
| 196 | + |
| 197 | + def test_start_and_quit(self): |
| 198 | + stdout = self.run_browser(commands=['quit']) |
| 199 | + self.assertEqual(stdout, |
| 200 | + 'Welcome to the profile statistics browser.\n' |
| 201 | + '% Goodbye.\n') |
| 202 | + |
| 203 | + def test_eof(self): |
| 204 | + # Closing stdin leaves the browser, like the "quit" command. |
| 205 | + stdout = self.run_browser(commands=[]) |
| 206 | + self.assertEqual(stdout, |
| 207 | + 'Welcome to the profile statistics browser.\n' |
| 208 | + '% \nGoodbye.\n') |
| 209 | + |
| 210 | + def test_profile_argument(self): |
| 211 | + stdout = self.run_browser(self.profile, commands=['stats', 'quit']) |
| 212 | + # The prompt shows the name of the loaded file. |
| 213 | + self.assertIn(f'\n{self.profile}% ', stdout) |
| 214 | + # So does the header of the report. |
| 215 | + self.assertIn(f' {self.profile}\n', stdout) |
| 216 | + self.assertRegex(stdout, r'\n +\d+ function calls in ') |
| 217 | + self.assertRegex(stdout, r'\n +3 +.*\(_profiled_callee\)\n') |
| 218 | + self.assertRegex(stdout, r'\n +1 +.*\(_profiled_caller\)\n') |
| 219 | + |
| 220 | + def test_multiple_profile_arguments(self): |
| 221 | + # The statistics of all the files given on the command line |
| 222 | + # are added together. |
| 223 | + stdout = self.run_browser(self.profile, self.profile, |
| 224 | + commands=['stats', 'quit']) |
| 225 | + self.assertEqual(stdout.count(f' {self.profile}\n'), 2) |
| 226 | + self.assertRegex(stdout, r'\n +6 +.*\(_profiled_callee\)\n') |
| 227 | + |
| 228 | + def test_missing_profile_argument(self): |
| 229 | + missing = os.path.join(self.tmpdir, 'missing.prof') |
| 230 | + stdout = self.run_browser(missing, commands=['stats', 'quit']) |
| 231 | + self.assertIn('No such file or directory\n', stdout) |
| 232 | + self.assertIn('\n% No statistics object is loaded.\n', stdout) |
| 233 | + |
| 234 | + def test_commands_without_statistics(self): |
| 235 | + for command in ('stats', 'callers', 'callees', 'sort calls', |
| 236 | + 'strip', 'reverse', f'add {self.profile}'): |
| 237 | + with self.subTest(command=command): |
| 238 | + stdout = self.run_browser(commands=[command, 'quit']) |
| 239 | + self.assertIn('\n% No statistics object is loaded.\n', |
| 240 | + stdout) |
| 241 | + stdout = self.run_browser(commands=['read', 'quit']) |
| 242 | + self.assertIn( |
| 243 | + '\n% No statistics object is current -- cannot reload.\n', stdout) |
| 244 | + |
| 245 | + def test_read(self): |
| 246 | + stdout = self.run_browser(commands=[f'read {self.profile}', |
| 247 | + 'stats', 'quit']) |
| 248 | + self.assertIn(f'\n{self.profile}% ', stdout) |
| 249 | + self.assertRegex(stdout, r'\n +3 +.*\(_profiled_callee\)\n') |
| 250 | + |
| 251 | + def test_read_errors(self): |
| 252 | + missing = os.path.join(self.tmpdir, 'missing.prof') |
| 253 | + stdout = self.run_browser(commands=[f'read {missing}', 'quit']) |
| 254 | + # The error is reported and no file is loaded. |
| 255 | + self.assertEqual(stdout, |
| 256 | + 'Welcome to the profile statistics browser.\n' |
| 257 | + '% No such file or directory\n' |
| 258 | + '% Goodbye.\n') |
| 259 | + |
| 260 | + empty = os.path.join(self.tmpdir, 'empty.prof') |
| 261 | + os_helper.create_empty_file(empty) |
| 262 | + stdout = self.run_browser(commands=[f'read {empty}', 'quit']) |
| 263 | + self.assertIn('\n% EOFError: ', stdout) |
| 264 | + self.assertIn('\n% Goodbye.\n', stdout) |
| 265 | + |
| 266 | + def test_add(self): |
| 267 | + stdout = self.run_browser(self.profile, |
| 268 | + commands=[f'add {self.profile}', |
| 269 | + 'stats', 'quit']) |
| 270 | + self.assertEqual(stdout.count(f' {self.profile}\n'), 2) |
| 271 | + self.assertRegex(stdout, r'\n +6 +.*\(_profiled_callee\)\n') |
| 272 | + |
| 273 | + def test_add_errors(self): |
| 274 | + missing = os.path.join(self.tmpdir, 'missing.prof') |
| 275 | + stdout = self.run_browser(self.profile, |
| 276 | + commands=[f'add {missing}', 'quit']) |
| 277 | + self.assertIn(f'% Failed to load statistics for {missing}: ', stdout) |
| 278 | + |
| 279 | + def test_stats_restrictions(self): |
| 280 | + stdout = self.run_browser(self.profile, |
| 281 | + commands=['stats 2', 'stats 0.5', |
| 282 | + 'stats _profiled_callee', 'quit']) |
| 283 | + self.assertRegex(stdout, |
| 284 | + r'List reduced from \d+ to 2 due to restriction <2>') |
| 285 | + self.assertRegex(stdout, r'List reduced from \d+ to \d+ ' |
| 286 | + r'due to restriction <0\.5>') |
| 287 | + self.assertRegex(stdout, |
| 288 | + r"List reduced from \d+ to 1 due to restriction " |
| 289 | + r"<'_profiled_callee'>") |
| 290 | + by_name = stdout.split("<'_profiled_callee'>")[1] |
| 291 | + self.assertIn('(_profiled_callee)', by_name) |
| 292 | + self.assertNotIn('(_profiled_caller)', by_name) |
| 293 | + |
| 294 | + def test_stats_invalid_fraction(self): |
| 295 | + stdout = self.run_browser(self.profile, |
| 296 | + commands=['stats 1.5', 'quit']) |
| 297 | + self.assertIn('% Fraction argument must be in [0, 1]\n', stdout) |
| 298 | + # The invalid argument is ignored and the full report is printed. |
| 299 | + self.assertIn('(_profiled_caller)', stdout) |
| 300 | + self.assertIn('(_profiled_callee)', stdout) |
| 301 | + |
| 302 | + def test_sort(self): |
| 303 | + # Unique prefixes of the sort keys are accepted. |
| 304 | + stdout = self.run_browser(self.profile, |
| 305 | + commands=['sort calls', 'stats', |
| 306 | + 'sort cumu', 'stats', 'quit']) |
| 307 | + self.assertIn('Ordered by: call count', stdout) |
| 308 | + self.assertIn('Ordered by: cumulative time', stdout) |
| 309 | + by_calls = stdout.split('Ordered by: ')[1] |
| 310 | + self.assertLess(by_calls.index('(_profiled_callee)'), |
| 311 | + by_calls.index('(_profiled_caller)')) |
| 312 | + |
| 313 | + def test_sort_without_valid_keys(self): |
| 314 | + for command in ('sort', 'sort bogus'): |
| 315 | + with self.subTest(command=command): |
| 316 | + stdout = self.run_browser(self.profile, |
| 317 | + commands=[command, 'quit']) |
| 318 | + self.assertIn( |
| 319 | + 'Valid sort keys (unique prefixes are accepted):', stdout) |
| 320 | + self.assertIn('\ncalls -- call count\n', stdout) |
| 321 | + self.assertIn('\ncumulative -- cumulative time\n', stdout) |
| 322 | + |
| 323 | + def test_reverse(self): |
| 324 | + stdout = self.run_browser(self.profile, |
| 325 | + commands=['sort calls', 'reverse', |
| 326 | + 'stats', 'quit']) |
| 327 | + self.assertIn('Ordered by: call count', stdout) |
| 328 | + self.assertLess(stdout.index('(_profiled_caller)'), |
| 329 | + stdout.index('(_profiled_callee)')) |
| 330 | + |
| 331 | + def test_strip(self): |
| 332 | + filename = _profiled_callee.__code__.co_filename |
| 333 | + basename = os.path.basename(filename) |
| 334 | + stdout = self.run_browser(self.profile, |
| 335 | + commands=['stats _profiled_callee', 'strip', |
| 336 | + 'stats _profiled_callee', 'quit']) |
| 337 | + before, after = stdout.split('due to restriction')[1:] |
| 338 | + self.assertIn(f'{filename}:', before) |
| 339 | + # Only the base name is left after "strip". |
| 340 | + self.assertRegex( |
| 341 | + after, r' ' + re.escape(basename) + r':\d+\(_profiled_callee\)') |
| 342 | + |
| 343 | + def test_read_reloads_current_file(self): |
| 344 | + # "read" without an argument reloads the current file, which |
| 345 | + # undoes "strip". |
| 346 | + filename = _profiled_callee.__code__.co_filename |
| 347 | + stdout = self.run_browser(self.profile, |
| 348 | + commands=['strip', 'read', |
| 349 | + 'stats _profiled_callee', 'quit']) |
| 350 | + self.assertIn(f'{filename}:', stdout.split('due to restriction')[1]) |
| 351 | + |
| 352 | + def test_callers_and_callees(self): |
| 353 | + stdout = self.run_browser(self.profile, |
| 354 | + commands=['callers _profiled_callee', |
| 355 | + 'callees _profiled_caller', |
| 356 | + 'quit']) |
| 357 | + self.assertIn('was called by...', stdout) |
| 358 | + self.assertRegex( |
| 359 | + stdout, r'\(_profiled_callee\) +<- +3 .*\(_profiled_caller\)\n') |
| 360 | + self.assertIn('called...', stdout) |
| 361 | + self.assertRegex( |
| 362 | + stdout, r'\(_profiled_caller\) +-> +3 .*\(_profiled_callee\)\n') |
| 363 | + |
| 364 | + def test_help(self): |
| 365 | + stdout = self.run_browser(commands=['help', 'help stats', 'quit']) |
| 366 | + lines = stdout.splitlines() |
| 367 | + header = 'Documented commands (type help <topic>):' |
| 368 | + self.assertIn(header, lines) |
| 369 | + commands = lines[lines.index(header) + 2] |
| 370 | + self.assertEqual(commands.split(), |
| 371 | + ['EOF', 'add', 'callees', 'callers', 'help', 'quit', |
| 372 | + 'read', 'reverse', 'sort', 'stats', 'strip']) |
| 373 | + self.assertIn('\n% Print statistics from the current stat object.\n', |
| 374 | + stdout) |
| 375 | + self.assertIn('\nArguments may be:\n', stdout) |
| 376 | + |
| 377 | + |
158 | 378 | if __name__ == "__main__": |
159 | 379 | unittest.main() |
0 commit comments