fix(PLU-326): stop treating failed analyses as complete in status poller

reai_toolkit/utils/monitoring/process_binary_monitor.py

@@ -4,80 +4,99 @@
 import revengai
 from binaryninja import log_info, log_error, BinaryView
 from requests.exceptions import RequestException
-from PySide6.QtCore import QObject, Signal
+from PySide6.QtCore import QObject
 from reai_toolkit.utils.core.sync import AnalysisSyncService
 
+_IN_PROGRESS_STATUSES = (
+ revengai.StatusInput.UPLOADED,
+ revengai.StatusInput.QUEUED,
+ revengai.StatusInput.PROCESSING,
+)
+
+
 class PeriodicChecker(QObject):
- update_text_signal = Signal(object, str)
  sync_service: AnalysisSyncService
-
+
  def __init__(self, config):
  super().__init__()
  self._current_timer: Optional[Timer] = None
  self.number_of_clicks = 0
- self.update_text_signal.connect(self._update_text_slot)
  self.config = config
  self.sync_service = AnalysisSyncService(config)
 
- def _update_text_slot(self, callback, text):
- """Slot that runs in the main thread to safely update UI"""
- try:
- if hasattr(callback, '__call__'):
- if hasattr(self, '_current_editor'):
- callback(self._current_editor, text)
- except Exception as ex:
- log_error(f"RevEng.AI | Error updating UI: {str(ex)}")
-
  def stop(self):
  if self._current_timer:
  self._current_timer.cancel()
  self._current_timer = None
  log_info("RevEng.AI | Stopped periodic status check")
-
 
- def start_checking(self, binary_view: BinaryView, analysis_id: int, binary_id: int, callback, interval: float = 60) -> None:
+ def start_checking(
+ self,
+ binary_view: BinaryView,
+ analysis_id: int,
+ binary_id: int,
+ callback,
+ interval: float = 60,
+ ) -> None:
  def _worker(bv: BinaryView, bid: int, aid: int):
  try:
  with self.config.create_api_client() as api_client:
  api_instance = revengai.AnalysesCoreApi(api_client)
- api_response = api_instance.get_analysis_status(aid)
+ api_response = api_instance.get_analysis_status(aid)
  status = api_response.data.analysis_status
- log_info(f"RevEng.AI | Current status for analysis [Binary ID: {bid}] [Analysis ID: {aid}]: {status}")
+ log_info(
+ f"RevEng.AI | Current status for analysis [Binary ID: {bid}] [Analysis ID: {aid}]: {status}"
+ )
 
- if status in ("Queued", "Processing"):
+ if status in _IN_PROGRESS_STATUSES:
  if bv and bv.file and bv.file.filename:
  self._current_timer = Timer(
- interval,
- _worker,
- args=(bv, bid, aid)
+ interval, _worker, args=(bv, bid, aid)
  )
  self._current_timer.start()
  log_info(
  f"RevEng.AI | Scheduled next status check for: {basename(bv.file.filename)} [Binary ID: {bid}] [Analysis ID: {aid}]"
  )
- else:
-
+ elif status == revengai.StatusInput.COMPLETE:
  # Analysis is complete, fetch model_id and invoke callback
  with self.config.create_api_client() as api_client:
  api_instance = revengai.AnalysesCoreApi(api_client)
- analysis_details: revengai.BaseResponseBasic = api_instance.get_analysis_basic_info(
- analysis_id=analysis_id
+ analysis_details: revengai.BaseResponseBasic = (
+ api_instance.get_analysis_basic_info(
+ analysis_id=analysis_id
+ )
  )
  model_id = analysis_details.data.model_id
  callback(bid, aid, model_id)
 
- bv = self.sync_service.sync_analysis_data(analysis_id=aid, bv=bv)
+ bv = self.sync_service.sync_analysis_data(
+ analysis_id=aid, bv=bv
+ )
 
- log_info(f"RevEng.AI | Analysis completed with status: {status} for Binary ID: {bid} | Analysis ID: {aid} | Model ID: {model_id}")
+ log_info(
+ f"RevEng.AI | Analysis completed with status: {status} for Binary ID: {bid} | Analysis ID: {aid} | Model ID: {model_id}"
+ )
+ else:
+ log_error(
+ f"RevEng.AI | Analysis failed with status '{status}' "
+ f"[Binary ID: {bid}] [Analysis ID: {aid}]. "
+ "Check the analysis log in the RevEng.AI portal, then re-run the analysis."
+ )
  except RequestException as ex:
- log_error(f"RevEng.AI | Error getting binary analysis status: {str(ex)}")
+ log_error(
+ f"RevEng.AI | Network error while monitoring analysis [Binary ID: {bid}] [Analysis ID: {aid}]: {ex}"
+ )
  except Exception as ex:
- log_error(f"RevEng.AI | Unexpected error during status check: {str(ex)}")
+ log_error(
+ f"RevEng.AI | Unexpected error while monitoring analysis [Binary ID: {bid}] [Analysis ID: {aid}]: {ex}"
+ )
 
  self.stop()
 
- self._current_timer = Timer(30, _worker, args=(binary_view, binary_id, analysis_id))
+ self._current_timer = Timer(
+ 30, _worker, args=(binary_view, binary_id, analysis_id)
+ )
  self._current_timer.start()
  log_info(
  f"RevEng.AI | Started periodic status check for: {basename(binary_view.file.filename)} [Binary ID: {binary_id}] [Analysis ID: {analysis_id}]"
- )
+ )

tests/unit/monitoring/test_process_binary_monitor.py

@@ -57,3 +57,74 @@ def fake_timer(interval, function, args=()):
 
  assert len(timers) == before + 1
  checker.sync_service.sync_analysis_data.assert_not_called()
+
+
+@pytest.mark.parametrize("status", ["Uploaded", "Queued", "Processing"])
+def test_worker_reschedules_for_in_progress_statuses(mocker, status):
+ timers = []
+
+ def fake_timer(interval, function, args=()):
+ timers.append((interval, function, args))
+ return MagicMock()
+
+ mocker.patch.object(pbm_mod, "Timer", side_effect=fake_timer)
+ api = mocker.patch.object(pbm_mod.revengai, "AnalysesCoreApi").return_value
+ api.get_analysis_status.return_value.data.analysis_status = status
+
+ checker = pbm_mod.PeriodicChecker(MagicMock())
+ checker.sync_service = MagicMock()
+
+ checker.start_checking(_bv(), analysis_id=2, binary_id=1, callback=MagicMock())
+ before = len(timers)
+ timers[-1][1](*timers[-1][2])
+
+ assert len(timers) == before + 1
+ checker.sync_service.sync_analysis_data.assert_not_called()
+
+
+def test_worker_treats_error_status_as_terminal_failure(mocker):
+ timers = []
+
+ def fake_timer(interval, function, args=()):
+ timers.append((interval, function, args))
+ return MagicMock()
+
+ mocker.patch.object(pbm_mod, "Timer", side_effect=fake_timer)
+ api = mocker.patch.object(pbm_mod.revengai, "AnalysesCoreApi").return_value
+ api.get_analysis_status.return_value.data.analysis_status = "Error"
+
+ checker = pbm_mod.PeriodicChecker(MagicMock())
+ checker.sync_service = MagicMock()
+ callback = MagicMock()
+
+ checker.start_checking(_bv(), analysis_id=2, binary_id=1, callback=callback)
+ before = len(timers)
+ timers[-1][1](*timers[-1][2])
+
+ assert len(timers) == before
+ callback.assert_not_called()
+ checker.sync_service.sync_analysis_data.assert_not_called()
+
+
+def test_worker_treats_unrecognised_status_as_terminal_failure(mocker):
+ timers = []
+
+ def fake_timer(interval, function, args=()):
+ timers.append((interval, function, args))
+ return MagicMock()
+
+ mocker.patch.object(pbm_mod, "Timer", side_effect=fake_timer)
+ api = mocker.patch.object(pbm_mod.revengai, "AnalysesCoreApi").return_value
+ api.get_analysis_status.return_value.data.analysis_status = "Weird"
+
+ checker = pbm_mod.PeriodicChecker(MagicMock())
+ checker.sync_service = MagicMock()
+ callback = MagicMock()
+
+ checker.start_checking(_bv(), analysis_id=2, binary_id=1, callback=callback)
+ before = len(timers)
+ timers[-1][1](*timers[-1][2])
+
+ assert len(timers) == before
+ callback.assert_not_called()
+ checker.sync_service.sync_analysis_data.assert_not_called()

tests/unit/sdk/test_sdk_schemas.py

@@ -223,6 +223,12 @@ def test_task_status_covers_plugin_state_machine():
  assert {"UNINITIALISED", "COMPLETED", "FAILED"} <= set(TaskStatus.__members__)
 
 
+def test_status_input_covers_analysis_state_machine():
+ assert {"UPLOADED", "QUEUED", "PROCESSING", "COMPLETE", "ERROR"} <= set(
+ revengai.StatusInput.__members__
+ )
+
+
 def test_binary_search_result_has_plugin_fields():
  assert {
  "binary_id",