learningcircuit--local-deep-research/tests/web/services/test_research_service_lifecycle.py

"""
Tests for research_service lifecycle management.

Tests cover:
- Resource cleanup
- Progress callbacks
"""

from unittest.mock import Mock, patch
import threading


class TestResourceCleanup:
    """Tests for resource cleanup."""

    @patch("local_deep_research.settings.env_registry.is_test_mode")
    @patch("local_deep_research.web.queue.processor_v2.queue_processor")
    @patch("local_deep_research.web.routes.globals.cleanup_research")
    @patch("local_deep_research.web.services.socket_service.SocketIOService")
    def test_cleanup_removes_from_active_dict(
        self,
        mock_socket,
        mock_cleanup,
        mock_queue,
        mock_test_mode,
    ):
        """Cleanup calls cleanup_research to remove from active dicts."""
        from local_deep_research.web.services.research_service import (
            cleanup_research_resources,
        )

        mock_test_mode.return_value = False

        cleanup_research_resources(123, "testuser")

        mock_cleanup.assert_called_once_with(123)

    @patch("local_deep_research.settings.env_registry.is_test_mode")
    @patch("local_deep_research.web.queue.processor_v2.queue_processor")
    @patch("local_deep_research.web.routes.globals.cleanup_research")
    @patch("local_deep_research.web.services.socket_service.SocketIOService")
    def test_cleanup_notifies_queue_processor(
        self,
        mock_socket,
        mock_cleanup,
        mock_queue,
        mock_test_mode,
    ):
        """Cleanup notifies queue processor of completion."""
        from local_deep_research.web.services.research_service import (
            cleanup_research_resources,
        )

        mock_test_mode.return_value = False

        cleanup_research_resources(123, "testuser")

        mock_queue.notify_research_completed.assert_called_once_with(
            "testuser", 123, user_password=None
        )

    def test_cleanup_socket_emit_success(self):
        """Cleanup emits socket event on success."""
        # Simulate socket emit
        mock_socket = Mock()

        # Emit to subscribers
        mock_socket.emit_to_subscribers(
            "research_complete",
            123,
            {"status": "completed"},
        )

        mock_socket.emit_to_subscribers.assert_called_once()

    @patch("local_deep_research.settings.env_registry.is_test_mode")
    @patch("local_deep_research.web.queue.processor_v2.queue_processor")
    @patch("local_deep_research.web.routes.globals.cleanup_research")
    @patch("local_deep_research.web.services.socket_service.SocketIOService")
    def test_cleanup_socket_emit_failure_handling(
        self,
        mock_socket_class,
        mock_cleanup,
        mock_queue,
        mock_test_mode,
    ):
        """Cleanup handles socket emit failure gracefully."""
        from local_deep_research.web.services.research_service import (
            cleanup_research_resources,
        )

        mock_test_mode.return_value = False
        mock_socket = Mock()
        mock_socket.emit_to_subscribers.side_effect = Exception("Socket error")
        mock_socket_class.return_value = mock_socket

        # Should not raise even with socket error
        cleanup_research_resources(123, "testuser")

    @patch("local_deep_research.settings.env_registry.is_test_mode")
    @patch("local_deep_research.web.queue.processor_v2.queue_processor")
    @patch("local_deep_research.web.routes.globals.cleanup_research")
    @patch("local_deep_research.web.services.socket_service.SocketIOService")
    def test_cleanup_database_session_handling(
        self,
        mock_socket,
        mock_cleanup,
        mock_queue,
        mock_test_mode,
    ):
        """Cleanup handles database session properly."""
        from local_deep_research.web.services.research_service import (
            cleanup_research_resources,
        )

        mock_test_mode.return_value = False

        # Should complete without database errors
        cleanup_research_resources(123, "testuser")

        mock_queue.notify_research_completed.assert_called()

    def test_cleanup_file_handle_closure(self):
        """Cleanup closes file handles."""
        # Simulate file handle management
        file_handles = []

        class MockFileHandle:
            def __init__(self):
                self.closed = False
                file_handles.append(self)

            def close(self):
                self.closed = True

        # Create handles
        MockFileHandle()
        MockFileHandle()

        # Cleanup closes handles
        for handle in file_handles:
            handle.close()

        assert all(h.closed for h in file_handles)

    def test_cleanup_memory_release(self):
        """Cleanup releases memory references."""
        # Simulate memory management
        large_data = {"data": "x" * 10000}
        active_research = {123: {"results": large_data}}

        # Cleanup removes references
        del active_research[123]

        assert 123 not in active_research

    def test_cleanup_concurrent_access_safety(self):
        """Cleanup is thread-safe."""
        active_research = {}
        lock = threading.Lock()
        cleanup_count = [0]

        def cleanup(research_id):
            with lock:
                if research_id in active_research:
                    del active_research[research_id]
                cleanup_count[0] += 1

        # Add some research
        for i in range(10):
            active_research[i] = {"data": i}

        # Concurrent cleanup
        threads = []
        for i in range(10):
            t = threading.Thread(target=cleanup, args=(i,))
            threads.append(t)
            t.start()

        for t in threads:
            t.join()

        assert len(active_research) == 0
        assert cleanup_count[0] == 10


class TestProgressCallbacks:
    """Tests for progress callback functionality."""

    def test_progress_callback_sequencing(self):
        """Progress callbacks are called in sequence."""
        calls = []

        def progress_callback(message, progress, metadata):
            calls.append((message, progress, metadata.get("phase")))

        # Simulate progress sequence
        progress_callback("Starting", 0, {"phase": "init"})
        progress_callback("Searching", 20, {"phase": "search"})
        progress_callback("Analyzing", 50, {"phase": "analysis"})
        progress_callback("Synthesizing", 80, {"phase": "synthesis"})
        progress_callback("Complete", 100, {"phase": "complete"})

        assert len(calls) == 5
        assert calls[0][1] == 0
        assert calls[-1][1] == 100
        assert calls[-1][2] == "complete"

    def test_progress_callback_percentage_accuracy(self):
        """Progress callbacks report accurate percentages."""
        progress_values = []

        def progress_callback(message, progress, metadata):
            if progress is not None:
                progress_values.append(progress)

        # Simulate progress
        for p in [0, 10, 25, 50, 75, 90, 100]:
            progress_callback("Progress", p, {})

        # All values should be in valid range
        assert all(0 <= p <= 100 for p in progress_values)
        # Values should be non-decreasing (progress doesn't go backwards)
        for i in range(1, len(progress_values)):
            assert progress_values[i] >= progress_values[i - 1]

    def test_progress_callback_status_message_updates(self):
        """Progress callbacks include meaningful status messages."""
        messages = []

        def progress_callback(message, progress, metadata):
            messages.append(message)

        progress_callback("Starting research process", 5, {"phase": "init"})
        progress_callback("Searching for information", 30, {"phase": "search"})
        progress_callback(
            "Generating summary", 90, {"phase": "output_generation"}
        )

        assert "Starting" in messages[0]
        assert "Searching" in messages[1]
        assert "summary" in messages[2].lower()

    @patch("local_deep_research.web.services.socket_service.SocketIOService")
    def test_progress_callback_socket_integration(self, mock_socket_class):
        """Progress callbacks integrate with socket service."""
        mock_socket = Mock()
        mock_socket_class.return_value = mock_socket

        def progress_callback(research_id, progress, metadata):
            mock_socket.emit_to_subscribers(
                "progress",
                research_id,
                {"progress": progress, **metadata},
            )

        progress_callback(123, 50, {"phase": "search"})

        mock_socket.emit_to_subscribers.assert_called_once_with(
            "progress",
            123,
            {"progress": 50, "phase": "search"},
        )

    @patch("local_deep_research.web.queue.processor_v2.queue_processor")
    def test_progress_callback_database_update(self, mock_queue):
        """Progress callbacks queue database updates."""

        def progress_callback(username, research_id, progress):
            mock_queue.queue_progress_update(username, research_id, progress)

        progress_callback("testuser", 123, 75)

        mock_queue.queue_progress_update.assert_called_once_with(
            "testuser", 123, 75
        )

    def test_progress_callback_rate_limiting(self):
        """Progress callbacks respect rate limiting."""
        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: DELETE).
        import time

        last_call_time = [0]
        min_interval = 0.1  # 100ms
        calls_made = []

        def rate_limited_callback(progress):
            current_time = time.time()
            if current_time - last_call_time[0] >= min_interval:
                calls_made.append(progress)
                last_call_time[0] = current_time

        # Rapid calls
        for i in range(10):
            rate_limited_callback(i * 10)
            time.sleep(0.05)  # 50ms between calls

        # Only about half should go through due to rate limiting
        assert len(calls_made) < 10

    def test_progress_callback_error_handling(self):
        """Progress callbacks handle errors gracefully."""
        errors = []

        def progress_callback(progress, metadata):
            try:
                if metadata.get("fail"):
                    raise ValueError("Callback error")
            except Exception as e:
                errors.append(str(e))

        # Normal call
        progress_callback(50, {})

        # Error call
        progress_callback(60, {"fail": True})

        assert len(errors) == 1
        assert "Callback error" in errors[0]


class TestResearchPhaseTracking:
    """Tests for research phase tracking."""

    def test_phase_transition_init_to_search(self):
        """Phase transitions from init to search."""
        phases = []

        def track_phase(phase):
            phases.append(phase)

        track_phase("init")
        track_phase("search")

        assert phases == ["init", "search"]

    def test_phase_transition_search_to_analysis(self):
        """Phase transitions from search to analysis."""
        current_phase = "search"
        next_phase = "analysis"

        assert current_phase != next_phase

    def test_phase_transition_analysis_to_synthesis(self):
        """Phase transitions from analysis to synthesis."""
        phases = ["init", "search", "analysis", "synthesis"]

        # Synthesis follows analysis
        analysis_idx = phases.index("analysis")
        synthesis_idx = phases.index("synthesis")

        assert synthesis_idx == analysis_idx + 1

    def test_phase_transition_synthesis_to_complete(self):
        """Phase transitions from synthesis to complete."""
        all_phases = [
            "init",
            "search",
            "analysis",
            "synthesis",
            "output_generation",
            "complete",
        ]

        # Complete is the final phase
        assert all_phases[-1] == "complete"

    def test_phase_metadata_preservation(self):
        """Phase metadata is preserved across transitions."""
        phase_metadata = {}

        def update_phase(phase, **kwargs):
            phase_metadata[phase] = kwargs

        update_phase("search", engine="google", results_count=10)
        update_phase("analysis", findings_count=5)
        update_phase("synthesis", tokens_used=1000)

        assert phase_metadata["search"]["results_count"] == 10
        assert phase_metadata["analysis"]["findings_count"] == 5
        assert phase_metadata["synthesis"]["tokens_used"] == 1000


class TestActiveResearchManagement:
    """Tests for active research dictionary management."""

    def test_active_research_creation(self):
        """Active research entry is created correctly."""
        active_research = {}

        active_research[123] = {
            "thread": Mock(),
            "progress": 0,
            "status": "in_progress",
            "log": [],
            "settings": {"model": "gpt-4"},
        }

        assert 123 in active_research
        assert active_research[123]["status"] == "in_progress"

    def test_active_research_progress_update(self):
        """Active research progress is updated correctly."""
        active_research = {123: {"progress": 0, "status": "in_progress"}}

        active_research[123]["progress"] = 50

        assert active_research[123]["progress"] == 50

    def test_active_research_status_transition(self):
        """Active research status transitions correctly."""
        active_research = {123: {"status": "in_progress"}}

        # Transition to completed
        active_research[123]["status"] = "completed"

        assert active_research[123]["status"] == "completed"

    def test_active_research_removal(self):
        """Active research entry is removed on cleanup."""
        active_research = {123: {"status": "completed"}}

        del active_research[123]

        assert 123 not in active_research

    def test_active_research_thread_safety(self):
        """Active research is thread-safe with locking."""
        active_research = {}
        lock = threading.Lock()

        def add_research(research_id):
            with lock:
                active_research[research_id] = {"status": "in_progress"}

        def remove_research(research_id):
            with lock:
                if research_id in active_research:
                    del active_research[research_id]

        # Add and remove concurrently
        threads = []
        for i in range(10):
            t1 = threading.Thread(target=add_research, args=(i,))
            threads.append(t1)
            t1.start()

        for t in threads:
            t.join()

        # All should be added
        assert len(active_research) == 10

        # Remove all
        threads = []
        for i in range(10):
            t = threading.Thread(target=remove_research, args=(i,))
            threads.append(t)
            t.start()

        for t in threads:
            t.join()

        assert len(active_research) == 0