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