diff --git a/src/test/acceptance/test_upload_analyze_delete_firmware.py b/src/test/acceptance/test_upload_analyze_delete_firmware.py index 34f86f0b96..5c74133526 100644 --- a/src/test/acceptance/test_upload_analyze_delete_firmware.py +++ b/src/test/acceptance/test_upload_analyze_delete_firmware.py @@ -1,10 +1,10 @@ -import time from pathlib import Path import pytest from storage.fsorganizer import FSOrganizer from test.acceptance.conftest import test_fw_a, upload_test_firmware +from test.common_helper import wait_for_event def _upload_firmware_get(test_client, intercom): @@ -85,8 +85,7 @@ def _delete_firmware(test_client): assert b'Deleted 4 file(s) from database' in rv.data, 'deletion success page not shown' rv = test_client.get(f'/analysis/{test_fw_a.uid}') assert b'File not found in database' in rv.data, 'file is still available after delete' - time.sleep(3) - assert not local_firmware_path.exists(), 'file not deleted' + assert wait_for_event(lambda: not local_firmware_path.exists()), 'file not deleted' @pytest.mark.SchedulerTestConfig( diff --git a/src/test/common_helper.py b/src/test/common_helper.py index 22f2be87d5..2c6fbec598 100644 --- a/src/test/common_helper.py +++ b/src/test/common_helper.py @@ -1,12 +1,13 @@ from __future__ import annotations import os +import time from base64 import standard_b64encode from contextlib import contextmanager from copy import deepcopy from http import HTTPStatus from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Callable from helperFunctions.fileSystem import get_src_dir from helperFunctions.tag import TagColor @@ -331,3 +332,15 @@ def assert_search_result(response: TestResponse, included: list[FileObject], exc assert f"href='/analysis/{fo.uid}'" in html, f'file {fo.uid} should be included in the result' for fo in excluded: assert f"href='/analysis/{fo.uid}'" not in html, f'file {fo.uid} should not be included in the result' + + +def wait_for_event( + expression: Callable[[], bool], timeout: float = 5.0, check_interval: float = 0.1, inverted: bool = False +) -> bool: + check: Callable[[], bool] = expression if not inverted else lambda: not expression() + start_time = time.time() + while time.time() - start_time < timeout: + if check(): + return True + time.sleep(check_interval) + return check()