1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
|
#-----------------------------------------------------------------------------
# Copyright (c) 2021-2023, PyInstaller Development Team.
#
# Distributed under the terms of the GNU General Public License (version 2
# or later) with exception for distributing the bootloader.
#
# The full license is in the file COPYING.txt, distributed with this software.
#
# SPDX-License-Identifier: (GPL-2.0-or-later WITH Bootloader-exception)
#-----------------------------------------------------------------------------
"""
Test for unbuffered stdio (stdout/stderr) mode.
"""
import os
import asyncio
import pytest
from PyInstaller.compat import is_win
@pytest.mark.parametrize('stream_mode', ['binary', 'text'])
@pytest.mark.parametrize('output_stream', ['stdout', 'stderr'])
def test_unbuffered_stdio(tmp_path, output_stream, stream_mode, pyi_builder_spec):
# Freeze the test program; test_spec() builds the app and runs it, so explicitly set the number of
# stars to 0 for this run.
pyi_builder_spec.test_spec('pyi_unbuffered_output.spec', app_args=['--num-stars', '0'])
# Path to the frozen executable
executable = os.path.join(tmp_path, 'dist', 'pyi_unbuffered_output', 'pyi_unbuffered_output')
# Expected number of stars
EXPECTED_STARS = 5
# Run the test program via asyncio.SubprocessProtocol and monitor the output.
class SubprocessDotCounter(asyncio.SubprocessProtocol):
def __init__(self, loop, output='stdout'):
self.count = 0
self.loop = loop
# Select stdout vs stderr
assert output in {'stdout', 'stderr'}
self.out_fd = 1 if output == 'stdout' else 2
def pipe_data_received(self, fd, data):
if fd == self.out_fd:
# Treat any data batch that does not end with the * as irregularity
if not data.endswith(b'*'):
return
self.count += data.count(b'*')
def connection_lost(self, exc):
self.loop.stop() # end loop.run_forever()
# Create event loop
if is_win:
loop = asyncio.ProactorEventLoop() # for subprocess' pipes on Windows
else:
loop = asyncio.SelectorEventLoop()
asyncio.set_event_loop(loop)
counter_proto = SubprocessDotCounter(loop, output=output_stream)
# Run
try:
proc = loop.subprocess_exec(
lambda: counter_proto,
executable,
"--num-stars", str(EXPECTED_STARS),
"--output-stream", output_stream,
"--stream-mode", stream_mode
) # yapf: disable
transport, _ = loop.run_until_complete(proc)
loop.run_forever()
finally:
loop.close()
transport.close()
# Check the number of received stars
assert counter_proto.count == EXPECTED_STARS
|
