Test speedup

src/config.py

@@ -153,6 +153,7 @@ class Plugin(BaseModel):
     ssdeep_ignore: int
 
     intercom_poll_delay: float
+    analysis_status_update_interval: float = 4.5
 
     throw_exceptions: bool
 

src/config/fact-core-config.toml

@@ -61,6 +61,7 @@ firmware-file-storage-directory = "/media/data/fact_fw_data"
 block-delay = 0.1
 ssdeep-ignore = 1
 intercom-poll-delay = 1.0
+analysis-status-update-interval = 4.5
 scheduling-worker-count = 4
 collector-worker-count = 2
 throw-exceptions = false

src/conftest.py

@@ -105,7 +105,8 @@ def backend_config(request, common_config, firmware_file_storage_directory) -> c
         'firmware_file_storage_directory': firmware_file_storage_directory,
         'block_delay': 0.1,
         'ssdeep_ignore': 1,
-        'intercom_poll_delay': 1.0,
+        'intercom_poll_delay': 0.1,
+        'analysis_status_update_interval': 0.2,
         'throw_exceptions': True,  # Always throw exceptions to avoid miraculous timeouts in test cases
         'plugin_defaults': {'processes': 1},
         'unpacking': {

src/scheduler/analysis_status.py

@@ -10,14 +10,14 @@
 from time import time
 from typing import TYPE_CHECKING, Dict, Set
 
+import config
 from helperFunctions.process import stop_process
 from objects.firmware import Firmware
 from storage.redis_status_interface import RedisStatusInterface
 
 if TYPE_CHECKING:
     from objects.file import FileObject
 
-UPDATE_INTERVAL = 4.5  # a bit less than the update interval on the system health page, FixMe: -> configuration
 RECENTLY_FINISHED_DISPLAY_TIME_IN_SEC = 300
 
 
@@ -116,11 +116,11 @@ def _worker_loop(self):
                 logging.debug(f'updating status (queue: {self.queue.qsize()})')
                 self._clear_recently_finished()
                 self._store_status()
-                next_update_time = current_time + UPDATE_INTERVAL
+                next_update_time = current_time + config.backend.analysis_status_update_interval
         logging.debug('stopped analysis status worker')
 
     def _update_status(self):
-        update_type, *args = self.queue.get(timeout=UPDATE_INTERVAL)
+        update_type, *args = self.queue.get(timeout=config.backend.analysis_status_update_interval)
         if update_type == _UpdateType.add_update:
             self._add_update(*args)
         elif update_type == _UpdateType.add_firmware:

src/test/acceptance/test_misc.py

@@ -1,5 +1,4 @@
 import json
-import time
 from urllib.parse import quote
 
 import pytest
@@ -110,12 +109,11 @@ def test_misc(
         for fw in [test_fw_a, test_fw_c]:
             self._upload_firmware_put(test_client, fw.path, fw.name, fw.uid)
         self._show_about(test_client)
-        time.sleep(4)
+        assert analysis_finished_event.wait(timeout=10)
         workload_statistic.update(
             unpacking_workload=unpacking_scheduler.get_scheduled_workload(),
             analysis_workload=analysis_scheduler.get_scheduled_workload(),
         )
-        assert analysis_finished_event.wait(timeout=10)
         self._show_system_monitor(test_client)
 
         stats_updater.update_all_stats()

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(

src/test/common_helper.py

@@ -1,11 +1,12 @@
 from __future__ import annotations
 
+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
@@ -326,3 +327,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()