4651204881

Build # 4651204881

Build Type

push

github-actions

Committed by GitHub

Commit Message

chore: separate shaky tests (#3381)

Run Details

3 of 4 new or added lines in 2 files covered. (75.0%)

378 existing lines in 31 files now uncovered.

25407 of 29796 relevant lines covered (85.27%)

9.82 hits per line

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

83.33

/renku/ui/cli/utils/callback.py

#
# Copyright 2017-2023 - Swiss Data Science Center (SDSC)
# A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
# Eidgenössische Technische Hochschule Zürich (ETHZ).
#
# 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.
"""Communicator class for printing click output."""

import sys
from contextlib import contextmanager

import click
from tqdm import tqdm
from yaspin import yaspin

import renku.ui.cli.utils.color as color
from renku.core.util.communication import CommunicationCallback


class StandardOutput(CommunicationCallback):
    """Communication listener that outputs to stdout/stderr."""

    def __init__(self):
        super().__init__()
        self._progress_bars = {}
        self._progress_types = ["download"]
        self._spinner = None

    @contextmanager
    def pause_spinner(self):
        """Pause Spinner so output gets actually written."""
        if self._spinner:
            self._spinner.hide()

        yield

        if self._spinner:
            self._spinner.show()

    def echo(self, msg, end="\n"):
        """Write a message."""
        with CommunicationCallback.lock, self.pause_spinner():
            print(msg, end=end)

    def info(self, msg):
        """Write an info message."""
        with CommunicationCallback.lock, self.pause_spinner():
            print(msg)

    def warn(self, msg):
        """Write a warning message."""
        with CommunicationCallback.lock, self.pause_spinner():
            print(msg)

    def error(self, msg):
        """Write an error message."""
        with CommunicationCallback.lock, self.pause_spinner():
            print(msg, file=sys.stderr)

    def confirm(self, msg, abort=False, warning=False, default=False):
        """Get confirmation for an action."""
        return False

    def start_progress(self, name, total, **kwargs):
        """Start a new tqdm progressbar."""
        if name in self._progress_bars:
            raise ValueError(f"Name {name} is already a registered progressbar.")

        if "type" not in kwargs:
            kwargs["type"] = "download"

        if kwargs["type"] not in self._progress_types:
            self._progress_bars[name] = None
        elif kwargs["type"] == "download":
            self._progress_bars[name] = tqdm(
                total=total,
                unit="iB",
                unit_scale=True,
                desc=name,
                leave=False,
                bar_format="{desc:.32}: {percentage:3.0f}%|{bar}{r_bar}",
            )

    def update_progress(self, name, amount):
        """Update a progressbar."""
        if name not in self._progress_bars or not self._progress_bars[name]:
            return

        self._progress_bars[name].update(amount)

    def finalize_progress(self, name):
        """End a progressbar."""
        if name not in self._progress_bars or not self._progress_bars[name]:
            return

        self._progress_bars[name].close()
        del self._progress_bars[name]

    @contextmanager
    def busy(self, msg):
        """Indicate busy status using a spinner."""
        self._spinner = yaspin(text=msg)
        try:
            self._spinner.start()
            yield
        finally:
            self._spinner.stop()
            self._spinner = None


class ClickCallback(StandardOutput):
    """CommunicationCallback implementation for ``click`` messages."""

    INFO = click.style("Info: ", bold=True, fg=color.BLUE)
    WARNING = click.style("Warning: ", bold=True, fg=color.YELLOW)
    ERROR = click.style("Error: ", bold=True, fg=color.RED)

    def echo(self, msg, end="\n"):
        """Write a message."""
        new_line = True
        if end != "\n":
            msg = msg + end
            new_line = False
        with self.pause_spinner():
            click.echo(msg, nl=new_line)

    def info(self, msg):
        """Write an info message."""
        with self.pause_spinner():
            click.echo(self.INFO + msg)

    def warn(self, msg):
        """Write a warning message."""
        with self.pause_spinner():
            click.echo(self.WARNING + msg)

    def error(self, msg):
        """Write an error message."""
        with self.pause_spinner():
            click.echo(self.ERROR + msg, err=True)

    def has_prompt(self):
        """Return True if communicator provides a direct prompt to users."""
        return True

    def confirm(self, msg, abort=False, warning=False, default=False):
        """Get confirmation for an action using a prompt."""
        prefix = self.WARNING if warning else ""
        with self.pause_spinner():
            return click.confirm(prefix + msg, abort=abort, default=default)

    def prompt(self, msg, type=None, default=None, **kwargs):
        """Show a message prompt from the first callback that has a prompt."""
        with self.pause_spinner():
            return click.prompt(msg, type=type, default=default, **kwargs)

1	#
2	# Copyright 2017-2023 - Swiss Data Science Center (SDSC)
3	# A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
4	# Eidgenössische Technische Hochschule Zürich (ETHZ).
5	#
6	# Licensed under the Apache License, Version 2.0 (the "License");
7	# you may not use this file except in compliance with the License.
8	# You may obtain a copy of the License at
9	#
10	# http://www.apache.org/licenses/LICENSE-2.0
11	#
12	# Unless required by applicable law or agreed to in writing, software
13	# distributed under the License is distributed on an "AS IS" BASIS,
14	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15	# See the License for the specific language governing permissions and
16	# limitations under the License.
17	"""Communicator class for printing click output."""	22✔
18
19	import sys	22✔
20	from contextlib import contextmanager	22✔
21
22	import click	22✔
23	from tqdm import tqdm	22✔
24	from yaspin import yaspin	22✔
25
26	import renku.ui.cli.utils.color as color	22✔
27	from renku.core.util.communication import CommunicationCallback	22✔
28
29
30	class StandardOutput(CommunicationCallback):	22✔
31	"""Communication listener that outputs to stdout/stderr."""
32
33	def __init__(self):	22✔
34	super().__init__()	20✔
35	self._progress_bars = {}	20✔
36	self._progress_types = ["download"]	20✔
37	self._spinner = None	20✔
38
39	@contextmanager	22✔
40	def pause_spinner(self):	22✔
41	"""Pause Spinner so output gets actually written."""
42	if self._spinner:	16✔
43	self._spinner.hide()	4✔
44
45	yield	16✔
46
47	if self._spinner:	16✔
48	self._spinner.show()	4✔
49
50	def echo(self, msg, end="\n"):	22✔
51	"""Write a message."""
52	with CommunicationCallback.lock, self.pause_spinner():	×
53	print(msg, end=end)	×
54
55	def info(self, msg):	22✔
56	"""Write an info message."""
57	with CommunicationCallback.lock, self.pause_spinner():	×
58	print(msg)	×
59
60	def warn(self, msg):	22✔
61	"""Write a warning message."""
62	with CommunicationCallback.lock, self.pause_spinner():	×
63	print(msg)	×
64
65	def error(self, msg):	22✔
66	"""Write an error message."""
67	with CommunicationCallback.lock, self.pause_spinner():	×
68	print(msg, file=sys.stderr)	×
69
70	def confirm(self, msg, abort=False, warning=False, default=False):	22✔
71	"""Get confirmation for an action."""
72	return False	×
73
74	def start_progress(self, name, total, **kwargs):	22✔
75	"""Start a new tqdm progressbar."""
76	if name in self._progress_bars:	8✔
77	raise ValueError(f"Name {name} is already a registered progressbar.")	×
78
79	if "type" not in kwargs:	8✔
80	kwargs["type"] = "download"	8✔
81
82	if kwargs["type"] not in self._progress_types:	8✔
83	self._progress_bars[name] = None	×
84	elif kwargs["type"] == "download":	8✔
85	self._progress_bars[name] = tqdm(	8✔
86	total=total,
87	unit="iB",
88	unit_scale=True,
89	desc=name,
90	leave=False,
91	bar_format="{desc:.32}: {percentage:3.0f}%\|{bar}{r_bar}",
92	)
93
94	def update_progress(self, name, amount):	22✔
95	"""Update a progressbar."""
96	if name not in self._progress_bars or not self._progress_bars[name]:	8✔
UNCOV 97	return	×
98
99	self._progress_bars[name].update(amount)	8✔
100
101	def finalize_progress(self, name):	22✔
102	"""End a progressbar."""
103	if name not in self._progress_bars or not self._progress_bars[name]:	8✔
UNCOV 104	return	×
105
106	self._progress_bars[name].close()	8✔
107	del self._progress_bars[name]	8✔
108
109	@contextmanager	22✔
110	def busy(self, msg):	22✔
111	"""Indicate busy status using a spinner."""
112	self._spinner = yaspin(text=msg)	10✔
113	try:	10✔
114	self._spinner.start()	10✔
115	yield	10✔
116	finally:
117	self._spinner.stop()	10✔
118	self._spinner = None	10✔
119
120
121	class ClickCallback(StandardOutput):	22✔
122	"""CommunicationCallback implementation for ``click`` messages."""
123
124	INFO = click.style("Info: ", bold=True, fg=color.BLUE)	22✔
125	WARNING = click.style("Warning: ", bold=True, fg=color.YELLOW)	22✔
126	ERROR = click.style("Error: ", bold=True, fg=color.RED)	22✔
127
128	def echo(self, msg, end="\n"):	22✔
129	"""Write a message."""
130	new_line = True	14✔
131	if end != "\n":	14✔
132	msg = msg + end	8✔
133	new_line = False	8✔
134	with self.pause_spinner():	14✔
135	click.echo(msg, nl=new_line)	14✔
136
137	def info(self, msg):	22✔
138	"""Write an info message."""
139	with self.pause_spinner():	14✔
140	click.echo(self.INFO + msg)	14✔
141
142	def warn(self, msg):	22✔
143	"""Write a warning message."""
144	with self.pause_spinner():	12✔
145	click.echo(self.WARNING + msg)	12✔
146
147	def error(self, msg):	22✔
148	"""Write an error message."""
149	with self.pause_spinner():	×
150	click.echo(self.ERROR + msg, err=True)	×
151
152	def has_prompt(self):	22✔
153	"""Return True if communicator provides a direct prompt to users."""
154	return True	10✔
155
156	def confirm(self, msg, abort=False, warning=False, default=False):	22✔
157	"""Get confirmation for an action using a prompt."""
158	prefix = self.WARNING if warning else ""	6✔
159	with self.pause_spinner():	6✔
160	return click.confirm(prefix + msg, abort=abort, default=default)	6✔
161
162	def prompt(self, msg, type=None, default=None, **kwargs):	22✔
163	"""Show a message prompt from the first callback that has a prompt."""
164	with self.pause_spinner():	10✔
165	return click.prompt(msg, type=type, default=default, **kwargs)	10✔

SwissDataScienceCenter / renku-python / 4651204881

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