(generate_threads, argument, python, tmpdir)
| 467 | @ALL_PYTHONS |
| 468 | @ALL_SOURCES |
| 469 | def test_output_limiting(generate_threads, argument, python, tmpdir): |
| 470 | # GIVEN |
| 471 | _, python_executable = python |
| 472 | |
| 473 | program_template = """ |
| 474 | import sys |
| 475 | import time |
| 476 | |
| 477 | VAR = {argument} |
| 478 | |
| 479 | def first_func(): |
| 480 | the_local = VAR |
| 481 | second_func(VAR) |
| 482 | |
| 483 | def second_func(the_argument): |
| 484 | fifo = sys.argv[1] |
| 485 | with open(sys.argv[1], "w") as fifo: |
| 486 | fifo.write("ready") |
| 487 | time.sleep(1000) |
| 488 | |
| 489 | first_func() |
| 490 | """ |
| 491 | |
| 492 | script = tmpdir / "the_script.py" |
| 493 | script.write_text(program_template.format(argument=argument), encoding="utf-8") |
| 494 | |
| 495 | # WHEN |
| 496 | |
| 497 | threads = generate_threads(python_executable, script, tmpdir) |
| 498 | |
| 499 | # THEN |
| 500 | |
| 501 | assert len(threads) == 1 |
| 502 | (thread,) = threads |
| 503 | |
| 504 | frames = list(thread.frames) |
| 505 | assert (len(frames)) == 3 |
| 506 | |
| 507 | first_func_frame = find_frame(frames, "first_func") |
| 508 | assert "the_local" in first_func_frame.locals |
| 509 | assert len(first_func_frame.locals["the_local"]) <= MAX_OUTPUT_LEN |
| 510 | assert "..." in first_func_frame.locals["the_local"] |
| 511 | |
| 512 | second_func_frame = find_frame(frames, "second_func") |
| 513 | assert "the_argument" in second_func_frame.arguments |
| 514 | assert len(second_func_frame.arguments["the_argument"]) <= MAX_OUTPUT_LEN |
| 515 | assert "..." in second_func_frame.arguments["the_argument"] |
| 516 | |
| 517 | |
| 518 | @ALL_PYTHONS |
nothing calls this directly
no test coverage detected