11392139613

Committed 17 Oct 2024 08:00PM UTC coverage: 59.856% (-0.05%) from 59.901%

Build # 11392139613

Build Type

push

github

Committed by

angelofenoglio

Commit Message

Move tasks load and list to run command

Run Details

206 of 504 branches covered (40.87%)

Branch coverage included in aggregate %.

6 of 11 new or added lines in 2 files covered. (54.55%)

33 existing lines in 4 files now uncovered.

2448 of 3930 relevant lines covered (62.29%)

0.62 hits per line

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

51.06

/leverage/modules/run.py

"""
    Tasks running module.
"""
import re

import click
from click.exceptions import Exit

from leverage import logger
from leverage.tasks import load_tasks
from leverage.tasks import list_tasks as _list_tasks
from leverage.logger import get_tasks_logger
from leverage._parsing import parse_task_args
from leverage._parsing import InvalidArgumentOrderError
from leverage._parsing import DuplicateKeywordArgumentError
from leverage._utils import clean_exception_traceback
from leverage._internals import pass_state


_TASK_PATTERN = re.compile(r"^(?P<name>[^\[\],\s]+)(\[(?P<arguments>[^\]]*)\])?$")

_logger = None


class MalformedTaskArgumentError(RuntimeError):
    pass


class TaskNotFoundError(RuntimeError):
    pass


@click.command()
@click.option(
    "--filename",
    "-f",
    default="build.py",
    show_default=True,
    help="Name of the build file containing the tasks definitions.",
)
@click.option("--list-tasks", "-l", is_flag=True, help="List available tasks to run.")
@click.argument("tasks", nargs=-1)
@pass_state
def run(state, filename, list_tasks, tasks):
    """Perform specified task(s) and all of its dependencies.

    When no task is given, the default (__DEFAULT__) task is run, if no default task has been defined, all available tasks are listed.
    """
    global _logger
    _logger = get_tasks_logger()

    # Load build file as a module
    state.module = load_tasks(build_script_filename=filename)

    if list_tasks:
        # --list-tasks|-l
        _list_tasks(state.module)
        return

    if tasks:
        # Run the given tasks
        try:
            tasks_to_run = _prepare_tasks_to_run(state.module, tasks)

        except (TaskNotFoundError, MalformedTaskArgumentError) as exc:
            logger.error(str(exc))
            raise Exit(1)

        _run_tasks(tasks=tasks_to_run)

    else:
        # Run the default task or list available tasks
        default_task = state.module.default_task
        if default_task is not None:
            prepared_default_task = [(default_task, [], {})]
            _run_tasks(prepared_default_task)

        else:
            list_tasks(state.module)


def _prepare_tasks_to_run(module, input_tasks):
    """Validate input tasks and arguments and pair them with the corresponding module's task.

    Args:
        module (dict): Dict containing the tasks from the build script.
        input_tasks (list): Strings containing the tasks to invoke and their arguments as received
            from user input.

    Raises:
        MalformedTaskArgumentError: When the string representing the invocation of a task does not conform
            to the required pattern.
        TaskNotFoundError: When the specified task is not found in the ones defined in the build script.

    Returns:
        list(tuple): List of tasks paired with their corresponding args and kwargs as provided by the user.
    """
    tasks = []
    for input_task in input_tasks:
        match = _TASK_PATTERN.match(input_task)
        if not match:
            raise MalformedTaskArgumentError(f"Malformed task argument in `{input_task}`.")

        name = match.group("name")
        arguments = match.group("arguments")

        try:
            args, kwargs = parse_task_args(arguments=arguments)

        except (InvalidArgumentOrderError, DuplicateKeywordArgumentError) as exc:
            logger.error(str(exc).format(task=name))
            raise Exit(1)

        task = [task for task in module.tasks if task.name == name]

        if not task:
            raise TaskNotFoundError(f"Unrecognized task `{name}`.")

        tasks.append((task[0], args, kwargs))

    return tasks


def _run_tasks(tasks):
    """Run the tasks provided.

    Args:
        tasks (list(tuple)): List of 3-tuples containing a task and it's positional and keyword arguments to run.
    """
    completed_tasks = set()
    for task, args, kwargs in tasks:
        # Remove current task from dependencies set to force it to run, as it was invoked by the user explicitly
        completed_tasks.discard(task)
        _run(task, completed_tasks, *args, **kwargs)


def _run(task, completed_tasks, *args, **kwargs):
    """Run the given task and all it's required dependencies, keeping track of all the already
    completed tasks as not to repeat them.

    Args:
        task (list): Tasks to run.
        completed_tasks (set): Tasks that have already ran.

    Returns:
        set: Updated set of already executed tasks.
    """
    # Satisfy dependencies recursively.
    for dependency in task.dependencies:
        _completed_tasks = _run(dependency, completed_tasks)
        completed_tasks.update(_completed_tasks)

    if task not in completed_tasks:
        if task.is_ignored:
            _logger.info(f"[bold yellow]⤳[/bold yellow] Ignoring task [bold italic]{task.name}[/bold italic]")

        else:
            _logger.info(f"[bold yellow]➜[/bold yellow] Starting task [bold italic]{task.name}[/bold italic]")

            try:
                task(*args, **kwargs)
            except Exception as exc:
                # Remove the two topmost frames of the traceback since they are internal leverage function calls,
                # only frames pertaining to the build script and its dependencies are shown.
                exc.__traceback__ = exc.__traceback__.tb_next.tb_next
                exc = clean_exception_traceback(exception=exc)

                _logger.exception(
                    f"[bold red]![/bold red] Error in task [bold italic]{task.name}[/bold italic]", exc_info=exc
                )
                _logger.critical("[red]✘[/red] [bold on red]Aborting build[/bold on red]")
                raise Exit(1)

            _logger.info(f"[green]✔[/green] Completed task [bold italic]{task.name}[/bold italic]")

        completed_tasks.add(task)

    return completed_tasks

1	"""
2	Tasks running module.
3	"""
4	import re	1✔
5
6	import click	1✔
7	from click.exceptions import Exit	1✔
8
9	from leverage import logger	1✔
10	from leverage.tasks import load_tasks	1✔
11	from leverage.tasks import list_tasks as _list_tasks	1✔
12	from leverage.logger import get_tasks_logger	1✔
13	from leverage._parsing import parse_task_args	1✔
14	from leverage._parsing import InvalidArgumentOrderError	1✔
15	from leverage._parsing import DuplicateKeywordArgumentError	1✔
16	from leverage._utils import clean_exception_traceback	1✔
17	from leverage._internals import pass_state	1✔
18
19
20	_TASK_PATTERN = re.compile(r"^(?P<name>[^\[\],\s]+)(\[(?P<arguments>[^\]]*)\])?$")	1✔
21
22	_logger = None	1✔
23
24
25	class MalformedTaskArgumentError(RuntimeError):	1✔
26	pass	1✔
27
28
29	class TaskNotFoundError(RuntimeError):	1✔
30	pass	1✔
31
32
33	@click.command()	1✔
34	@click.option(	1✔
35	"--filename",
36	"-f",
37	default="build.py",
38	show_default=True,
39	help="Name of the build file containing the tasks definitions.",
40	)
41	@click.option("--list-tasks", "-l", is_flag=True, help="List available tasks to run.")	1✔
42	@click.argument("tasks", nargs=-1)	1✔
43	@pass_state	1✔
44	def run(state, filename, list_tasks, tasks):	1✔
45	"""Perform specified task(s) and all of its dependencies.
46
47	When no task is given, the default (__DEFAULT__) task is run, if no default task has been defined, all available tasks are listed.
48	"""
49	global _logger
50	_logger = get_tasks_logger()	×
51
52	# Load build file as a module
NEW 53	state.module = load_tasks(build_script_filename=filename)	×
54
NEW 55	if list_tasks:	×
56	# --list-tasks\|-l
NEW 57	_list_tasks(state.module)	×
NEW 58	return	×
59
UNCOV 60	if tasks:	×
61	# Run the given tasks
62	try:	×
63	tasks_to_run = _prepare_tasks_to_run(state.module, tasks)	×
64
65	except (TaskNotFoundError, MalformedTaskArgumentError) as exc:	×
66	logger.error(str(exc))	×
67	raise Exit(1)	×
68
69	_run_tasks(tasks=tasks_to_run)	×
70
71	else:
72	# Run the default task or list available tasks
73	default_task = state.module.default_task	×
74	if default_task is not None:	×
75	prepared_default_task = [(default_task, [], {})]	×
76	_run_tasks(prepared_default_task)	×
77
78	else:
79	list_tasks(state.module)	×
80
81
82	def _prepare_tasks_to_run(module, input_tasks):	1✔
83	"""Validate input tasks and arguments and pair them with the corresponding module's task.
84
85	Args:
86	module (dict): Dict containing the tasks from the build script.
87	input_tasks (list): Strings containing the tasks to invoke and their arguments as received
88	from user input.
89
90	Raises:
91	MalformedTaskArgumentError: When the string representing the invocation of a task does not conform
92	to the required pattern.
93	TaskNotFoundError: When the specified task is not found in the ones defined in the build script.
94
95	Returns:
96	list(tuple): List of tasks paired with their corresponding args and kwargs as provided by the user.
97	"""
98	tasks = []	1✔
99	for input_task in input_tasks:	1✔
100	match = _TASK_PATTERN.match(input_task)	1✔
101	if not match:	1✔
102	raise MalformedTaskArgumentError(f"Malformed task argument in `{input_task}`.")	1✔
103
104	name = match.group("name")	1✔
105	arguments = match.group("arguments")	1✔
106
107	try:	1✔
108	args, kwargs = parse_task_args(arguments=arguments)	1✔
109
110	except (InvalidArgumentOrderError, DuplicateKeywordArgumentError) as exc:	1✔
111	logger.error(str(exc).format(task=name))	1✔
112	raise Exit(1)	1✔
113
114	task = [task for task in module.tasks if task.name == name]	1✔
115
116	if not task:	1✔
117	raise TaskNotFoundError(f"Unrecognized task `{name}`.")	1✔
118
119	tasks.append((task[0], args, kwargs))	1✔
120
121	return tasks	1✔
122
123
124	def _run_tasks(tasks):	1✔
125	"""Run the tasks provided.
126
127	Args:
128	tasks (list(tuple)): List of 3-tuples containing a task and it's positional and keyword arguments to run.
129	"""
130	completed_tasks = set()	×
131	for task, args, kwargs in tasks:	×
132	# Remove current task from dependencies set to force it to run, as it was invoked by the user explicitly
133	completed_tasks.discard(task)	×
134	_run(task, completed_tasks, args, *kwargs)	×
135
136
137	def _run(task, completed_tasks, args, *kwargs):	1✔
138	"""Run the given task and all it's required dependencies, keeping track of all the already
139	completed tasks as not to repeat them.
140
141	Args:
142	task (list): Tasks to run.
143	completed_tasks (set): Tasks that have already ran.
144
145	Returns:
146	set: Updated set of already executed tasks.
147	"""
148	# Satisfy dependencies recursively.
149	for dependency in task.dependencies:	×
150	_completed_tasks = _run(dependency, completed_tasks)	×
151	completed_tasks.update(_completed_tasks)	×
152
153	if task not in completed_tasks:	×
154	if task.is_ignored:	×
155	_logger.info(f"[bold yellow]⤳[/bold yellow] Ignoring task [bold italic]{task.name}[/bold italic]")	×
156
157	else:
158	_logger.info(f"[bold yellow]➜[/bold yellow] Starting task [bold italic]{task.name}[/bold italic]")	×
159
160	try:	×
161	task(args, *kwargs)	×
162	except Exception as exc:	×
163	# Remove the two topmost frames of the traceback since they are internal leverage function calls,
164	# only frames pertaining to the build script and its dependencies are shown.
165	exc.__traceback__ = exc.__traceback__.tb_next.tb_next	×
166	exc = clean_exception_traceback(exception=exc)	×
167
168	_logger.exception(	×
169	f"[bold red]![/bold red] Error in task [bold italic]{task.name}[/bold italic]", exc_info=exc
170	)
171	_logger.critical("[red]✘[/red] [bold on red]Aborting build[/bold on red]")	×
172	raise Exit(1)	×
173
174	_logger.info(f"[green]✔[/green] Completed task [bold italic]{task.name}[/bold italic]")	×
175
176	completed_tasks.add(task)	×
177
178	return completed_tasks	×

binbashar / leverage / 11392139613

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