test: cover background output manager edge cases

Author: harvisaiagent-bot

vulcano/app/background_test.py

@@ -1,8 +1,10 @@
 """Tests for background task management."""
 
+import queue
 import threading
 import time
 import unittest
+import unittest.mock
 
 from vulcano.app.background import (
     BackgroundTaskManager,
@@ -244,6 +246,57 @@ def worker():
         self.assertEqual(len(results), 1)
         self.assertEqual(results[0], "done")
 
+    def test_mark_completed_triggers_ui_invalidate_callback(self):
+        """Completing a task invalidates the UI when callback is configured."""
+
+        callback = unittest.mock.MagicMock()
+        self.manager.set_ui_invalidate_callback(callback)
+        thread = threading.Thread(target=lambda: None)
+        task_id = self.manager.register_task("task", thread)
+
+        self.manager.mark_completed(task_id)
+
+        callback.assert_called_once()
+
+    def test_mark_failed_triggers_ui_invalidate_callback(self):
+        """Failing a task invalidates the UI when callback is configured."""
+
+        callback = unittest.mock.MagicMock()
+        self.manager.set_ui_invalidate_callback(callback)
+        thread = threading.Thread(target=lambda: None)
+        task_id = self.manager.register_task("task", thread)
+
+        self.manager.mark_failed(task_id, RuntimeError("boom"))
+
+        callback.assert_called_once()
+
+    def test_enqueue_output_triggers_ui_invalidate_callback(self):
+        """Enqueuing output invalidates the UI when callback is configured."""
+
+        callback = unittest.mock.MagicMock()
+        self.manager.set_ui_invalidate_callback(callback)
+
+        self.manager.enqueue_output("task_0", "hello")
+
+        callback.assert_called_once()
+
+    def test_get_queued_output_handles_empty_race(self):
+        """Queue empty races are handled gracefully."""
+
+        class RaceQueue(object):
+            def empty(self):
+                return False
+
+            def get_nowait(self):
+                raise queue.Empty
+
+        original_queue = self.manager._output_queue
+        self.manager._output_queue = RaceQueue()
+        try:
+            self.assertEqual(self.manager.get_queued_output(), [])
+        finally:
+            self.manager._output_queue = original_queue
+
 
 if __name__ == "__main__":
     unittest.main()

vulcano/app/classes_test.py

@@ -655,3 +655,37 @@ def queue_output():
             "task_0" in call and "test message" in call for call in print_calls
         )
         self.assertTrue(found_message)
+
+    @patch(print_builtin)
+    @patch("vulcano.app.classes.print_formatted_text")
+    @patch("vulcano.app.classes.sys")
+    def test_display_background_output_falls_back_to_print(
+        self, sys_mock, prompt_print_mock, print_mock
+    ):
+        """Prompt-toolkit print failures fall back to plain print."""
+        sys_mock.argv = ["ensure_no_repl", "noop"]
+        app = VulcanoApp()
+        app.background_tasks.enqueue_output("task_0", "hello")
+        prompt_print_mock.side_effect = RuntimeError("not in prompt")
+
+        app._display_background_output(use_prompt_toolkit=True)
+
+        print_mock.assert_called_with("[task_0] hello")
+
+    @patch("vulcano.app.classes.threading.Thread")
+    @patch("vulcano.app.classes.sys")
+    def test_cli_mode_skips_wait_when_no_background_tasks(self, sys_mock, thread_mock):
+        """CLI mode does not wait when there are no active background tasks."""
+        sys_mock.argv = ["ensure_no_repl", "noop"]
+        app = VulcanoApp()
+        app.background_tasks.wait_for_all_tasks = MagicMock()
+
+        @app.command()
+        def noop():
+            return "ok"
+
+        app.run(print_result=False)
+
+        app.background_tasks.wait_for_all_tasks.assert_not_called()
+        thread_mock.return_value.start.assert_called_once()
+        thread_mock.return_value.join.assert_called_once_with(timeout=1.0)