""" Coverage tests for benchmarks/efficiency/resource_monitor.py. Covers paths not exercised by the existing high-value test file: - _monitor_resources thread: collects process/system data, skips disabled tracking, handles exceptions during collection - Start/stop cycle: creates thread, joins thread, sets flag, resets data - Context manager: delegates to start/stop and collects data - print_summary: output with both sections, system-only section - get_combined_stats: zero system memory edge case, no system stats present """ from unittest.mock import MagicMock, patch from local_deep_research.benchmarks.efficiency.resource_monitor import ( ResourceMonitor, ) # --------------------------------------------------------------------------- # _monitor_resources thread # --------------------------------------------------------------------------- class TestMonitorResourcesThread: """Tests for the _monitor_resources background collection loop.""" @patch( "local_deep_research.benchmarks.efficiency.resource_monitor.PSUTIL_AVAILABLE", True, ) @patch("local_deep_research.benchmarks.efficiency.resource_monitor.psutil") @patch("local_deep_research.benchmarks.efficiency.resource_monitor.time") def test_collects_process_and_system_data(self, mock_time, mock_psutil): """A single iteration collects both process and system samples.""" mock_time.time.return_value = 1000.0 # Stop after one sleep def sleep_side_effect(_interval): monitor.monitoring = False mock_time.sleep.side_effect = sleep_side_effect # Mock process mock_process = MagicMock() mock_process.cpu_percent.return_value = 12.5 mem_info = MagicMock( rss=50 * 1024 * 1024, vms=100 * 1024 * 1024, shared=10 ) mock_process.memory_info.return_value = mem_info mock_process.num_threads.return_value = 3 mock_process.open_files.return_value = [MagicMock()] mock_process.status.return_value = "running" mock_psutil.Process.return_value = mock_process # Mock system calls mock_psutil.cpu_percent.return_value = 40.0 mock_psutil.virtual_memory.return_value = MagicMock( total=8 * 1024**3, available=4 * 1024**3, used=4 * 1024**3, percent=50.0, ) mock_psutil.disk_usage.return_value = MagicMock( total=256 * 1024**3, used=128 * 1024**3, percent=50.0, ) monitor = ResourceMonitor(sampling_interval=0.01) monitor.monitoring = True monitor._monitor_resources() assert len(monitor.process_data) == 1 assert monitor.process_data[0]["cpu_percent"] == 12.5 assert monitor.process_data[0]["memory_rss"] == 50 * 1024 * 1024 assert monitor.process_data[0]["num_threads"] == 3 assert monitor.process_data[0]["open_files"] == 1 assert monitor.process_data[0]["status"] == "running" assert len(monitor.system_data) == 1 assert monitor.system_data[0]["cpu_percent"] == 40.0 assert monitor.system_data[0]["memory_percent"] == 50.0 @patch( "local_deep_research.benchmarks.efficiency.resource_monitor.PSUTIL_AVAILABLE", True, ) @patch("local_deep_research.benchmarks.efficiency.resource_monitor.psutil") @patch("local_deep_research.benchmarks.efficiency.resource_monitor.time") def test_skips_process_tracking_when_disabled(self, mock_time, mock_psutil): """When track_process=False no process data is collected.""" mock_time.time.return_value = 1000.0 def sleep_side_effect(_interval): monitor.monitoring = False mock_time.sleep.side_effect = sleep_side_effect mock_psutil.Process.return_value = MagicMock() mock_psutil.cpu_percent.return_value = 10.0 mock_psutil.virtual_memory.return_value = MagicMock( total=8 * 1024**3, available=4 * 1024**3, used=4 * 1024**3, percent=50.0, ) mock_psutil.disk_usage.return_value = MagicMock( total=256 * 1024**3, used=128 * 1024**3, percent=50.0, ) monitor = ResourceMonitor(track_process=False, track_system=True) monitor.monitoring = True monitor._monitor_resources() assert len(monitor.process_data) == 0 assert len(monitor.system_data) == 1 @patch( "local_deep_research.benchmarks.efficiency.resource_monitor.PSUTIL_AVAILABLE", True, ) @patch("local_deep_research.benchmarks.efficiency.resource_monitor.psutil") @patch("local_deep_research.benchmarks.efficiency.resource_monitor.time") def test_skips_system_tracking_when_disabled(self, mock_time, mock_psutil): """When track_system=False no system data is collected.""" mock_time.time.return_value = 1000.0 def sleep_side_effect(_interval): monitor.monitoring = False mock_time.sleep.side_effect = sleep_side_effect mock_process = MagicMock() mock_process.cpu_percent.return_value = 5.0 mem_info = MagicMock(rss=30 * 1024 * 1024, vms=60 * 1024 * 1024) mock_process.memory_info.return_value = mem_info mock_process.num_threads.return_value = 2 mock_process.open_files.return_value = [] mock_process.status.return_value = "sleeping" mock_psutil.Process.return_value = mock_process monitor = ResourceMonitor(track_process=True, track_system=False) monitor.monitoring = True monitor._monitor_resources() assert len(monitor.process_data) == 1 assert len(monitor.system_data) == 0 @patch( "local_deep_research.benchmarks.efficiency.resource_monitor.PSUTIL_AVAILABLE", True, ) @patch("local_deep_research.benchmarks.efficiency.resource_monitor.psutil") @patch("local_deep_research.benchmarks.efficiency.resource_monitor.time") def test_handles_exception_during_collection(self, mock_time, mock_psutil): """An exception inside the loop is caught; monitoring continues.""" mock_time.time.return_value = 1000.0 iteration = {"count": 0} def sleep_side_effect(_interval): iteration["count"] += 1 if iteration["count"] >= 2: monitor.monitoring = False mock_time.sleep.side_effect = sleep_side_effect mock_process = MagicMock() # First call raises, second call succeeds mock_process.cpu_percent.side_effect = [RuntimeError("boom"), 10.0] mem_info = MagicMock(rss=50 * 1024 * 1024, vms=100 * 1024 * 1024) mock_process.memory_info.return_value = mem_info mock_process.num_threads.return_value = 2 mock_process.open_files.return_value = [] mock_process.status.return_value = "running" mock_psutil.Process.return_value = mock_process monitor = ResourceMonitor() monitor.monitoring = True monitor._monitor_resources() # First iteration raised, second succeeded => 1 sample assert len(monitor.process_data) == 1 @patch( "local_deep_research.benchmarks.efficiency.resource_monitor.PSUTIL_AVAILABLE", False, ) def test_returns_immediately_without_psutil(self): """_monitor_resources returns immediately when psutil unavailable.""" monitor = ResourceMonitor() monitor.monitoring = True monitor._monitor_resources() assert len(monitor.process_data) == 0 assert len(monitor.system_data) == 0 # --------------------------------------------------------------------------- # Start / stop lifecycle # --------------------------------------------------------------------------- class TestStartStopCycle: """Tests for the start/stop lifecycle with thread management.""" @patch( "local_deep_research.benchmarks.efficiency.resource_monitor.PSUTIL_AVAILABLE", True, ) @patch( "local_deep_research.benchmarks.efficiency.resource_monitor.threading" ) @patch("local_deep_research.benchmarks.efficiency.resource_monitor.time") def test_start_creates_and_starts_thread(self, mock_time, mock_threading): """start() creates a daemon thread and calls .start() on it.""" mock_time.time.return_value = 5000.0 mock_thread = MagicMock() mock_threading.Thread.return_value = mock_thread monitor = ResourceMonitor() monitor.can_monitor = True monitor.start() mock_threading.Thread.assert_called_once() mock_thread.start.assert_called_once() assert monitor.monitoring is True assert monitor.start_time == 5000.0 assert monitor.process_data == [] assert monitor.system_data == [] @patch( "local_deep_research.benchmarks.efficiency.resource_monitor.PSUTIL_AVAILABLE", True, ) @patch( "local_deep_research.benchmarks.efficiency.resource_monitor.threading" ) @patch("local_deep_research.benchmarks.efficiency.resource_monitor.time") def test_stop_joins_thread_and_resets(self, mock_time, mock_threading): """stop() joins the thread, clears it, and records end_time.""" mock_time.time.return_value = 6000.0 mock_thread = MagicMock() mock_threading.Thread.return_value = mock_thread monitor = ResourceMonitor() monitor.can_monitor = True monitor.start() mock_time.time.return_value = 7000.0 monitor.stop() mock_thread.join.assert_called_once_with(timeout=2.0) assert monitor.monitoring is False assert monitor.end_time == 7000.0 assert monitor.monitor_thread is None @patch( "local_deep_research.benchmarks.efficiency.resource_monitor.PSUTIL_AVAILABLE", True, ) @patch( "local_deep_research.benchmarks.efficiency.resource_monitor.threading" ) @patch("local_deep_research.benchmarks.efficiency.resource_monitor.time") def test_start_when_already_monitoring_is_noop( self, mock_time, mock_threading ): """Calling start() a second time while monitoring logs warning only.""" mock_time.time.return_value = 5000.0 mock_thread = MagicMock() mock_threading.Thread.return_value = mock_thread monitor = ResourceMonitor() monitor.can_monitor = True monitor.start() monitor.start() # second call should not create another thread assert mock_threading.Thread.call_count == 1 # --------------------------------------------------------------------------- # Context manager # --------------------------------------------------------------------------- class TestContextManager: """Tests for the monitor() context manager.""" @patch( "local_deep_research.benchmarks.efficiency.resource_monitor.PSUTIL_AVAILABLE", True, ) @patch( "local_deep_research.benchmarks.efficiency.resource_monitor.threading" ) @patch("local_deep_research.benchmarks.efficiency.resource_monitor.time") def test_context_manager_calls_start_and_stop( self, mock_time, mock_threading ): """The context manager starts monitoring on enter and stops on exit.""" mock_time.time.return_value = 1000.0 mock_thread = MagicMock() mock_threading.Thread.return_value = mock_thread monitor = ResourceMonitor() monitor.can_monitor = True with monitor.monitor(): assert monitor.monitoring is True assert monitor.monitoring is False mock_thread.join.assert_called_once() # --------------------------------------------------------------------------- # print_summary # --------------------------------------------------------------------------- class TestPrintSummary: """Tests for print_summary output.""" def test_print_summary_with_both_sections(self, capsys): """print_summary outputs both process and system sections.""" monitor = ResourceMonitor() monitor.start_time = 100.0 monitor.end_time = 110.0 monitor.process_data = [ { "timestamp": 101.0, "cpu_percent": 15.0, "memory_rss": 64 * 1024 * 1024, "memory_vms": 128 * 1024 * 1024, "memory_shared": 0, "num_threads": 5, "open_files": 1, "status": "running", }, ] monitor.system_data = [ { "timestamp": 101.0, "cpu_percent": 25.0, "memory_total": 16 * 1024**3, "memory_available": 8 * 1024**3, "memory_used": 8 * 1024**3, "memory_percent": 50.0, "disk_total": 500 * 1024**3, "disk_used": 250 * 1024**3, "disk_percent": 50.0, }, ] monitor.print_summary() captured = capsys.readouterr().out assert "Process Resources" in captured assert "System Resources" in captured assert "15.0%" in captured assert "64.0 MB" in captured assert "Threads: 5" in captured def test_print_summary_system_only(self, capsys): """print_summary outputs system section only when no process data.""" monitor = ResourceMonitor() monitor.start_time = 100.0 monitor.end_time = 110.0 monitor.system_data = [ { "timestamp": 101.0, "cpu_percent": 30.0, "memory_total": 8 * 1024**3, "memory_available": 4 * 1024**3, "memory_used": 4 * 1024**3, "memory_percent": 50.0, "disk_total": 256 * 1024**3, "disk_used": 128 * 1024**3, "disk_percent": 50.0, }, ] monitor.print_summary() captured = capsys.readouterr().out assert "Process Resources" not in captured assert "System Resources" in captured # --------------------------------------------------------------------------- # get_combined_stats edge cases # --------------------------------------------------------------------------- class TestGetCombinedStatsEdgeCases: """Edge-case tests for get_combined_stats.""" def test_zero_system_memory_yields_zero_percent(self): """When system_memory_mb is 0, process_memory_percent is 0.""" monitor = ResourceMonitor() monitor.start_time = 100.0 monitor.end_time = 110.0 monitor.process_data = [ { "timestamp": 101.0, "cpu_percent": 10.0, "memory_rss": 50 * 1024 * 1024, "memory_vms": 100 * 1024 * 1024, "memory_shared": 0, "num_threads": 2, "open_files": 0, "status": "running", }, ] monitor.system_data = [ { "timestamp": 101.0, "cpu_percent": 20.0, "memory_total": 0, # zero total memory "memory_available": 0, "memory_used": 0, "memory_percent": 0.0, "disk_total": 100 * 1024**3, "disk_used": 50 * 1024**3, "disk_percent": 50.0, }, ] stats = monitor.get_combined_stats() assert stats["process_memory_percent"] == 0 def test_no_system_stats_omits_system_keys(self): """When there is no system data, combined stats have no system_ keys.""" monitor = ResourceMonitor() monitor.start_time = 100.0 monitor.end_time = 110.0 monitor.process_data = [ { "timestamp": 101.0, "cpu_percent": 10.0, "memory_rss": 50 * 1024 * 1024, "memory_vms": 100 * 1024 * 1024, "memory_shared": 0, "num_threads": 2, "open_files": 0, "status": "running", }, ] stats = monitor.get_combined_stats() assert "process_cpu_avg" in stats system_keys = [k for k in stats if k.startswith("system_")] assert len(system_keys) == 0 assert "process_memory_percent" not in stats