From 07cfb81980fbb039e6a3fa40d4d2e379f8813cc9 Mon Sep 17 00:00:00 2001 From: Akash Bhattacharya Date: Tue, 25 Aug 2026 15:47:42 +0100 Subject: [PATCH] fix(PLU-326): stop treating failed analyses as complete in status poller **Context:** `PeriodicChecker._worker` treated any status other than `Queued`/`Processing` as a completed, successful analysis, including `Error`. That persisted the failed analysis as attached and then crashed on a downstream 422, since the backend has no function map to return for a failed analysis. **This diff:** - Branches on the real `revengai.StatusInput` enum: in-progress statuses (`Uploaded`/`Queued`/`Processing`) reschedule as before, `Complete` runs the existing success path, and anything else (`Error`, or unrecognised) is now a terminal failure that is logged via `log_error` with no callback, sync, or persistence. - Rewords the two `except` blocks' log messages, which wrap the whole worker rather than just the status check and previously mislabeled failures as "during status check". - Removes the dead, never-emitted `update_text_signal`/`_update_text_slot` Qt signal copy-pasted from an unrelated class. --- .../monitoring/process_binary_monitor.py | 81 ++++++++++++------- .../monitoring/test_process_binary_monitor.py | 71 ++++++++++++++++ tests/unit/sdk/test_sdk_schemas.py | 6 ++ 3 files changed, 127 insertions(+), 31 deletions(-) diff --git a/reai_toolkit/utils/monitoring/process_binary_monitor.py b/reai_toolkit/utils/monitoring/process_binary_monitor.py index 6839e6f..ffd494b 100755 --- a/reai_toolkit/utils/monitoring/process_binary_monitor.py +++ b/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}]" - ) \ No newline at end of file + ) diff --git a/tests/unit/monitoring/test_process_binary_monitor.py b/tests/unit/monitoring/test_process_binary_monitor.py index 5c11ecb..b8270df 100644 --- a/tests/unit/monitoring/test_process_binary_monitor.py +++ b/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() diff --git a/tests/unit/sdk/test_sdk_schemas.py b/tests/unit/sdk/test_sdk_schemas.py index 3e49bc3..eeed6eb 100644 --- a/tests/unit/sdk/test_sdk_schemas.py +++ b/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",