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

"""
Tests for globals.py — thread-safe accessor functions.

Tests cover:
- Accessor functions for active_research, termination_flags
- Thread safety (concurrent read/write)
- Atomic operations (update_progress_if_higher, cleanup_research)
- Snapshot isolation (returned data cannot mutate shared state)
"""

import threading

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()


# ===================================================================
# active_research accessors
# ===================================================================


class TestIsResearchActive:
    def test_returns_false_when_empty(self):
        assert g.is_research_active("r1") is False

    def test_returns_true_when_present(self):
        g.set_active_research("r1", {"progress": 0})
        assert g.is_research_active("r1") is True

    def test_returns_false_after_removal(self):
        g.set_active_research("r1", {"progress": 0})
        g.remove_active_research("r1")
        assert g.is_research_active("r1") is False


class TestGetActiveResearchIds:
    def test_empty(self):
        assert g.get_active_research_ids() == []

    def test_returns_all_ids(self):
        g.set_active_research("r1", {})
        g.set_active_research("r2", {})
        ids = g.get_active_research_ids()
        assert sorted(ids) == ["r1", "r2"]

    def test_returns_snapshot(self):
        """Returned list should not be affected by later mutations."""
        g.set_active_research("r1", {})
        ids = g.get_active_research_ids()
        g.set_active_research("r2", {})
        assert ids == ["r1"]


class TestGetActiveResearchSnapshot:
    def test_returns_none_when_missing(self):
        assert g.get_active_research_snapshot("r1") is None

    def test_returns_projected_fields(self):
        g.set_active_research(
            "r1",
            {
                "thread": threading.current_thread(),
                "progress": 42,
                "status": "in_progress",
                "log": [{"time": "t1", "message": "msg"}],
                "settings": {"key": "val"},
            },
        )
        snap = g.get_active_research_snapshot("r1")
        assert snap["progress"] == 42
        assert snap["status"] == "in_progress"
        assert len(snap["log"]) == 1
        assert snap["settings"] == {"key": "val"}
        assert "thread" not in snap

    def test_log_is_a_copy(self):
        """Mutating the returned log should not affect shared state."""
        g.set_active_research(
            "r1",
            {
                "log": [{"m": "a"}],
                "progress": 0,
                "status": None,
                "settings": None,
            },
        )
        snap = g.get_active_research_snapshot("r1")
        snap["log"].append({"m": "b"})
        assert len(g.get_active_research_snapshot("r1")["log"]) == 1

    def test_settings_is_a_copy(self):
        """Mutating returned settings must not affect internal state."""
        g.set_active_research(
            "r1", {"settings": {"key": "val"}, "log": [], "progress": 0}
        )
        snap = g.get_active_research_snapshot("r1")
        snap["settings"]["injected"] = True
        assert (
            "injected" not in g.get_active_research_snapshot("r1")["settings"]
        )


class TestGetResearchField:
    def test_returns_default_when_missing(self):
        assert g.get_research_field("r1", "progress", 99) == 99

    def test_returns_value(self):
        g.set_active_research("r1", {"progress": 50})
        assert g.get_research_field("r1", "progress") == 50

    def test_returns_shallow_copy_of_list(self):
        g.set_active_research("r1", {"log": [1, 2, 3]})
        log = g.get_research_field("r1", "log")
        log.append(4)
        assert g.get_research_field("r1", "log") == [1, 2, 3]

    def test_returns_shallow_copy_of_dict(self):
        g.set_active_research("r1", {"settings": {"a": 1}})
        s = g.get_research_field("r1", "settings")
        s["b"] = 2
        assert "b" not in g.get_research_field("r1", "settings")


class TestSetActiveResearch:
    def test_insert(self):
        g.set_active_research("r1", {"progress": 0})
        assert g.is_research_active("r1")

    def test_overwrite(self):
        g.set_active_research("r1", {"progress": 10})
        g.set_active_research("r1", {"progress": 20})
        assert g.get_research_field("r1", "progress") == 20


class TestUpdateActiveResearch:
    def test_updates_existing(self):
        g.set_active_research("r1", {"progress": 0, "status": "pending"})
        g.update_active_research("r1", progress=50)
        assert g.get_research_field("r1", "progress") == 50
        assert g.get_research_field("r1", "status") == "pending"

    def test_noop_when_missing(self):
        # Should not raise
        g.update_active_research("r1", progress=50)


class TestAppendResearchLog:
    def test_appends(self):
        g.set_active_research("r1", {"log": []})
        g.append_research_log("r1", {"time": "t1"})
        g.append_research_log("r1", {"time": "t2"})
        assert g.get_research_field("r1", "log") == [
            {"time": "t1"},
            {"time": "t2"},
        ]

    def test_creates_log_list_if_missing(self):
        g.set_active_research("r1", {})
        g.append_research_log("r1", {"time": "t1"})
        assert g.get_research_field("r1", "log") == [{"time": "t1"}]

    def test_noop_when_research_missing(self):
        g.append_research_log("r1", {"time": "t1"})


class TestUpdateProgressIfHigher:
    def test_updates_when_higher(self):
        g.set_active_research("r1", {"progress": 10})
        result = g.update_progress_if_higher("r1", 50)
        assert result == 50
        assert g.get_research_field("r1", "progress") == 50

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

    def test_returns_none_when_missing(self):
        assert g.update_progress_if_higher("r1", 50) is None

    def test_handles_none_progress(self):
        g.set_active_research("r1", {"progress": 30})
        result = g.update_progress_if_higher("r1", None)
        assert result == 30


class TestRemoveActiveResearch:
    def test_removes(self):
        g.set_active_research("r1", {})
        g.remove_active_research("r1")
        assert not g.is_research_active("r1")

    def test_noop_when_missing(self):
        g.remove_active_research("r1")  # Should not raise


class TestIterActiveResearch:
    def test_yields_all_entries(self):
        g.set_active_research("r1", {"progress": 10})
        g.set_active_research("r2", {"progress": 20})
        entries = dict(g.iter_active_research())
        assert "r1" in entries
        assert "r2" in entries
        assert entries["r1"]["progress"] == 10

    def test_yields_copies(self):
        g.set_active_research("r1", {"progress": 10})
        for _, data in g.iter_active_research():
            data["progress"] = 999
        assert g.get_research_field("r1", "progress") == 10

    def test_does_not_hold_lock_during_iteration(self):
        """Another thread should be able to write while we iterate."""
        g.set_active_research("r1", {"progress": 10})
        g.set_active_research("r2", {"progress": 20})

        write_succeeded = threading.Event()

        for _, _ in g.iter_active_research():
            # Mid-iteration: a writer thread should not be blocked
            def writer():
                g.set_active_research("r3", {"progress": 30})
                write_succeeded.set()

            t = threading.Thread(target=writer)
            t.start()
            t.join(timeout=2)
            break  # Only need one iteration to test

        assert write_succeeded.is_set(), "Writer was blocked by iter lock"


class TestGetActiveResearchCount:
    def test_empty(self):
        assert g.get_active_research_count() == 0

    def test_counts(self):
        g.set_active_research("r1", {})
        g.set_active_research("r2", {})
        assert g.get_active_research_count() == 2


# ===================================================================
# termination_flags accessors
# ===================================================================


class TestTerminationFlags:
    def test_default_is_false(self):
        assert g.is_termination_requested("r1") is False

    def test_set_and_check(self):
        g.set_termination_flag("r1")
        assert g.is_termination_requested("r1") is True

    def test_clear(self):
        g.set_termination_flag("r1")
        g.clear_termination_flag("r1")
        assert g.is_termination_requested("r1") is False

    def test_clear_noop_when_missing(self):
        g.clear_termination_flag("r1")  # Should not raise

    def test_independent_flags(self):
        g.set_termination_flag("r1")
        assert g.is_termination_requested("r1") is True
        assert g.is_termination_requested("r2") is False


# ===================================================================
# cleanup_research
# ===================================================================


class TestCleanupResearch:
    def test_removes_from_all_dicts(self):
        g.set_active_research("r1", {"progress": 50})
        g.set_termination_flag("r1")
        g.cleanup_research("r1")
        assert not g.is_research_active("r1")
        assert not g.is_termination_requested("r1")

    def test_noop_when_missing(self):
        g.cleanup_research("r1")  # Should not raise


# ===================================================================
# Thread safety
# ===================================================================


class TestThreadSafety:
    def test_concurrent_progress_updates(self):
        """Multiple threads updating progress should not lose data."""
        g.set_active_research("r1", {"progress": 0})
        barrier = threading.Barrier(10)

        def updater(value):
            barrier.wait()
            g.update_progress_if_higher("r1", value)

        threads = [
            threading.Thread(target=updater, args=(i * 10,)) for i in range(10)
        ]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        # Progress should be the maximum value
        assert g.get_research_field("r1", "progress") == 90

    def test_concurrent_set_and_remove(self):
        """Concurrent add/remove should not crash."""
        errors = []

        def worker(i):
            try:
                for _ in range(100):
                    rid = f"r-{i}"
                    g.set_active_research(rid, {"progress": 0})
                    g.is_research_active(rid)
                    g.remove_active_research(rid)
            except Exception as e:
                errors.append(e)

        threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert errors == []

    def test_concurrent_cleanup(self):
        """cleanup_research from multiple threads should not crash."""
        for i in range(20):
            g.set_active_research(f"r-{i}", {"progress": i})
            g.set_termination_flag(f"r-{i}")

        errors = []

        def cleaner(i):
            try:
                g.cleanup_research(f"r-{i}")
            except Exception as e:
                errors.append(e)

        threads = [
            threading.Thread(target=cleaner, args=(i,)) for i in range(20)
        ]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert errors == []
        assert g.get_active_research_count() == 0