learningcircuit--local-deep-research/tests/web/routes/test_globals_gaps.py

"""Tests for globals.py — gap coverage.

Covers the three functions with zero test coverage:
- get_usernames_with_active_research
- is_research_thread_alive
- update_progress_and_check_active
"""

from unittest.mock import Mock

import pytest

from local_deep_research.web.routes import globals as g


@pytest.fixture(autouse=True)
def _clean_globals():
    """Reset all global state before each test."""
    with g._lock:
        g._active_research.clear()
        g._termination_flags.clear()
    yield
    with g._lock:
        g._active_research.clear()
        g._termination_flags.clear()


# ===================================================================
# get_usernames_with_active_research
# ===================================================================


class TestGetUsernamesWithActiveResearch:
    def test_empty_returns_empty_set(self):
        assert g.get_usernames_with_active_research() == set()

    def test_entry_with_username(self):
        g.set_active_research("r1", {"settings": {"username": "alice"}})
        assert g.get_usernames_with_active_research() == {"alice"}

    def test_entry_with_username_none_filtered(self):
        g.set_active_research("r1", {"settings": {"username": None}})
        assert g.get_usernames_with_active_research() == set()

    def test_entry_with_no_settings_key(self):
        g.set_active_research("r1", {"progress": 0})
        assert g.get_usernames_with_active_research() == set()

    def test_duplicate_usernames_deduplicated(self):
        g.set_active_research("r1", {"settings": {"username": "alice"}})
        g.set_active_research("r2", {"settings": {"username": "alice"}})
        assert g.get_usernames_with_active_research() == {"alice"}

    def test_different_usernames(self):
        g.set_active_research("r1", {"settings": {"username": "alice"}})
        g.set_active_research("r2", {"settings": {"username": "bob"}})
        assert g.get_usernames_with_active_research() == {"alice", "bob"}


# ===================================================================
# is_research_thread_alive
# ===================================================================


class TestIsResearchThreadAlive:
    def test_unknown_research_id(self):
        assert g.is_research_thread_alive("r1") is False

    def test_entry_with_no_thread_key(self):
        g.set_active_research("r1", {"progress": 0})
        assert g.is_research_thread_alive("r1") is False

    def test_entry_with_thread_none(self):
        g.set_active_research("r1", {"thread": None})
        assert g.is_research_thread_alive("r1") is False

    def test_thread_alive(self):
        mock_thread = Mock()
        mock_thread.is_alive.return_value = True
        g.set_active_research("r1", {"thread": mock_thread})
        assert g.is_research_thread_alive("r1") is True

    def test_thread_dead(self):
        mock_thread = Mock()
        mock_thread.is_alive.return_value = False
        g.set_active_research("r1", {"thread": mock_thread})
        assert g.is_research_thread_alive("r1") is False


# ===================================================================
# update_progress_and_check_active
# ===================================================================


class TestUpdateProgressAndCheckActive:
    def test_not_found(self):
        assert g.update_progress_and_check_active("r1", 50) == (None, False)

    def test_new_progress_higher_updates(self):
        g.set_active_research("r1", {"progress": 10})
        result = g.update_progress_and_check_active("r1", 50)
        assert result == (50, True)
        assert g.get_research_field("r1", "progress") == 50

    def test_new_progress_lower_no_update(self):
        g.set_active_research("r1", {"progress": 80})
        result = g.update_progress_and_check_active("r1", 50)
        assert result == (80, True)
        assert g.get_research_field("r1", "progress") == 80

    def test_new_progress_equal_no_update(self):
        g.set_active_research("r1", {"progress": 50})
        result = g.update_progress_and_check_active("r1", 50)
        assert result == (50, True)

    def test_new_progress_none(self):
        g.set_active_research("r1", {"progress": 30})
        result = g.update_progress_and_check_active("r1", None)
        assert result == (30, True)

    def test_no_progress_key_defaults_to_zero(self):
        g.set_active_research("r1", {})
        result = g.update_progress_and_check_active("r1", 10)
        assert result == (10, True)
        assert g.get_research_field("r1", "progress") == 10


# ===================================================================
# check_and_start_research
# ===================================================================


class TestCheckAndStartResearch:
    def _make_data(self, alive=True):
        thread = Mock()
        thread.is_alive.return_value = alive
        return thread, {
            "thread": thread,
            "progress": 0,
            "status": "in_progress",
            "log": [],
            "settings": {},
        }

    def test_registers_when_no_prior_entry(self):
        thread, data = self._make_data()
        result = g.check_and_start_research("r1", data)
        assert result is True
        thread.start.assert_called_once()
        assert g.is_research_thread_alive("r1")

    def test_refuses_when_existing_thread_alive(self):
        existing_thread = Mock()
        existing_thread.is_alive.return_value = True
        g.set_active_research("r1", {"thread": existing_thread})

        new_thread, new_data = self._make_data()
        result = g.check_and_start_research("r1", new_data)

        assert result is False
        new_thread.start.assert_not_called()
        # Existing entry is untouched.
        assert g.get_research_field("r1", "thread") is existing_thread

    def test_overwrites_when_existing_thread_finished(self):
        finished_thread = Mock()
        finished_thread.is_alive.return_value = False
        g.set_active_research("r1", {"thread": finished_thread})

        new_thread, new_data = self._make_data()
        result = g.check_and_start_research("r1", new_data)

        assert result is True
        new_thread.start.assert_called_once()
        assert g.get_research_field("r1", "thread") is new_thread

    def test_overwrites_when_existing_has_no_thread(self):
        g.set_active_research("r1", {"progress": 50})

        new_thread, new_data = self._make_data()
        result = g.check_and_start_research("r1", new_data)

        assert result is True
        new_thread.start.assert_called_once()

    def test_rejects_data_without_thread(self):
        with pytest.raises(ValueError):
            g.check_and_start_research("r1", {"progress": 0})

    def test_rejects_data_with_thread_none(self):
        with pytest.raises(ValueError):
            g.check_and_start_research("r1", {"thread": None})