File: pyi_threading_module2.py

package info (click to toggle)
pyinstaller 6.13.0%2Bds-2
  • links: PTS, VCS
  • area: main
  • in suites: trixie
  • size: 11,520 kB
  • sloc: python: 41,347; ansic: 11,334; makefile: 176; sh: 136; xml: 19
file content (68 lines) | stat: -rw-r--r-- 2,283 bytes parent folder | download | duplicates (3)
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
#-----------------------------------------------------------------------------
# Copyright (c) 2013-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 bootloader behavior for threading code. The default behavior of Python interpreter is to wait for all threads
# before exiting the main process. Bootloader should behave in the same way.

import os
import sys
import threading

_OUT_EXPECTED = ['ONE', 'TWO', 'THREE']

# Code for the subprocess.
if 'PYI_THREAD_TEST_CASE' in os.environ:

    class TestThreadClass(threading.Thread):
        def __init__(self):
            threading.Thread.__init__(self)

        def run(self):
            print('ONE')
            print('TWO')
            print('THREE')

    # Main process should not exit before the thread stops. This is the behaviour of Python interpreter.
    TestThreadClass().start()

# Execute itself in a subprocess.
else:
    # Differentiate subprocess code.
    itself = sys.argv[0]
    # Run subprocess.
    import subprocess

    # Preserve environment to avoid `Failed to initialize Windows random API (CryptoGen)`
    env = dict(os.environ)
    env['PYI_THREAD_TEST_CASE'] = 'yes'

    proc = subprocess.Popen([itself], stdout=subprocess.PIPE, env=env, stderr=subprocess.PIPE, shell=False)
    # Waits for subprocess to complete.
    out, err = proc.communicate()

    # Make output from subprocess visible.
    print(out)
    out = out.decode('ascii')
    print(out)

    # Remove empty lines from output.
    out = out.strip().splitlines()
    for line in out:
        if not line.strip():  # Empty line detected.
            out.remove(line)
    # Check output.
    if out != _OUT_EXPECTED:
        print(" +++++++ SUBPROCESS ERROR OUTPUT +++++++")
        print(err)
        raise SystemExit(
            'Subprocess did not print ONE, TWO, THREE in correct order. (output was %r, return code was %s)' %
            (out, proc.returncode)
        )