(make_timeline, daemon, order)
| 559 | @pytest.mark.timeout(3) |
| 560 | @pytest.mark.parametrize("order", ["mark_then_wait", "wait_then_mark"]) |
| 561 | def test_concurrency(make_timeline, daemon, order): |
| 562 | timeline, initial_history = make_timeline() |
| 563 | |
| 564 | occurrences = [] |
| 565 | worker_can_proceed = threading.Event() |
| 566 | |
| 567 | @daemon |
| 568 | def worker(): |
| 569 | worker_can_proceed.wait() |
| 570 | mark = timeline.mark("tada") |
| 571 | occurrences.append(mark) |
| 572 | |
| 573 | if order == "mark_then_wait": |
| 574 | worker_can_proceed.set() |
| 575 | unblock_worker_later = None |
| 576 | else: |
| 577 | |
| 578 | @daemon |
| 579 | def unblock_worker_later(): |
| 580 | time.sleep(0.1) |
| 581 | worker_can_proceed.set() |
| 582 | |
| 583 | mark = timeline.wait_for_next(Mark("tada"), freeze=True) |
| 584 | |
| 585 | worker.join() |
| 586 | assert mark is occurrences[0] |
| 587 | assert timeline.last is mark |
| 588 | assert timeline.history() == initial_history + occurrences |
| 589 | |
| 590 | if unblock_worker_later: |
| 591 | unblock_worker_later.join() |
nothing calls this directly
no test coverage detected
searching dependent graphs…