Function test_script_streams_continuously

tests/test_magic.py:1579–1607

(capsys)

@pytest.mark.asyncio
async def test_script_streams_continuously(capsys):
    ip = get_ipython()
    # Windows is slow to start up a thread on CI
    is_windows = os.name == "nt"
    step = 3 if is_windows else 1
    code = dedent(
        f"""\
    import time
    for _ in range(3):
        time.sleep({step})
        print(".", flush=True, end="")
    """
    )

    def print_numbers():
        for i in range(3):
            sleep(step)
            print(i, flush=True, end="")

    thread = Thread(target=print_numbers)
    thread.start()
    sleep(step / 2)
    ip.run_cell_magic("script", sys.executable, code)
    thread.join()

    captured = capsys.readouterr()
    # If the streaming was line-wise or broken
    # we would get `012...`
    assert captured.out == "0.1.2."


def test_script_streams_multibyte_unicode(capsys):