learningcircuit--local-deep-research/tests/benchmarks/efficiency/test_resource_monitor_coverage.py

"""
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