diff --git a/pytest_anki/_session.py b/pytest_anki/_session.py index 752382f..af878b5 100644 --- a/pytest_anki/_session.py +++ b/pytest_anki/_session.py @@ -289,13 +289,14 @@ def run_in_thread_and_wait( task: Callable, task_args: Optional[Tuple[Any, ...]] = None, task_kwargs: Optional[Dict[str, Any]] = None, + timeout: int = 5000, ): thread_pool = QThreadPool.globalInstance() worker = SignallingWorker( task=task, task_args=task_args, task_kwargs=task_kwargs ) - with self._qtbot.wait_signal(worker.signals.finished): + with self._qtbot.wait_signal(worker.signals.finished, timeout=timeout): thread_pool.start(worker) if exception := worker.error: @@ -341,12 +342,14 @@ def run_with_chrome_driver( self, test_function: Callable[[webdriver.Chrome], Optional[bool]], target_web_view: Optional[Union[AnkiWebViewType, str]] = None, + timeout: int = 5000, ): """[summary] Args: test_function (Callable[[webdriver.Chrome], Optional[bool]]): [description] target_web_view: Web view as identified by its title. Defaults to None. + timeout: Time to wait for task to complete until qtbot raises a TimeoutError """ if self._web_debugging_port is None: raise AnkiSessionError("Web debugging interface is not active") @@ -374,7 +377,7 @@ def test_wrapper() -> Optional[bool]: return test_function(self._chrome_driver) with self._allow_selenium_to_detect_anki(): - return self.run_in_thread_and_wait(test_wrapper) + return self.run_in_thread_and_wait(test_wrapper, timeout=timeout) def reset_chrome_driver(self): if not self._chrome_driver: diff --git a/tests/test_web_debugging.py b/tests/test_web_debugging.py index 241605d..8bd2398 100644 --- a/tests/test_web_debugging.py +++ b/tests/test_web_debugging.py @@ -28,9 +28,12 @@ # # Any modifications to this file must keep this entire header intact. +import time from unittest.mock import Mock +import pytest import requests +from pytestqt.qtbot import TimeoutError from selenium import webdriver from pytest_anki import AnkiSession, AnkiWebViewType @@ -46,6 +49,31 @@ def test_run_in_thread(anki_session: AnkiSession): mock_task.assert_called_once_with(*args, **kwargs) +def test_can_supply_timeout(anki_session: AnkiSession): + timeout_duration = 4 + task_duration = 3 + + def mock_task(): + time.sleep(task_duration) + + start_time = time.time() + try: + anki_session.run_in_thread_and_wait( + task=mock_task, timeout=timeout_duration * 1000 + ) + except TimeoutError: + pytest.fail("Call unexpectedly timed out") + + wait_time = time.time() - start_time + + assert timeout_duration > wait_time >= task_duration + + with pytest.raises(TimeoutError): + anki_session.run_in_thread_and_wait( + task=mock_task, timeout=(task_duration - 1) * 1000 + ) + + def test_web_debugging_available_on_launch(anki_session: AnkiSession): port = anki_session.web_debugging_port