|
1 | 1 | # Copyright 2016-2018 Dirk Thomas |
2 | 2 | # Licensed under the Apache License, Version 2.0 |
3 | 3 |
|
| 4 | +import argparse |
4 | 5 | import asyncio |
5 | 6 | from collections import OrderedDict |
6 | 7 | import os |
|
9 | 10 | from threading import Thread |
10 | 11 | import time |
11 | 12 | from types import SimpleNamespace |
| 13 | +from unittest.mock import Mock |
| 14 | +from unittest.mock import patch |
12 | 15 |
|
13 | 16 | from colcon_core.executor import Job |
14 | 17 | from colcon_core.executor import OnError |
15 | 18 | from colcon_core.subprocess import SIGINT_RESULT |
| 19 | +from colcon_parallel_executor.executor.parallel import counting_number |
16 | 20 | from colcon_parallel_executor.executor.parallel \ |
17 | 21 | import ParallelExecutorExtension |
18 | 22 | import pytest |
|
22 | 26 |
|
23 | 27 | class Job1(Job): |
24 | 28 |
|
25 | | - def __init__(self): |
| 29 | + def __init__(self, identifier='job1'): |
26 | 30 | super().__init__( |
27 | | - identifier='job1', dependencies=set(), task=None, |
| 31 | + identifier=identifier, dependencies=set(), task=None, |
28 | 32 | task_context=None) |
29 | 33 |
|
30 | 34 | async def __call__(self, *args, **kwargs): |
@@ -213,3 +217,186 @@ def delayed_sigint(): |
213 | 217 |
|
214 | 218 | assert rc == signal.SIGINT |
215 | 219 | ran_jobs.clear() |
| 220 | + |
| 221 | + |
| 222 | +def test_counting_number(): |
| 223 | + assert counting_number('0') == 0 |
| 224 | + assert counting_number(1) == 1 |
| 225 | + with pytest.raises(ValueError): |
| 226 | + counting_number('-1') |
| 227 | + |
| 228 | + |
| 229 | +def test_add_arguments(): |
| 230 | + parser = argparse.ArgumentParser() |
| 231 | + extension = ParallelExecutorExtension() |
| 232 | + extension.add_arguments(parser=parser) |
| 233 | + args = parser.parse_args(['--parallel-workers', '5']) |
| 234 | + assert args.parallel_workers == 5 |
| 235 | + |
| 236 | + |
| 237 | +def test_parallel_run_until_complete_exception(): |
| 238 | + extension = ParallelExecutorExtension() |
| 239 | + args = SimpleNamespace(parallel_workers=2) |
| 240 | + jobs = OrderedDict() |
| 241 | + jobs['one'] = Job1() |
| 242 | + |
| 243 | + def mock_run(future): |
| 244 | + future.get_coro().close() |
| 245 | + raise RuntimeError('mock error') |
| 246 | + |
| 247 | + with patch( |
| 248 | + 'asyncio.base_events.BaseEventLoop.run_until_complete', |
| 249 | + side_effect=mock_run |
| 250 | + ): |
| 251 | + rc = extension.execute(args, jobs) |
| 252 | + assert rc == 1 |
| 253 | + |
| 254 | + |
| 255 | +def test_parallel_timeout(): |
| 256 | + extension = ParallelExecutorExtension() |
| 257 | + extension.put_event_into_queue = Mock() |
| 258 | + |
| 259 | + args = SimpleNamespace(parallel_workers=2) |
| 260 | + jobs = OrderedDict() |
| 261 | + jobs['eight'] = Job8() # sleep 3 |
| 262 | + |
| 263 | + original_wait = asyncio.wait |
| 264 | + |
| 265 | + def mock_wait(*args, **kwargs): |
| 266 | + kwargs['timeout'] = 0.01 |
| 267 | + return original_wait(*args, **kwargs) |
| 268 | + |
| 269 | + with patch( |
| 270 | + 'colcon_parallel_executor.executor.parallel.asyncio.wait', |
| 271 | + new=mock_wait |
| 272 | + ): |
| 273 | + rc = extension.execute(args, jobs) |
| 274 | + |
| 275 | + assert rc == 0 |
| 276 | + assert extension.put_event_into_queue.called |
| 277 | + ran_jobs.clear() |
| 278 | + |
| 279 | + |
| 280 | +class Job9(Job): |
| 281 | + |
| 282 | + def __init__(self): |
| 283 | + super().__init__( |
| 284 | + identifier='job9', dependencies=set(), task=None, |
| 285 | + task_context=None) |
| 286 | + |
| 287 | + async def __call__(self, *args, **kwargs): |
| 288 | + raise KeyboardInterrupt() |
| 289 | + |
| 290 | + |
| 291 | +def test_parallel_job_keyboard_interrupt(): |
| 292 | + extension = ParallelExecutorExtension() |
| 293 | + args = SimpleNamespace(parallel_workers=2) |
| 294 | + jobs = OrderedDict() |
| 295 | + jobs['nine'] = Job9() |
| 296 | + rc = extension.execute(args, jobs) |
| 297 | + assert rc == signal.SIGINT |
| 298 | + |
| 299 | + |
| 300 | +class Job10(Job): |
| 301 | + |
| 302 | + def __init__(self): |
| 303 | + super().__init__( |
| 304 | + identifier='job10', dependencies=set(), task=None, |
| 305 | + task_context=None) |
| 306 | + |
| 307 | + async def __call__(self, *args, **kwargs): |
| 308 | + return SIGINT_RESULT |
| 309 | + |
| 310 | + |
| 311 | +def test_parallel_job_sigint_result(): |
| 312 | + extension = ParallelExecutorExtension() |
| 313 | + args = SimpleNamespace(parallel_workers=2) |
| 314 | + jobs = OrderedDict() |
| 315 | + jobs['ten'] = Job10() |
| 316 | + rc = extension.execute(args, jobs) |
| 317 | + assert rc == signal.SIGINT |
| 318 | + |
| 319 | + |
| 320 | +def test_parallel_future_keyboard_interrupt(): |
| 321 | + extension = ParallelExecutorExtension() |
| 322 | + args = SimpleNamespace(parallel_workers=2) |
| 323 | + jobs = OrderedDict() |
| 324 | + jobs['eleven'] = Job1() |
| 325 | + |
| 326 | + original_wait = asyncio.wait |
| 327 | + |
| 328 | + async def mock_wait(*args, **kwargs): |
| 329 | + done, pending = await original_wait(*args, **kwargs) |
| 330 | + for f in done: |
| 331 | + f.exception = Mock(return_value=KeyboardInterrupt()) |
| 332 | + return done, pending |
| 333 | + |
| 334 | + with patch( |
| 335 | + 'colcon_parallel_executor.executor.parallel.asyncio.wait', |
| 336 | + new=mock_wait |
| 337 | + ): |
| 338 | + rc = extension.execute(args, jobs) |
| 339 | + |
| 340 | + assert rc == signal.SIGINT |
| 341 | + |
| 342 | + |
| 343 | +def test_parallel_future_cancelled(): |
| 344 | + extension = ParallelExecutorExtension() |
| 345 | + args = SimpleNamespace(parallel_workers=2) |
| 346 | + jobs = OrderedDict() |
| 347 | + jobs['twelve'] = Job1() |
| 348 | + |
| 349 | + original_wait = asyncio.wait |
| 350 | + |
| 351 | + async def mock_wait(*args, **kwargs): |
| 352 | + done, pending = await original_wait(*args, **kwargs) |
| 353 | + for f in done: |
| 354 | + f.cancelled = Mock(return_value=True) |
| 355 | + return done, pending |
| 356 | + |
| 357 | + with patch( |
| 358 | + 'colcon_parallel_executor.executor.parallel.asyncio.wait', |
| 359 | + new=mock_wait |
| 360 | + ): |
| 361 | + rc = extension.execute(args, jobs) |
| 362 | + |
| 363 | + assert rc == signal.SIGINT |
| 364 | + |
| 365 | + |
| 366 | +def test_priority(): |
| 367 | + extension = ParallelExecutorExtension() |
| 368 | + assert extension.PRIORITY > 100 |
| 369 | + |
| 370 | + |
| 371 | +class NonCoroutineJob(Job): |
| 372 | + |
| 373 | + def __init__(self): |
| 374 | + super().__init__( |
| 375 | + identifier='non_coro', dependencies=set(), task=None, |
| 376 | + task_context=None) |
| 377 | + |
| 378 | + def __call__(self, *args, **kwargs): |
| 379 | + pass |
| 380 | + |
| 381 | + |
| 382 | +def test_job_not_coroutine(): |
| 383 | + extension = ParallelExecutorExtension() |
| 384 | + args = SimpleNamespace(parallel_workers=2) |
| 385 | + jobs = OrderedDict() |
| 386 | + jobs['one'] = NonCoroutineJob() |
| 387 | + |
| 388 | + rc = extension.execute(args, jobs) |
| 389 | + assert rc == 1 |
| 390 | + |
| 391 | + |
| 392 | +def test_parallel_workers_zero(): |
| 393 | + extension = ParallelExecutorExtension() |
| 394 | + args = SimpleNamespace(parallel_workers=0) |
| 395 | + jobs = OrderedDict() |
| 396 | + jobs['one'] = Job1('job1') |
| 397 | + jobs['two'] = Job1('job2') |
| 398 | + |
| 399 | + rc = extension.execute(args, jobs) |
| 400 | + assert rc == 0 |
| 401 | + assert set(ran_jobs) == {'job1', 'job2'} |
| 402 | + ran_jobs.clear() |
0 commit comments