14471580514

Committed 11 Jul 2024 02:47PM UTC coverage: 57.068% (-20.2%) from 77.316%

Build # 14471580514

Build Type

push

github

Committed by

web-flow

Commit Message

fix: properly remove activities in workflow remove if catalog is corrupt (#3729)

Run Details

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

6141 existing lines in 213 files now uncovered.

17395 of 30481 relevant lines covered (57.07%)

1.77 hits per line

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

71.11

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

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

SwissDataScienceCenter / renku-python / 14471580514

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