19383496280

Committed 15 Nov 2025 03:08AM UTC coverage: 82.899%. Remained the same

Build # 19383496280

Build Type

Pull #934

github

Committed by

web-flow

Commit Message

Merge 9d7f3d7ea into d403e1a82

Pull Request Pull Request #934: Vendorize 2025-11-14

Run Details

1388 of 1840 branches covered (75.43%)

Branch coverage included in aggregate %.

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

71.64

/src/main/python/pybuilder/plugins/python/integrationtest_plugin.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 sys

from pybuilder.core import init, use_plugin, task, description, before
from pybuilder.plugins.python.test_plugin_helper import ReportsProcessor
from pybuilder.python_utils import mp_get_context
from pybuilder.terminal import print_text_line, print_file_content, print_text
from pybuilder.terminal import styled_text, fg, GREEN, MAGENTA, GREY
from pybuilder.utils import discover_files_matching, Timer, read_file

use_plugin("core")

from pybuilder.plugins.python.core_plugin import create_venv  # noqa: E402


@init
def initialize_integrationtest_plugin(project):
    project.set_property_if_unset("dir_source_integrationtest_python", "src/integrationtest/python")

    project.set_property_if_unset("integrationtest_breaks_build", True)
    project.set_property_if_unset("integrationtest_parallel", False)
    project.set_property_if_unset("integrationtest_file_glob", "*_tests.py")
    project.set_property_if_unset("integrationtest_additional_environment", {})
    project.set_property_if_unset("integrationtest_additional_commandline", "")
    project.set_property_if_unset("integrationtest_inherit_environment", False)
    project.set_property_if_unset("integrationtest_always_verbose", False)
    project.set_property_if_unset("integrationtest_cpu_scaling_factor", 4)
    project.set_property_if_unset("integrationtest_python_env", "test")
    project.set_property_if_unset("integrationtest_python_env_recreate", False)

    project.set_property_if_unset("integrationtest_file_suffix", None)  # deprecated, use integrationtest_file_glob.


@before("prepare")
def coverage_init(project, logger, reactor):
    em = reactor.execution_manager

    if em.is_task_in_current_execution_plan("coverage") and em.is_task_in_current_execution_plan(
            "run_integration_tests"):
        project.get_property("_coverage_tasks").append(run_integration_tests)
        project.get_property("_coverage_config_prefixes")[run_integration_tests] = "it"
        project.set_property("it_coverage_name", "Python integration test")
        project.set_property("it_coverage_python_env", project.get_property("integrationtest_python_env"))


@task
@description("Runs integration tests based on Python's unittest module")
def run_integration_tests(project, logger, reactor):
    if project.get_property("integrationtest_parallel"):
        logger.warn("Parallel integration test execution is temporarily disabled")

    # if not project.get_property("integrationtest_parallel"):
    reports, total_time = run_integration_tests_sequentially(project, logger, reactor)
    # else:
    #    reports, total_time = run_integration_tests_in_parallel(project, logger)

    reports_processor = ReportsProcessor(project, logger)
    reports_processor.process_reports(reports, total_time)
    reports_processor.report_to_ci_server(project)
    reports_processor.write_report_and_ensure_all_tests_passed()


def run_integration_tests_sequentially(project, logger, reactor):
    logger.debug("Running integration tests sequentially")
    reports_dir = prepare_reports_directory(project)

    report_items = []

    total_time = Timer.start()

    for test in discover_integration_tests_for_project(project, logger):
        report_item = run_single_test(logger, project, reactor, reports_dir, test)
        report_items.append(report_item)

    total_time.stop()

    return report_items, total_time


def run_integration_tests_in_parallel(project, logger):
    logger.info("Running integration tests in parallel")
    ctx = mp_get_context("spawn")
    tests = ctx.Queue()
    reports = ConsumingQueue(ctx)
    reports_dir = prepare_reports_directory(project)
    cpu_scaling_factor = project.get_property("integrationtest_cpu_scaling_factor")
    cpu_count = ctx.cpu_count()
    worker_pool_size = cpu_count * cpu_scaling_factor
    logger.debug(
        "Running integration tests in parallel with {0} processes ({1} cpus found)".format(
            worker_pool_size,
            cpu_count))

    total_time = Timer.start()
    # fail OSX has no sem_getvalue() implementation so no queue size
    total_tests_count = 0
    for test in discover_integration_tests_for_project(project, logger):
        tests.put(test)
        total_tests_count += 1
    progress = TaskPoolProgress(total_tests_count, worker_pool_size)

    def pick_and_run_tests_then_report(tests, reports, reports_dir, logger, project):
        while True:
            try:
                test = tests.get_nowait()
                report_item = run_single_test(
                    logger, project, reports_dir, test, not progress.can_be_displayed)
                reports.put(report_item)
            except ctx.Empty:
                break
            except Exception as e:
                logger.error("Failed to run test %r : %s" % (test, str(e)))
                failed_report = {
                    "test": test,
                    "test_file": test,
                    "time": 0,
                    "success": False,
                    "exception": str(e)
                }
                reports.put(failed_report)
                continue

    pool = []
    for i in range(worker_pool_size):
        p = ctx.Process(target=pick_and_run_tests_then_report,
                        args=(tests, reports, reports_dir, logger, project))
        pool.append(p)
        p.start()

    import time
    while not progress.is_finished:
        reports.consume_available_items()
        finished_tests_count = reports.size
        progress.update(finished_tests_count)
        progress.render_to_terminal()
        time.sleep(1)

    progress.mark_as_finished()

    total_time.stop()

    return reports.items, total_time


def discover_integration_tests(source_path, suffix=".py"):
    return discover_files_matching(source_path, "*{0}".format(suffix))


def discover_integration_tests_matching(source_path, file_glob):
    return discover_files_matching(source_path, file_glob)


def discover_integration_tests_for_project(project, logger=None):
    integrationtest_source_dir = project.expand_path("$dir_source_integrationtest_python")
    integrationtest_suffix = project.get_property("integrationtest_file_suffix")
    if integrationtest_suffix is not None:
        if logger is not None:
            logger.warn(
                "integrationtest_file_suffix is deprecated, please use integrationtest_file_glob"
            )
        project.set_property("integrationtest_file_glob", "*{0}".format(integrationtest_suffix))
    integrationtest_glob = project.expand("$integrationtest_file_glob")
    return discover_files_matching(integrationtest_source_dir, integrationtest_glob)


def add_additional_environment_keys(env, project):
    additional_environment = project.get_property("integrationtest_additional_environment")

    if not isinstance(additional_environment, dict):
        raise ValueError("Additional environment %r is not a map." %
                         additional_environment)
    for key in additional_environment:
        env[key] = additional_environment[key]


def prepare_environment(project):
    env = {
        "PYTHONPATH": os.pathsep.join((project.expand_path("$dir_dist"),
                                       project.expand_path("$dir_source_integrationtest_python")))
    }

    add_additional_environment_keys(env, project)

    return env


def prepare_reports_directory(project):
    reports_dir = project.expand_path("$dir_reports/integrationtests")
    if not os.path.exists(reports_dir):
        os.mkdir(reports_dir)
    return reports_dir


def run_single_test(logger, project, reactor, reports_dir, test, output_test_names=True):
    additional_integrationtest_commandline_text = project.get_property("integrationtest_additional_commandline")

    if additional_integrationtest_commandline_text:
        additional_integrationtest_commandline = tuple(additional_integrationtest_commandline_text.split(" "))
    else:
        additional_integrationtest_commandline = ()

    name, _ = os.path.splitext(os.path.basename(test))

    if output_test_names:
        logger.info("Running integration test %s", name)

    venv_name = project.get_property("integrationtest_python_env")
    python_env = reactor.python_env_registry[venv_name]
    create_venv(project, logger, reactor, venv_name, True,
                recreate_if_exists=project.get_property("integrationtest_python_env_recreate"))
    env = prepare_environment(project)
    command_and_arguments = python_env.executable + [test]
    command_and_arguments += additional_integrationtest_commandline

    report_file_name = os.path.join(reports_dir, name)
    error_file_name = report_file_name + ".err"

    test_time = Timer.start()
    return_code = python_env.execute_command(command_and_arguments, report_file_name, env,
                                             error_file_name=error_file_name,
                                             inherit_env=project.get_property("integrationtest_inherit_environment"))
    test_time.stop()
    report_item = {
        "test": name,
        "test_file": test,
        "time": test_time.get_millis(),
        "success": True
    }
    if return_code != 0:
        logger.error("Integration test failed: %s, exit code %d", test, return_code)
        report_item["success"] = False
        report_item["exception"] = ''.join(read_file(error_file_name)).replace('\'', '')

        if project.get_property("verbose") or project.get_property("integrationtest_always_verbose"):
            print_file_content(report_file_name)
            print_text_line()
            print_file_content(error_file_name)

    elif project.get_property("integrationtest_always_verbose"):
        print_file_content(report_file_name)
        print_text_line()
        print_file_content(error_file_name)

    return report_item


class ConsumingQueue(object):

    def __init__(self, ctx):
        self._items = []
        self._queue = ctx.Queue()
        self._ctx = ctx

    def consume_available_items(self):
        try:
            while True:
                item = self.get_nowait()
                self._items.append(item)
        except self._ctx.Empty:
            pass

    def put(self, *args, **kwargs):
        return self._queue.put(*args, **kwargs)

    def get_nowait(self, *args, **kwargs):
        return self._queue.get_nowait(*args, **kwargs)

    @property
    def items(self):
        return self._items

    @property
    def size(self):
        return len(self.items)


class TaskPoolProgress(object):
    """
    Class that renders progress for a set of tasks run in parallel.
    The progress is based on
    * the amount of total tasks, which must be static
    * the amount of workers running in parallel.
    The bar can be updated with the amount of tasks that have been successfully
    executed and render its progress.
    """

    BACKSPACE = "\b"
    FINISHED_SYMBOL = "-"
    PENDING_SYMBOL = "/"
    WAITING_SYMBOL = "|"
    PACMAN_FORWARD = "ᗧ"
    NO_PACMAN = ""

    def __init__(self, total_tasks_count, workers_count):
        self.total_tasks_count = total_tasks_count
        self.finished_tasks_count = 0
        self.workers_count = workers_count
        self.last_render_length = 0

    def update(self, finished_tasks_count):
        self.finished_tasks_count = finished_tasks_count

    def render(self):
        pacman = self.pacman_symbol
        finished_tests_progress = styled_text(
            self.FINISHED_SYMBOL * self.finished_tasks_count, fg(GREEN))
        running_tasks_count = self.running_tasks_count
        running_tests_progress = styled_text(
            self.PENDING_SYMBOL * running_tasks_count, fg(MAGENTA))
        waiting_tasks_count = self.waiting_tasks_count
        waiting_tasks_progress = styled_text(
            self.WAITING_SYMBOL * waiting_tasks_count, fg(GREY))
        trailing_space = ' ' if not pacman else ''

        return "[%s%s%s%s]%s" % (
            finished_tests_progress, pacman, running_tests_progress, waiting_tasks_progress, trailing_space)

    def render_to_terminal(self):
        if self.can_be_displayed:
            text_to_render = self.render()
            characters_to_be_erased = self.last_render_length
            self.last_render_length = len(text_to_render)
            text_to_render = "%s%s" % (characters_to_be_erased * self.BACKSPACE, text_to_render)
            print_text(text_to_render, flush=True)

    def mark_as_finished(self):
        if self.can_be_displayed:
            print_text_line()

    @property
    def pacman_symbol(self):
        if self.is_finished:
            return self.NO_PACMAN
        else:
            return self.PACMAN_FORWARD

    @property
    def running_tasks_count(self):
        pending_tasks = (self.total_tasks_count - self.finished_tasks_count)
        if pending_tasks > self.workers_count:
            return self.workers_count
        return pending_tasks

    @property
    def waiting_tasks_count(self):
        return self.total_tasks_count - self.finished_tasks_count - self.running_tasks_count

    @property
    def is_finished(self):
        return self.finished_tasks_count == self.total_tasks_count

    @property
    def can_be_displayed(self):
        if sys.stdout.isatty():
            return True
        return False

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 sys	48✔
21
22	from pybuilder.core import init, use_plugin, task, description, before	48✔
23	from pybuilder.plugins.python.test_plugin_helper import ReportsProcessor	48✔
24	from pybuilder.python_utils import mp_get_context	48✔
25	from pybuilder.terminal import print_text_line, print_file_content, print_text	48✔
26	from pybuilder.terminal import styled_text, fg, GREEN, MAGENTA, GREY	48✔
27	from pybuilder.utils import discover_files_matching, Timer, read_file	48✔
28
29	use_plugin("core")	48✔
30
31	from pybuilder.plugins.python.core_plugin import create_venv # noqa: E402	48✔
32
33
34	@init	48✔
35	def initialize_integrationtest_plugin(project):	48✔
36	project.set_property_if_unset("dir_source_integrationtest_python", "src/integrationtest/python")	48✔
37
38	project.set_property_if_unset("integrationtest_breaks_build", True)	48✔
39	project.set_property_if_unset("integrationtest_parallel", False)	48✔
40	project.set_property_if_unset("integrationtest_file_glob", "*_tests.py")	48✔
41	project.set_property_if_unset("integrationtest_additional_environment", {})	48✔
42	project.set_property_if_unset("integrationtest_additional_commandline", "")	48✔
43	project.set_property_if_unset("integrationtest_inherit_environment", False)	48✔
44	project.set_property_if_unset("integrationtest_always_verbose", False)	48✔
45	project.set_property_if_unset("integrationtest_cpu_scaling_factor", 4)	48✔
46	project.set_property_if_unset("integrationtest_python_env", "test")	48✔
47	project.set_property_if_unset("integrationtest_python_env_recreate", False)	48✔
48
49	project.set_property_if_unset("integrationtest_file_suffix", None) # deprecated, use integrationtest_file_glob.	48✔
50
51
52	@before("prepare")	48✔
53	def coverage_init(project, logger, reactor):	48✔
54	em = reactor.execution_manager	48✔
55
56	if em.is_task_in_current_execution_plan("coverage") and em.is_task_in_current_execution_plan(	48!
57	"run_integration_tests"):
58	project.get_property("_coverage_tasks").append(run_integration_tests)	×
59	project.get_property("_coverage_config_prefixes")[run_integration_tests] = "it"	×
60	project.set_property("it_coverage_name", "Python integration test")	×
61	project.set_property("it_coverage_python_env", project.get_property("integrationtest_python_env"))	×
62
63
64	@task	48✔
65	@description("Runs integration tests based on Python's unittest module")	48✔
66	def run_integration_tests(project, logger, reactor):	48✔
67	if project.get_property("integrationtest_parallel"):	48!
68	logger.warn("Parallel integration test execution is temporarily disabled")	×
69
70	# if not project.get_property("integrationtest_parallel"):
71	reports, total_time = run_integration_tests_sequentially(project, logger, reactor)	48✔
72	# else:
73	# reports, total_time = run_integration_tests_in_parallel(project, logger)
74
75	reports_processor = ReportsProcessor(project, logger)	48✔
76	reports_processor.process_reports(reports, total_time)	48✔
77	reports_processor.report_to_ci_server(project)	48✔
78	reports_processor.write_report_and_ensure_all_tests_passed()	48✔
79
80
81	def run_integration_tests_sequentially(project, logger, reactor):	48✔
82	logger.debug("Running integration tests sequentially")	48✔
83	reports_dir = prepare_reports_directory(project)	48✔
84
85	report_items = []	48✔
86
87	total_time = Timer.start()	48✔
88
89	for test in discover_integration_tests_for_project(project, logger):	48✔
90	report_item = run_single_test(logger, project, reactor, reports_dir, test)	48✔
91	report_items.append(report_item)	48✔
92
93	total_time.stop()	48✔
94
95	return report_items, total_time	48✔
96
97
98	def run_integration_tests_in_parallel(project, logger):	48✔
99	logger.info("Running integration tests in parallel")	×
100	ctx = mp_get_context("spawn")	×
101	tests = ctx.Queue()	×
102	reports = ConsumingQueue(ctx)	×
103	reports_dir = prepare_reports_directory(project)	×
104	cpu_scaling_factor = project.get_property("integrationtest_cpu_scaling_factor")	×
105	cpu_count = ctx.cpu_count()	×
106	worker_pool_size = cpu_count * cpu_scaling_factor	×
107	logger.debug(	×
108	"Running integration tests in parallel with {0} processes ({1} cpus found)".format(
109	worker_pool_size,
110	cpu_count))
111
112	total_time = Timer.start()	×
113	# fail OSX has no sem_getvalue() implementation so no queue size
114	total_tests_count = 0	×
115	for test in discover_integration_tests_for_project(project, logger):	×
116	tests.put(test)	×
117	total_tests_count += 1	×
118	progress = TaskPoolProgress(total_tests_count, worker_pool_size)	×
119
120	def pick_and_run_tests_then_report(tests, reports, reports_dir, logger, project):	×
121	while True:
122	try:	×
123	test = tests.get_nowait()	×
124	report_item = run_single_test(	×
125	logger, project, reports_dir, test, not progress.can_be_displayed)
126	reports.put(report_item)	×
127	except ctx.Empty:	×
128	break	×
129	except Exception as e:	×
130	logger.error("Failed to run test %r : %s" % (test, str(e)))	×
131	failed_report = {	×
132	"test": test,
133	"test_file": test,
134	"time": 0,
135	"success": False,
136	"exception": str(e)
137	}
138	reports.put(failed_report)	×
139	continue	×
140
141	pool = []	×
142	for i in range(worker_pool_size):	×
143	p = ctx.Process(target=pick_and_run_tests_then_report,	×
144	args=(tests, reports, reports_dir, logger, project))
145	pool.append(p)	×
146	p.start()	×
147
148	import time	×
149	while not progress.is_finished:	×
150	reports.consume_available_items()	×
151	finished_tests_count = reports.size	×
152	progress.update(finished_tests_count)	×
153	progress.render_to_terminal()	×
154	time.sleep(1)	×
155
156	progress.mark_as_finished()	×
157
158	total_time.stop()	×
159
160	return reports.items, total_time	×
161
162
163	def discover_integration_tests(source_path, suffix=".py"):	48✔
164	return discover_files_matching(source_path, "*{0}".format(suffix))	×
165
166
167	def discover_integration_tests_matching(source_path, file_glob):	48✔
168	return discover_files_matching(source_path, file_glob)	×
169
170
171	def discover_integration_tests_for_project(project, logger=None):	48✔
172	integrationtest_source_dir = project.expand_path("$dir_source_integrationtest_python")	48✔
173	integrationtest_suffix = project.get_property("integrationtest_file_suffix")	48✔
174	if integrationtest_suffix is not None:	48!
175	if logger is not None:	×
176	logger.warn(	×
177	"integrationtest_file_suffix is deprecated, please use integrationtest_file_glob"
178	)
179	project.set_property("integrationtest_file_glob", "*{0}".format(integrationtest_suffix))	×
180	integrationtest_glob = project.expand("$integrationtest_file_glob")	48✔
181	return discover_files_matching(integrationtest_source_dir, integrationtest_glob)	48✔
182
183
184	def add_additional_environment_keys(env, project):	48✔
185	additional_environment = project.get_property("integrationtest_additional_environment")	48✔
186
187	if not isinstance(additional_environment, dict):	48✔
188	raise ValueError("Additional environment %r is not a map." %	48✔
189	additional_environment)
190	for key in additional_environment:	48✔
191	env[key] = additional_environment[key]	48✔
192
193
194	def prepare_environment(project):	48✔
195	env = {	48✔
196	"PYTHONPATH": os.pathsep.join((project.expand_path("$dir_dist"),
197	project.expand_path("$dir_source_integrationtest_python")))
198	}
199
200	add_additional_environment_keys(env, project)	48✔
201
202	return env	48✔
203
204
205	def prepare_reports_directory(project):	48✔
206	reports_dir = project.expand_path("$dir_reports/integrationtests")	48✔
207	if not os.path.exists(reports_dir):	48!
208	os.mkdir(reports_dir)	48✔
209	return reports_dir	48✔
210
211
212	def run_single_test(logger, project, reactor, reports_dir, test, output_test_names=True):	48✔
213	additional_integrationtest_commandline_text = project.get_property("integrationtest_additional_commandline")	48✔
214
215	if additional_integrationtest_commandline_text:	48!
216	additional_integrationtest_commandline = tuple(additional_integrationtest_commandline_text.split(" "))	×
217	else:
218	additional_integrationtest_commandline = ()	48✔
219
220	name, _ = os.path.splitext(os.path.basename(test))	48✔
221
222	if output_test_names:	48!
223	logger.info("Running integration test %s", name)	48✔
224
225	venv_name = project.get_property("integrationtest_python_env")	48✔
226	python_env = reactor.python_env_registry[venv_name]	48✔
227	create_venv(project, logger, reactor, venv_name, True,	48✔
228	recreate_if_exists=project.get_property("integrationtest_python_env_recreate"))
229	env = prepare_environment(project)	48✔
230	command_and_arguments = python_env.executable + [test]	48✔
231	command_and_arguments += additional_integrationtest_commandline	48✔
232
233	report_file_name = os.path.join(reports_dir, name)	48✔
234	error_file_name = report_file_name + ".err"	48✔
235
236	test_time = Timer.start()	48✔
237	return_code = python_env.execute_command(command_and_arguments, report_file_name, env,	48✔
238	error_file_name=error_file_name,
239	inherit_env=project.get_property("integrationtest_inherit_environment"))
240	test_time.stop()	48✔
241	report_item = {	48✔
242	"test": name,
243	"test_file": test,
244	"time": test_time.get_millis(),
245	"success": True
246	}
247	if return_code != 0:	48✔
248	logger.error("Integration test failed: %s, exit code %d", test, return_code)	48✔
249	report_item["success"] = False	48✔
250	report_item["exception"] = ''.join(read_file(error_file_name)).replace('\'', '')	48✔
251
252	if project.get_property("verbose") or project.get_property("integrationtest_always_verbose"):	48✔
253	print_file_content(report_file_name)	48✔
254	print_text_line()	48✔
255	print_file_content(error_file_name)	48✔
256
257	elif project.get_property("integrationtest_always_verbose"):	48!
258	print_file_content(report_file_name)	×
259	print_text_line()	×
260	print_file_content(error_file_name)	×
261
262	return report_item	48✔
263
264
265	class ConsumingQueue(object):	48✔
266
267	def __init__(self, ctx):	48✔
268	self._items = []	48✔
269	self._queue = ctx.Queue()	48✔
270	self._ctx = ctx	48✔
271
272	def consume_available_items(self):	48✔
273	try:	48✔
274	while True:	40✔
275	item = self.get_nowait()	48✔
276	self._items.append(item)	48✔
277	except self._ctx.Empty:	48✔
278	pass	48✔
279
280	def put(self, args, *kwargs):	48✔
281	return self._queue.put(args, *kwargs)	×
282
283	def get_nowait(self, args, *kwargs):	48✔
284	return self._queue.get_nowait(args, *kwargs)	×
285
286	@property	48✔
287	def items(self):	48✔
288	return self._items	48✔
289
290	@property	48✔
291	def size(self):	48✔
292	return len(self.items)	48✔
293
294
295	class TaskPoolProgress(object):	48✔
296	"""
297	Class that renders progress for a set of tasks run in parallel.
298	The progress is based on
299	* the amount of total tasks, which must be static
300	* the amount of workers running in parallel.
301	The bar can be updated with the amount of tasks that have been successfully
302	executed and render its progress.
303	"""
304
305	BACKSPACE = "\b"	48✔
306	FINISHED_SYMBOL = "-"	48✔
307	PENDING_SYMBOL = "/"	48✔
308	WAITING_SYMBOL = "\|"	48✔
309	PACMAN_FORWARD = "ᗧ"	48✔
310	NO_PACMAN = ""	48✔
311
312	def __init__(self, total_tasks_count, workers_count):	48✔
313	self.total_tasks_count = total_tasks_count	48✔
314	self.finished_tasks_count = 0	48✔
315	self.workers_count = workers_count	48✔
316	self.last_render_length = 0	48✔
317
318	def update(self, finished_tasks_count):	48✔
319	self.finished_tasks_count = finished_tasks_count	48✔
320
321	def render(self):	48✔
322	pacman = self.pacman_symbol	48✔
323	finished_tests_progress = styled_text(	48✔
324	self.FINISHED_SYMBOL * self.finished_tasks_count, fg(GREEN))
325	running_tasks_count = self.running_tasks_count	48✔
326	running_tests_progress = styled_text(	48✔
327	self.PENDING_SYMBOL * running_tasks_count, fg(MAGENTA))
328	waiting_tasks_count = self.waiting_tasks_count	48✔
329	waiting_tasks_progress = styled_text(	48✔
330	self.WAITING_SYMBOL * waiting_tasks_count, fg(GREY))
331	trailing_space = ' ' if not pacman else ''	48✔
332
333	return "[%s%s%s%s]%s" % (	48✔
334	finished_tests_progress, pacman, running_tests_progress, waiting_tasks_progress, trailing_space)
335
336	def render_to_terminal(self):	48✔
337	if self.can_be_displayed:	48!
338	text_to_render = self.render()	48✔
339	characters_to_be_erased = self.last_render_length	48✔
340	self.last_render_length = len(text_to_render)	48✔
341	text_to_render = "%s%s" % (characters_to_be_erased * self.BACKSPACE, text_to_render)	48✔
342	print_text(text_to_render, flush=True)	48✔
343
344	def mark_as_finished(self):	48✔
345	if self.can_be_displayed:	×
346	print_text_line()	×
347
348	@property	48✔
349	def pacman_symbol(self):	48✔
350	if self.is_finished:	48✔
351	return self.NO_PACMAN	48✔
352	else:
353	return self.PACMAN_FORWARD	48✔
354
355	@property	48✔
356	def running_tasks_count(self):	48✔
357	pending_tasks = (self.total_tasks_count - self.finished_tasks_count)	48✔
358	if pending_tasks > self.workers_count:	48✔
359	return self.workers_count	48✔
360	return pending_tasks	48✔
361
362	@property	48✔
363	def waiting_tasks_count(self):	48✔
364	return self.total_tasks_count - self.finished_tasks_count - self.running_tasks_count	48✔
365
366	@property	48✔
367	def is_finished(self):	48✔
368	return self.finished_tasks_count == self.total_tasks_count	48✔
369
370	@property	48✔
371	def can_be_displayed(self):	48✔
372	if sys.stdout.isatty():	48✔
373	return True	48✔
374	return False	48✔

pybuilder / pybuilder / 19383496280

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