1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
|
# Copyright 2016-2018 Dirk Thomas
# Licensed under the Apache License, Version 2.0
from argparse import ArgumentParser
from asyncio import CancelledError
from unittest.mock import Mock
from unittest.mock import patch
from colcon_core.event.job import JobEnded
from colcon_core.event.job import JobQueued
from colcon_core.event.job import JobSkipped
from colcon_core.event.job import JobStarted
from colcon_core.event.output import StderrLine
from colcon_core.executor import add_executor_arguments
from colcon_core.executor import DEFAULT_EXECUTOR_ENVIRONMENT_VARIABLE
from colcon_core.executor import execute_jobs
from colcon_core.executor import ExecutorExtensionPoint
from colcon_core.executor import get_executor_extensions
from colcon_core.executor import Job
from colcon_core.executor import OnError
from colcon_core.subprocess import SIGINT_RESULT
import pytest
from .environment_context import EnvironmentContext
from .extension_point_context import ExtensionPointContext
from .run_until_complete import run_until_complete
class Task:
def __init__(self):
self.return_value = None
def set_context(self, *, context):
pass
async def __call__(self, *args, **kwargs):
if isinstance(self.return_value, BaseException):
raise self.return_value
return self.return_value
def test_job():
task = Task()
task_context = Mock()
task_context.dependencies = Mock()
task_context.pkg = Mock()
task_context.pkg.name = 'name'
job = Job(
identifier='id', dependencies=set(), task=task,
task_context=task_context)
assert str(job) == 'id'
events = []
event_queue = Mock()
event_queue.put = lambda event: events.append(event)
job.set_event_queue(event_queue)
assert len(events) == 1
assert isinstance(events[-1][0], JobQueued)
assert events[-1][0].identifier == 'name'
assert events[-1][0].dependencies == task_context.dependencies
assert events[-1][1] == job
# successful task
rc = run_until_complete(job())
assert rc == 0
assert len(events) == 3
assert isinstance(events[-2][0], JobStarted)
assert events[-2][0].identifier == 'name'
assert events[-2][1] == job
assert isinstance(events[-1][0], JobEnded)
assert events[-1][0].identifier == 'name'
assert events[-1][0].rc == 0
assert events[-1][1] == job
# canceled task
job.returncode = None
task.return_value = CancelledError()
rc = run_until_complete(job())
assert rc is SIGINT_RESULT
assert len(events) == 5
assert isinstance(events[-2][0], JobStarted)
assert events[-2][0].identifier == 'name'
assert events[-2][1] == job
assert isinstance(events[-1][0], JobEnded)
assert events[-1][0].identifier == 'name'
assert events[-1][0].rc is SIGINT_RESULT
assert events[-1][1] == job
# task raising exception
job.returncode = None
task.return_value = RuntimeError('custom exception')
with pytest.raises(RuntimeError):
run_until_complete(job())
assert len(events) == 8
assert isinstance(events[-3][0], JobStarted)
assert events[-3][0].identifier == 'name'
assert events[-3][1] == job
assert isinstance(events[-2][0], StderrLine)
assert events[-2][0].line.endswith(b'\nRuntimeError: custom exception\n')
assert events[-2][1] == job
assert isinstance(events[-1][0], JobEnded)
assert events[-1][0].identifier == 'name'
assert events[-1][0].rc == 1
assert events[-1][1] == job
# override task return code
job.returncode = 2
task.return_value = 0
rc = run_until_complete(job())
assert rc == 2
assert len(events) == 10
assert isinstance(events[-2][0], JobStarted)
assert events[-2][0].identifier == 'name'
assert events[-2][1] == job
assert isinstance(events[-1][0], JobEnded)
assert events[-1][0].identifier == 'name'
assert events[-1][0].rc == 2
assert events[-1][1] == job
def test_interface():
interface = ExecutorExtensionPoint()
interface._flush()
event_controller = Mock()
event_controller.get_queue.returns = Mock()
interface.set_event_controller(event_controller)
interface.put_event_into_queue(object())
interface._flush()
assert event_controller.flush.call_count == 1
assert event_controller.get_queue().put.call_count == 1
class Extension1(ExecutorExtensionPoint):
"""Class documentation."""
class Extension2(ExecutorExtensionPoint):
PRIORITY = 110
class Extension3(ExecutorExtensionPoint):
pass
def test_add_executor_arguments():
parser = ArgumentParser()
# extensions with the same priority
with ExtensionPointContext(
extension1=Extension1, extension2=Extension2, extension3=Extension3
):
with pytest.raises(AssertionError) as e:
add_executor_arguments(parser)
assert 'Executor extensions must have unique priorities' in \
str(e.value)
# no extensions
with ExtensionPointContext():
with pytest.raises(AssertionError) as e:
add_executor_arguments(parser)
assert 'No executor extensions found' in str(e.value)
# choose executor by environment variable
with ExtensionPointContext(extension1=Extension1, extension2=Extension2):
extensions = get_executor_extensions()
extensions[110]['extension2'].add_arguments = Mock(
side_effect=RuntimeError('custom exception'))
extensions[100]['extension1'].add_arguments = Mock(return_value=None)
env = {DEFAULT_EXECUTOR_ENVIRONMENT_VARIABLE.name: 'extension1'}
with EnvironmentContext(**env):
with patch('colcon_core.executor.logger.error') as error:
add_executor_arguments(parser)
assert extensions[100]['extension1'].add_arguments.call_count == 1
# the raised exception is catched and results in an error message
assert error.call_count == 1
assert len(error.call_args[0]) == 1
assert error.call_args[0][0].startswith(
"Exception in executor extension 'extension2': custom exception\n")
args = parser.parse_args([])
assert args.executor == 'extension1'
# choose default executor
parser = ArgumentParser()
with ExtensionPointContext(extension1=Extension1, extension2=Extension2):
add_executor_arguments(parser)
args = parser.parse_args([])
assert args.executor == 'extension2'
def test_execute_jobs():
context = Mock()
context.args = Mock()
context.args.event_handlers = None
task_context = Mock()
task_context.pkg = Mock()
task_context.pkg.name = 'name'
jobs = {
'one': Job(
identifier='id', dependencies=set(), task=None,
task_context=task_context)}
event_reactor = Mock()
event_reactor.__enter__ = lambda self: self
event_reactor.__exit__ = lambda self, *args: None
with patch(
'colcon_core.executor.create_event_reactor', return_value=event_reactor
):
with ExtensionPointContext(
extension1=Extension1, extension2=Extension2
):
# no extension selected
with pytest.raises(AssertionError):
execute_jobs(context, jobs)
# execute method not implemented and sending skipped job event
context.args.executor = 'extension2'
with patch('colcon_core.executor.logger.error') as error:
rc = execute_jobs(context, jobs)
assert rc == 1
assert error.call_count == 1
assert len(error.call_args[0]) == 1
assert error.call_args[0][0].startswith(
"Exception in executor extension 'extension2': \n")
assert event_reactor.get_queue().put.call_count == 2
assert isinstance(
event_reactor.get_queue().put.call_args_list[0][0][0][0],
JobQueued)
assert isinstance(
event_reactor.get_queue().put.call_args_list[1][0][0][0],
JobSkipped)
# successful execution
event_reactor.get_queue().put.reset_mock()
jobs['one'].returncode = 0
extensions = get_executor_extensions()
extensions[110]['extension2'].execute = \
lambda args, jobs, on_error: 0
callback = Mock()
rc = execute_jobs(
context, jobs, on_error=OnError.interrupt,
pre_execution_callback=callback)
assert rc == 0
assert event_reactor.get_queue().put.call_count == 1
assert isinstance(
event_reactor.get_queue().put.call_args[0][0][0], JobQueued)
assert callback.call_count == 1
|
