Wait for the current screen and its children to have processed all pending events. Args: timeout: A timeout in seconds to wait. Returns: `True` if all events were processed. `False` if an exception occurred, meaning that not all events could be p
(self, timeout: float = 30.0)
| 471 | return widget is None or widget_at is target_widget |
| 472 | |
| 473 | async def _wait_for_screen(self, timeout: float = 30.0) -> bool: |
| 474 | """Wait for the current screen and its children to have processed all pending events. |
| 475 | |
| 476 | Args: |
| 477 | timeout: A timeout in seconds to wait. |
| 478 | |
| 479 | Returns: |
| 480 | `True` if all events were processed. `False` if an exception occurred, |
| 481 | meaning that not all events could be processed. |
| 482 | |
| 483 | Raises: |
| 484 | WaitForScreenTimeout: If the screen and its children didn't finish processing within the timeout. |
| 485 | """ |
| 486 | try: |
| 487 | screen = self.app.screen |
| 488 | except Exception: |
| 489 | return False |
| 490 | children = [self.app, *screen.walk_children(with_self=True)] |
| 491 | count = 0 |
| 492 | count_zero_event = asyncio.Event() |
| 493 | |
| 494 | def decrement_counter() -> None: |
| 495 | """Decrement internal counter, and set an event if it reaches zero.""" |
| 496 | nonlocal count |
| 497 | count -= 1 |
| 498 | if count == 0: |
| 499 | # When count is zero, all messages queued at the start of the method have been processed |
| 500 | count_zero_event.set() |
| 501 | |
| 502 | # Increase the count for every successful call_later |
| 503 | for child in children: |
| 504 | if child.call_later(decrement_counter): |
| 505 | count += 1 |
| 506 | |
| 507 | if count: |
| 508 | # Wait for the count to return to zero, or a timeout, or an exception |
| 509 | wait_for = [ |
| 510 | asyncio.create_task(count_zero_event.wait()), |
| 511 | asyncio.create_task(self.app._exception_event.wait()), |
| 512 | ] |
| 513 | _, pending = await asyncio.wait( |
| 514 | wait_for, |
| 515 | timeout=timeout, |
| 516 | return_when=asyncio.FIRST_COMPLETED, |
| 517 | ) |
| 518 | |
| 519 | for task in pending: |
| 520 | task.cancel() |
| 521 | |
| 522 | timed_out = len(wait_for) == len(pending) |
| 523 | if timed_out: |
| 524 | raise WaitForScreenTimeout( |
| 525 | "Timed out while waiting for widgets to process pending messages." |
| 526 | ) |
| 527 | |
| 528 | # We've either timed out, encountered an exception, or we've finished |
| 529 | # decrementing all the counters (all events processed in children). |
| 530 | if count > 0: |
no test coverage detected