19374862843

Committed 14 Nov 2025 07:06PM UTC coverage: 82.899% (-1.0%) from 83.907%

Build # 19374862843

Build Type

Pull #930

github

Committed by

web-flow

Commit Message

Merge e3fef2469 into fb7abd6b9

Pull Request Pull Request #930: Upgrade coveralls to ~=4.0

Run Details

1388 of 1840 branches covered (75.43%)

Branch coverage included in aggregate %.

0 of 10 new or added lines in 1 file covered. (0.0%)

23 existing lines in 2 files now uncovered.

5510 of 6481 relevant lines covered (85.02%)

39.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

60.93

/src/main/python/pybuilder/python_utils.py

#   -*- coding: utf-8 -*-
#
#   This file is part of PyBuilder
#
#   Copyright 2011-2020 PyBuilder Team
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.

import os
import platform
import sys
import traceback
from collections import OrderedDict
from io import StringIO


def is_windows(platform=sys.platform, win_platforms={"win32", "cygwin", "msys"}):
    return platform in win_platforms


StringIO = StringIO
IS_PYPY = '__pypy__' in sys.builtin_module_names
IS_WIN = is_windows()


def raise_exception(ex, tb):
    raise ex.with_traceback(tb)


def is_string(val):
    return isinstance(val, str)


from shutil import which  # noqa: E402


def save_tb(ex):
    pass


is_string = is_string
makedirs = os.makedirs
which = which

odict = OrderedDict

_mp_get_context = None  # This will be patched at runtime
mp_ForkingPickler = None  # This will be patched at runtime
mp_log_to_stderr = None  # This will be patched at runtime
_mp_billiard_pyb_env = None  # This will be patched at runtime

_old_billiard_spawn_passfds = None  # This will be patched at runtime
_installed_tblib = False

from multiprocessing import log_to_stderr as mp_log_to_stderr, get_context as _mp_get_context  # noqa: E402
from multiprocessing.reduction import ForkingPickler as mp_ForkingPickler  # noqa: E402


def patch_mp_pyb_env(pyb_env):
    global _mp_billiard_pyb_env

    if not _mp_billiard_pyb_env:
        _mp_billiard_pyb_env = pyb_env


def install_tblib():
    global _installed_tblib

    if not _installed_tblib:
        from pybuilder._vendor.tblib import pickling_support

        pickling_support.install()
        _installed_tblib = True


def patch_mp():
    install_tblib()


def mp_get_context(context):
    global _mp_get_context
    return _mp_get_context(context)


mp_ForkingPickler = mp_ForkingPickler
mp_log_to_stderr = mp_log_to_stderr
_mp_get_context = _mp_get_context


def _instrumented_target(q, target, *args, **kwargs):
    patch_mp()

    ex = tb = None
    try:
        send_value = (target(*args, **kwargs), None, None)
    except Exception:
        _, ex, tb = sys.exc_info()
        send_value = (None, ex, tb)

    try:
        q.put(send_value)
    except Exception:
        _, send_ex, send_tb = sys.exc_info()
        e_out = Exception(str(send_ex), send_tb, None if ex is None else str(ex), tb)
        q.put(e_out)


def spawn_process(target=None, args=(), kwargs={}, group=None, name=None):
    """
    Forks a child, making sure that all exceptions from the child are safely sent to the parent
    If a target raises an exception, the exception is re-raised in the parent process
    @return tuple consisting of process exit code and target's return value
    """
    ctx = mp_get_context("spawn")

    q = ctx.SimpleQueue()
    p = ctx.Process(group=group, target=_instrumented_target, name=name, args=[q, target] + list(args), kwargs=kwargs)
    p.start()
    result = q.get()
    p.join()
    if isinstance(result, tuple):
        if result[1]:
            raise_exception(result[1], result[2])
        return p.exitcode, result[0]
    else:
        msg = "Fatal error occurred in the forked process %s: %s" % (p, result.args[0])
        if result.args[2]:
            chained_message = "This error masked the send error '%s':\n%s" % (
                result.args[2], "".join(traceback.format_tb(result.args[3])))
            msg += "\n" + chained_message
        ex = Exception(msg)
        raise_exception(ex, result.args[1])


def prepend_env_to_path(python_env, sys_path):
    """type: (PythonEnv, List(str)) -> None
    Prepend venv directories to sys.path-like collection
    """
    for path in reversed(python_env.site_paths):
        if path not in sys_path:
            sys_path.insert(0, path)


def add_env_to_path(python_env, sys_path):
    """type: (PythonEnv, List(str)) -> None
    Adds venv directories to sys.path-like collection
    """
    for path in python_env.site_paths:
        if path not in sys_path:
            sys_path.append(path)


from glob import glob, iglob, escape  # noqa: E402

from os import symlink  # noqa: E402

symlink = symlink

sys_executable_suffix = sys.executable[len(sys.exec_prefix) + 1:]

python_specific_dir_name = "%s-%s" % (platform.python_implementation().lower(),
                                      ".".join(str(f) for f in sys.version_info))

_, _venv_python_exename = os.path.split(os.path.abspath(getattr(sys, "_base_executable", sys.executable)))

try:
    from imp import load_source
except ImportError:
    from importlib import machinery as importlib_machinery
    from importlib import util as importlib_util
    from importlib._bootstrap import _exec as importlib_exec
    from importlib._bootstrap import _load as importlib_load


    class _HackedGetData:

        """Compatibility support for 'file' arguments of various load_*()
        functions."""

        def __init__(self, fullname, path, file=None):
            super().__init__(fullname, path)
            self.file = file

        def get_data(self, path):
            """Gross hack to contort loader to deal w/ load_*()'s bad API."""
            if self.file and path == self.path:
                # The contract of get_data() requires us to return bytes. Reopen the
                # file in binary mode if needed.
                if not self.file.closed:
                    file = self.file
                    if 'b' not in file.mode:
                        file.close()
                if self.file.closed:
                    self.file = file = open(self.path, 'rb')

                with file:
                    return file.read()
            else:
                return super().get_data(path)


    class _LoadSourceCompatibility(_HackedGetData, importlib_machinery.SourceFileLoader):

        """Compatibility support for implementing load_source()."""


    def load_source(name, pathname, file=None):
        loader = _LoadSourceCompatibility(name, pathname, file)
        spec = importlib_util.spec_from_file_location(name, pathname, loader=loader)
        if name in sys.modules:
            module = importlib_exec(spec, sys.modules[name])
        else:
            module = importlib_load(spec)
        # To allow reloading to potentially work, use a non-hacked loader which
        # won't rely on a now-closed file object.
        module.__loader__ = importlib_machinery.SourceFileLoader(name, pathname)
        module.__spec__.loader = module.__loader__
        return module

__all__ = ["glob", "iglob", "escape"]

1	# -- coding: utf-8 --
2	#
3	# This file is part of PyBuilder
4	#
5	# Copyright 2011-2020 PyBuilder Team
6	#
7	# Licensed under the Apache License, Version 2.0 (the "License");
8	# you may not use this file except in compliance with the License.
9	# You may obtain a copy of the License at
10	#
11	# http://www.apache.org/licenses/LICENSE-2.0
12	#
13	# Unless required by applicable law or agreed to in writing, software
14	# distributed under the License is distributed on an "AS IS" BASIS,
15	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16	# See the License for the specific language governing permissions and
17	# limitations under the License.
18
19	import os	48✔
20	import platform	48✔
21	import sys	48✔
22	import traceback	48✔
23	from collections import OrderedDict	48✔
24	from io import StringIO	48✔
25
26
27	def is_windows(platform=sys.platform, win_platforms={"win32", "cygwin", "msys"}):	48✔
28	return platform in win_platforms	48✔
29
30
31	StringIO = StringIO	48✔
32	IS_PYPY = '__pypy__' in sys.builtin_module_names	48✔
33	IS_WIN = is_windows()	48✔
34
35
36	def raise_exception(ex, tb):	48✔
37	raise ex.with_traceback(tb)	48✔
38
39
40	def is_string(val):	48✔
41	return isinstance(val, str)	48✔
42
43
44	from shutil import which # noqa: E402	48✔
45
46
47	def save_tb(ex):	48✔
48	pass	×
49
50
51	is_string = is_string	48✔
52	makedirs = os.makedirs	48✔
53	which = which	48✔
54
55	odict = OrderedDict	48✔
56
57	_mp_get_context = None # This will be patched at runtime	48✔
58	mp_ForkingPickler = None # This will be patched at runtime	48✔
59	mp_log_to_stderr = None # This will be patched at runtime	48✔
60	_mp_billiard_pyb_env = None # This will be patched at runtime	48✔
61
62	_old_billiard_spawn_passfds = None # This will be patched at runtime	48✔
63	_installed_tblib = False	48✔
64
65	from multiprocessing import log_to_stderr as mp_log_to_stderr, get_context as _mp_get_context # noqa: E402	48✔
66	from multiprocessing.reduction import ForkingPickler as mp_ForkingPickler # noqa: E402	48✔
67
68
69	def patch_mp_pyb_env(pyb_env):	48✔
70	global _mp_billiard_pyb_env
71
72	if not _mp_billiard_pyb_env:	48✔
73	_mp_billiard_pyb_env = pyb_env	48✔
74
75
76	def install_tblib():	48✔
77	global _installed_tblib
78
79	if not _installed_tblib:	48!
80	from pybuilder._vendor.tblib import pickling_support	48✔
81
82	pickling_support.install()	48✔
83	_installed_tblib = True	48✔
84
85
86	def patch_mp():	48✔
87	install_tblib()	48✔
88
89
90	def mp_get_context(context):	48✔
91	global _mp_get_context
92	return _mp_get_context(context)	48✔
93
94
95	mp_ForkingPickler = mp_ForkingPickler	48✔
96	mp_log_to_stderr = mp_log_to_stderr	48✔
97	_mp_get_context = _mp_get_context	48✔
98
99
100	def _instrumented_target(q, target, args, *kwargs):	48✔
101	patch_mp()	×
102
103	ex = tb = None	×
104	try:	×
105	send_value = (target(args, *kwargs), None, None)	×
106	except Exception:	×
107	_, ex, tb = sys.exc_info()	×
108	send_value = (None, ex, tb)	×
109
110	try:	×
111	q.put(send_value)	×
112	except Exception:	×
113	_, send_ex, send_tb = sys.exc_info()	×
114	e_out = Exception(str(send_ex), send_tb, None if ex is None else str(ex), tb)	×
115	q.put(e_out)	×
116
117
118	def spawn_process(target=None, args=(), kwargs={}, group=None, name=None):	48✔
119	"""
120	Forks a child, making sure that all exceptions from the child are safely sent to the parent
121	If a target raises an exception, the exception is re-raised in the parent process
122	@return tuple consisting of process exit code and target's return value
123	"""
124	ctx = mp_get_context("spawn")	×
125
126	q = ctx.SimpleQueue()	×
127	p = ctx.Process(group=group, target=_instrumented_target, name=name, args=[q, target] + list(args), kwargs=kwargs)	×
128	p.start()	×
129	result = q.get()	×
130	p.join()	×
131	if isinstance(result, tuple):	×
132	if result[1]:	×
133	raise_exception(result[1], result[2])	×
134	return p.exitcode, result[0]	×
135	else:
136	msg = "Fatal error occurred in the forked process %s: %s" % (p, result.args[0])	×
137	if result.args[2]:	×
138	chained_message = "This error masked the send error '%s':\n%s" % (	×
139	result.args[2], "".join(traceback.format_tb(result.args[3])))
140	msg += "\n" + chained_message	×
141	ex = Exception(msg)	×
142	raise_exception(ex, result.args[1])	×
143
144
145	def prepend_env_to_path(python_env, sys_path):	48✔
146	"""type: (PythonEnv, List(str)) -> None
147	Prepend venv directories to sys.path-like collection
148	"""
149	for path in reversed(python_env.site_paths):	48✔
150	if path not in sys_path:	48✔
151	sys_path.insert(0, path)	48✔
152
153
154	def add_env_to_path(python_env, sys_path):	48✔
155	"""type: (PythonEnv, List(str)) -> None
156	Adds venv directories to sys.path-like collection
157	"""
158	for path in python_env.site_paths:	×
159	if path not in sys_path:	×
160	sys_path.append(path)	×
161
162
163	from glob import glob, iglob, escape # noqa: E402	48✔
164
165	from os import symlink # noqa: E402	48✔
166
167	symlink = symlink	48✔
168
169	sys_executable_suffix = sys.executable[len(sys.exec_prefix) + 1:]	48✔
170
171	python_specific_dir_name = "%s-%s" % (platform.python_implementation().lower(),	48✔
172	".".join(str(f) for f in sys.version_info))
173
174	_, _venv_python_exename = os.path.split(os.path.abspath(getattr(sys, "_base_executable", sys.executable)))	48✔
175
176	try:	48✔
177	from imp import load_source	48✔
UNCOV 178	except ImportError:	24✔
UNCOV 179	from importlib import machinery as importlib_machinery	24✔
UNCOV 180	from importlib import util as importlib_util	24✔
UNCOV 181	from importlib._bootstrap import _exec as importlib_exec	24✔
UNCOV 182	from importlib._bootstrap import _load as importlib_load	24✔
183
184
UNCOV 185	class _HackedGetData:	24✔
186
187	"""Compatibility support for 'file' arguments of various load_*()
188	functions."""
189
UNCOV 190	def __init__(self, fullname, path, file=None):	24✔
UNCOV 191	super().__init__(fullname, path)	24✔
UNCOV 192	self.file = file	24✔
193
UNCOV 194	def get_data(self, path):	24✔
195	"""Gross hack to contort loader to deal w/ load_*()'s bad API."""
UNCOV 196	if self.file and path == self.path:	24!
197	# The contract of get_data() requires us to return bytes. Reopen the
198	# file in binary mode if needed.
199	if not self.file.closed:	×
200	file = self.file	×
201	if 'b' not in file.mode:	×
202	file.close()	×
203	if self.file.closed:	×
204	self.file = file = open(self.path, 'rb')	×
205
206	with file:	×
207	return file.read()	×
208	else:
UNCOV 209	return super().get_data(path)	24✔
210
211
UNCOV 212	class _LoadSourceCompatibility(_HackedGetData, importlib_machinery.SourceFileLoader):	24✔
213
214	"""Compatibility support for implementing load_source()."""
215
216
UNCOV 217	def load_source(name, pathname, file=None):	24✔
UNCOV 218	loader = _LoadSourceCompatibility(name, pathname, file)	24✔
UNCOV 219	spec = importlib_util.spec_from_file_location(name, pathname, loader=loader)	24✔
UNCOV 220	if name in sys.modules:	24✔
UNCOV 221	module = importlib_exec(spec, sys.modules[name])	24✔
222	else:
UNCOV 223	module = importlib_load(spec)	24✔
224	# To allow reloading to potentially work, use a non-hacked loader which
225	# won't rely on a now-closed file object.
UNCOV 226	module.__loader__ = importlib_machinery.SourceFileLoader(name, pathname)	24✔
UNCOV 227	module.__spec__.loader = module.__loader__	24✔
UNCOV 228	return module	24✔
229
230	__all__ = ["glob", "iglob", "escape"]	48✔

pybuilder / pybuilder / 19374862843

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous