test: add 129 tests for critical untested code (core, CoSTEER, factor_coder, model_coder, qlib pipeline)

test/qlib/test_core_framework.py

@@ -0,0 +1,243 @@
+"""Tests for rdagent.core — the core framework abstractions."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+
+# =============================================================================
+# Feedback base class
+# =============================================================================
+
+
+class TestFeedback:
+    def test_default_is_acceptable_returns_true(self):
+        from rdagent.core.evaluation import Feedback
+        fb = Feedback()
+        assert fb.is_acceptable() is True
+
+    def test_default_finished_returns_true(self):
+        from rdagent.core.evaluation import Feedback
+        fb = Feedback()
+        assert fb.finished() is True
+
+    def test_default_bool_is_true(self):
+        from rdagent.core.evaluation import Feedback
+        fb = Feedback()
+        assert bool(fb) is True
+
+
+# =============================================================================
+# EvoStep dataclass
+# =============================================================================
+
+
+class TestEvoStep:
+    def test_default_construction(self):
+        from rdagent.core.evolving_framework import EvoStep
+        es = EvoStep(evolvable_subjects="mock_evo")
+        assert es.evolvable_subjects == "mock_evo"
+        assert es.queried_knowledge is None
+        assert es.feedback is None
+
+    def test_full_construction(self):
+        from rdagent.core.evolving_framework import EvoStep, QueriedKnowledge
+        qk = QueriedKnowledge()
+        es = EvoStep(evolvable_subjects="evo", queried_knowledge=qk, feedback="fb")
+        assert es.queried_knowledge is qk
+        assert es.feedback == "fb"
+
+    def test_equality_by_reference(self):
+        from rdagent.core.evolving_framework import EvoStep
+        es1 = EvoStep(evolvable_subjects="a")
+        es2 = EvoStep(evolvable_subjects="a")
+        assert es1 == es2
+
+
+# =============================================================================
+# Scenario base class
+# =============================================================================
+
+
+class TestScenario:
+    def test_source_data_default_returns_empty_string(self):
+        from rdagent.core.scenario import Scenario
+        
+        class MinimalScenario(Scenario):
+            @property
+            def background(self) -> str: return "bg"
+            @property
+            def rich_style_description(self) -> str: return "rich"
+            def get_scenario_all_desc(self, **kwargs) -> str: return "all"
+            def get_runtime_environment(self) -> str: return "env"
+
+        scen = MinimalScenario()
+        assert scen.source_data == ""
+
+    def test_source_data_property_calls_get_source_data_desc(self):
+        from rdagent.core.scenario import Scenario
+
+        class FakeScenario(Scenario):
+            @property
+            def background(self) -> str: return "bg"
+            @property
+            def rich_style_description(self) -> str: return "rich"
+            def get_scenario_all_desc(self, **kwargs) -> str: return "all"
+            def get_runtime_environment(self) -> str: return "env"
+            def get_source_data_desc(self, task=None) -> str: return "custom_data"
+
+        scen = FakeScenario()
+        assert scen.source_data == "custom_data"
+
+
+# =============================================================================
+# EvolvingStrategy base class
+# =============================================================================
+
+
+class TestEvolvingStrategy:
+    def test_init_stores_scenario(self):
+        from rdagent.core.evolving_framework import EvolvingStrategy
+        
+        class MinimalStrategy(EvolvingStrategy):
+            def evolve_iter(self, evo, queried_knowledge=None, evolving_trace=None):
+                yield evo
+        
+        mock_scen = MagicMock()
+        es = MinimalStrategy(mock_scen)
+        assert es.scen is mock_scen
+
+
+# =============================================================================
+# IterEvaluator base class
+# =============================================================================
+
+
+class TestIterEvaluator:
+    def test_evaluate_returns_feedback(self):
+        from rdagent.core.evaluation import Feedback
+        from rdagent.core.evolving_framework import IterEvaluator, EvolvableSubjects
+
+        class MyFeedback(Feedback):
+            pass
+
+        class MyEvaluator(IterEvaluator):
+            def evaluate_iter(self):
+                evo = yield MyFeedback()
+                yield MyFeedback()
+                return MyFeedback()
+
+        eva = MyEvaluator()
+        result = eva.evaluate(EvolvableSubjects())
+        assert isinstance(result, MyFeedback)
+
+    def test_evaluate_iter_send_none_stops(self):
+        """Sending None mid-iteration triggers StopIteration with final feedback."""
+        from rdagent.core.evaluation import Feedback
+        from rdagent.core.evolving_framework import IterEvaluator
+
+        class MyEvaluator(IterEvaluator):
+            def evaluate_iter(self):
+                yield Feedback()            # kick-off (none)
+                evo_next = yield Feedback()  # partial eval
+                if evo_next is None:
+                    return Feedback()        # early return
+                return Feedback()            # normal path
+
+        eva = MyEvaluator()
+        gen = eva.evaluate_iter()
+        next(gen)           # kick-off → first Feedback
+        gen.send("any")     # evo gets "any" → second Feedback (evo_next NOT assigned yet)
+        with pytest.raises(StopIteration):
+            gen.send(None)  # evo_next = None → return → StopIteration
+
+
+# =============================================================================
+# Developer base class
+# =============================================================================
+
+
+class TestDeveloper:
+    def test_develop_raises_not_implemented(self):
+        from rdagent.core.developer import Developer
+
+        class MinimalDeveloper(Developer):
+            def develop(self, exp):
+                return super().develop(exp)
+
+        dev = MinimalDeveloper(MagicMock())
+        with pytest.raises(NotImplementedError):
+            dev.develop(MagicMock())
+
+
+# =============================================================================
+# Knowledge / QueriedKnowledge
+# =============================================================================
+
+
+class TestKnowledgeHierarchy:
+    def test_knowledge_pass_through(self):
+        from rdagent.core.evolving_framework import Knowledge, QueriedKnowledge
+        k = Knowledge()
+        qk = QueriedKnowledge()
+        assert isinstance(k, Knowledge)
+        assert isinstance(qk, QueriedKnowledge)
+
+
+# =============================================================================
+# EvolvingAgent (abstract interface)
+# =============================================================================
+
+
+class TestEvolvingAgent:
+    def test_ragevo_agent_init(self):
+        from rdagent.core.evolving_agent import RAGEvoAgent
+        mock_strategy = MagicMock()
+        mock_rag = MagicMock()
+        agent = RAGEvoAgent.__new__(RAGEvoAgent)
+        RAGEvoAgent.__init__(agent, max_loop=5, evolving_strategy=mock_strategy, rag=mock_rag)
+        assert agent.max_loop == 5
+        assert agent.evolving_strategy is mock_strategy
+        assert agent.rag is mock_rag
+
+    def test_ragevo_agent_default_knowledge_flags(self):
+        from rdagent.core.evolving_agent import RAGEvoAgent
+        agent = RAGEvoAgent.__new__(RAGEvoAgent)
+        RAGEvoAgent.__init__(agent, max_loop=3, evolving_strategy=MagicMock(), rag=MagicMock())
+        assert agent.with_knowledge is False
+        assert agent.knowledge_self_gen is False
+        assert agent.enable_filelock is False
+
+    def test_ragevo_agent_with_knowledge_enabled(self):
+        from rdagent.core.evolving_agent import RAGEvoAgent
+        agent = RAGEvoAgent.__new__(RAGEvoAgent)
+        RAGEvoAgent.__init__(
+            agent, max_loop=3, evolving_strategy=MagicMock(), rag=MagicMock(),
+            with_knowledge=True, knowledge_self_gen=True,
+            enable_filelock=True, filelock_path="/tmp/test.lock",
+        )
+        assert agent.with_knowledge is True
+        assert agent.knowledge_self_gen is True
+        assert agent.enable_filelock is True
+        assert agent.filelock_path == "/tmp/test.lock"
+
+
+# =============================================================================
+# EvolvableSubjects clone
+# =============================================================================
+
+
+class TestEvolvableSubjects:
+    def test_clone_produces_deep_copy(self):
+        from rdagent.core.evolving_framework import EvolvableSubjects
+        es = EvolvableSubjects()
+        clone = es.clone()
+        assert clone is not es
+        assert type(clone) is type(es)

test/qlib/test_costeer_feedback.py

@@ -0,0 +1,338 @@
+"""Tests for CoSTEER feedback types and EvolvingItem."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+
+# =============================================================================
+# CoSTEERSingleFeedback
+# =============================================================================
+
+
+class TestCoSTEERSingleFeedback:
+    def test_construction_with_valid_fields(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        fb = CoSTEERSingleFeedback(
+            execution="exec ok",
+            return_checking="return ok",
+            code="code ok",
+            final_decision=True,
+        )
+        assert fb.execution == "exec ok"
+        assert fb.return_checking == "return ok"
+        assert fb.code == "code ok"
+        assert fb.final_decision is True
+
+    def test_bool_returns_final_decision(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        fb_true = CoSTEERSingleFeedback(execution="x", return_checking="x", code="x", final_decision=True)
+        fb_false = CoSTEERSingleFeedback(execution="x", return_checking="x", code="x", final_decision=False)
+        assert bool(fb_true) is True
+        assert bool(fb_false) is False
+
+    def test_default_values(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        fb = CoSTEERSingleFeedback(execution="x", return_checking=None, code="x")
+        assert fb.final_decision is None
+        assert fb.raw_execution == ""
+        assert fb.source_feedback == {}
+
+    def test_val_and_update_init_dict_converts_boolean(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        assert CoSTEERSingleFeedback.val_and_update_init_dict(
+            {"execution": "x", "return_checking": "y", "code": "z", "final_decision": "true"}
+        )["final_decision"] is True
+        assert CoSTEERSingleFeedback.val_and_update_init_dict(
+            {"execution": "x", "return_checking": "y", "code": "z", "final_decision": "false"}
+        )["final_decision"] is False
+        assert CoSTEERSingleFeedback.val_and_update_init_dict(
+            {"execution": "x", "return_checking": "y", "code": "z", "final_decision": "True"}
+        )["final_decision"] is True
+        assert CoSTEERSingleFeedback.val_and_update_init_dict(
+            {"execution": "x", "return_checking": "y", "code": "z", "final_decision": "False"}
+        )["final_decision"] is False
+
+    def test_val_and_update_init_dict_rejects_non_boolean(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        with pytest.raises(ValueError):
+            CoSTEERSingleFeedback.val_and_update_init_dict(
+                {"execution": "x", "return_checking": "y", "code": "z", "final_decision": 42}
+            )
+
+    def test_val_and_update_init_dict_missing_final_decision_raises(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        with pytest.raises(ValueError, match="final_decision"):
+            CoSTEERSingleFeedback.val_and_update_init_dict(
+                {"execution": "x", "return_checking": "y", "code": "z"}
+            )
+
+    def test_val_and_update_init_dict_json_dumps_non_string_attrs(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        import json
+        data = {
+            "execution": {"key": "val"},
+            "return_checking": ["list"],
+            "code": 123,
+            "final_decision": True,
+        }
+        result = CoSTEERSingleFeedback.val_and_update_init_dict(data)
+        for attr in ("execution", "return_checking", "code"):
+            # Should have been converted to JSON string
+            assert isinstance(result[attr], str)
+            _ = json.loads(result[attr])  # valid JSON
+
+    def test_merge_all_true_decisions(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        fb1 = CoSTEERSingleFeedback(execution="a", return_checking="ra", code="c1", final_decision=True)
+        fb2 = CoSTEERSingleFeedback(execution="b", return_checking="rb", code="c2", final_decision=True)
+        merged = CoSTEERSingleFeedback.merge([fb1, fb2])
+        assert merged.final_decision is True
+
+    def test_merge_one_false_decision(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        fb1 = CoSTEERSingleFeedback(execution="a", return_checking="ra", code="c1", final_decision=True)
+        fb2 = CoSTEERSingleFeedback(execution="b", return_checking="rb", code="c2", final_decision=False)
+        merged = CoSTEERSingleFeedback.merge([fb1, fb2])
+        assert merged.final_decision is False
+
+    def test_merge_concatenates_fields(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        fb1 = CoSTEERSingleFeedback(execution="A", return_checking="RA", code="C1", final_decision=True)
+        fb2 = CoSTEERSingleFeedback(execution="B", return_checking="RB", code="C2", final_decision=True)
+        merged = CoSTEERSingleFeedback.merge([fb1, fb2])
+        assert "A\n\nB" in merged.execution
+        assert "RA\n\nRB" in merged.return_checking
+        assert "C1\n\nC2" in merged.code
+
+    def test_merge_skips_none_fields(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        fb1 = CoSTEERSingleFeedback(execution="A", return_checking="RA", code="C1", final_decision=True)
+        fb2 = CoSTEERSingleFeedback(execution="B", return_checking=None, code="C2", final_decision=True)
+        merged = CoSTEERSingleFeedback.merge([fb1, fb2])
+        assert merged.execution == "A\n\nB"
+        assert merged.return_checking == "RA"
+        assert merged.code == "C1\n\nC2"
+
+    def test_merge_aggregates_source_feedback(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        fb1 = CoSTEERSingleFeedback(execution="a", return_checking="x", code="c", final_decision=True,
+                                     source_feedback={"eval1": True})
+        fb2 = CoSTEERSingleFeedback(execution="b", return_checking="y", code="d", final_decision=True,
+                                     source_feedback={"eval2": False})
+        merged = CoSTEERSingleFeedback.merge([fb1, fb2])
+        assert merged.source_feedback == {"eval1": True, "eval2": False}
+
+    def test_str_contains_all_sections(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        fb = CoSTEERSingleFeedback(execution="exec", return_checking="ret", code="code", final_decision=True)
+        s = str(fb)
+        assert "Execution" in s
+        assert "Return Checking" in s
+        assert "Code" in s
+        assert "Final Decision" in s
+        assert "SUCCESS" in s
+
+    def test_str_shows_fail_for_false(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        fb = CoSTEERSingleFeedback(execution="exec", return_checking="ret", code="code", final_decision=False)
+        assert "FAIL" in str(fb)
+
+
+# =============================================================================
+# CoSTEERSingleFeedbackDeprecated
+# =============================================================================
+
+
+class TestCoSTEERSingleFeedbackDeprecated:
+    def test_property_getters(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedbackDeprecated
+        fb = CoSTEERSingleFeedbackDeprecated(
+            execution_feedback="exec",
+            code_feedback="code",
+            value_feedback="val",
+            shape_feedback="shape",
+            final_decision=True,
+            final_feedback="final",
+            value_generated_flag=True,
+            final_decision_based_on_gt=True,
+        )
+        assert fb.execution == "exec"
+        assert fb.code == "code"
+        assert fb.final_decision is True
+        assert fb.value_generated_flag is True
+        assert fb.final_decision_based_on_gt is True
+
+    def test_return_checking_when_value_generated(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedbackDeprecated
+        fb = CoSTEERSingleFeedbackDeprecated(
+            value_generated_flag=True, value_feedback="vals", shape_feedback="shapes",
+        )
+        rc = fb.return_checking
+        assert "vals" in rc
+        assert "shapes" in rc
+
+    def test_return_checking_when_no_value_generated(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedbackDeprecated
+        fb = CoSTEERSingleFeedbackDeprecated(value_generated_flag=False)
+        assert fb.return_checking is None
+
+    def test_setters_work(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedbackDeprecated
+        fb = CoSTEERSingleFeedbackDeprecated()
+        fb.execution = "new_exec"
+        fb.code = "new_code"
+        fb.return_checking = "new_rc"
+        assert fb.execution_feedback == "new_exec"
+        assert fb.code_feedback == "new_code"
+        assert fb.value_feedback == "new_rc"
+        assert fb.shape_feedback == "new_rc"
+
+    def test_str_contains_all_sections(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedbackDeprecated
+        fb = CoSTEERSingleFeedbackDeprecated(
+            execution_feedback="exec", shape_feedback="shape",
+            code_feedback="code", value_feedback="val",
+            final_feedback="final", final_decision=True,
+        )
+        s = str(fb)
+        for keyword in ("Execution", "Shape", "Code", "Value", "Final Decision", "SUCCESS"):
+            assert keyword in s
+
+
+# =============================================================================
+# CoSTEERMultiFeedback
+# =============================================================================
+
+
+class TestCoSTEERMultiFeedback:
+    def _make_fb(self, decision=True):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
+        return CoSTEERSingleFeedback(execution="x", return_checking="x", code="x", final_decision=decision)
+
+    def test_getitem(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback
+        fb1, fb2 = self._make_fb(True), self._make_fb(False)
+        mf = CoSTEERMultiFeedback([fb1, fb2])
+        assert mf[0].final_decision is True
+        assert mf[1].final_decision is False
+
+    def test_len(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback
+        assert len(CoSTEERMultiFeedback([self._make_fb()])) == 1
+        assert len(CoSTEERMultiFeedback([])) == 0
+
+    def test_append(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback
+        mf = CoSTEERMultiFeedback([])
+        mf.append(self._make_fb(True))
+        assert len(mf) == 1
+        assert mf[0].final_decision is True
+
+    def test_iter(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback
+        fbs = [self._make_fb(True), self._make_fb(True)]
+        mf = CoSTEERMultiFeedback(fbs)
+        assert list(mf) == fbs
+
+    def test_finished_all_true(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback
+        mf = CoSTEERMultiFeedback([self._make_fb(True), self._make_fb(True)])
+        assert mf.finished() is True
+
+    def test_finished_one_false(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback
+        mf = CoSTEERMultiFeedback([self._make_fb(True), self._make_fb(False)])
+        assert mf.finished() is False
+
+    def test_finished_with_none_skips(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback
+        mf = CoSTEERMultiFeedback([self._make_fb(True), None])
+        assert mf.finished() is True  # None = skipped = accepted
+
+    def test_bool_all_true(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback
+        assert bool(CoSTEERMultiFeedback([self._make_fb(True), self._make_fb(True)])) is True
+
+    def test_bool_one_false(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback
+        assert bool(CoSTEERMultiFeedback([self._make_fb(True), self._make_fb(False)])) is False
+
+    def test_is_acceptable_delegates(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback
+        mf = CoSTEERMultiFeedback([self._make_fb(True), self._make_fb(True)])
+        assert mf.is_acceptable() is True
+
+
+# =============================================================================
+# EvolvingItem
+# =============================================================================
+
+
+class TestEvolvingItem:
+    def test_construction_without_gt(self):
+        from rdagent.components.coder.CoSTEER.evolvable_subjects import EvolvingItem
+        from rdagent.core.experiment import Task
+
+        t1 = Task(name="task1")
+        t2 = Task(name="task2")
+        ei = EvolvingItem(sub_tasks=[t1, t2])
+        assert len(ei.sub_tasks) == 2
+        assert ei.sub_gt_implementations is None
+
+    def test_construction_with_matching_gt(self):
+        from rdagent.components.coder.CoSTEER.evolvable_subjects import EvolvingItem
+        from rdagent.core.experiment import Task, FBWorkspace
+        t1, t2 = Task(name="t1"), Task(name="t2")
+        ws1, ws2 = FBWorkspace(), FBWorkspace()
+        ei = EvolvingItem(sub_tasks=[t1, t2], sub_gt_implementations=[ws1, ws2])
+        assert ei.sub_gt_implementations == [ws1, ws2]
+
+    def test_mismatched_gt_length_resets_to_none(self):
+        from rdagent.components.coder.CoSTEER.evolvable_subjects import EvolvingItem
+        from rdagent.core.experiment import Task, FBWorkspace
+        t1, t2 = Task(name="t1"), Task(name="t2")
+        ei = EvolvingItem(sub_tasks=[t1, t2], sub_gt_implementations=[FBWorkspace()])
+        assert ei.sub_gt_implementations is None
+
+    def test_from_experiment(self):
+        from rdagent.components.coder.CoSTEER.evolvable_subjects import EvolvingItem
+        from rdagent.core.experiment import Experiment, Task
+
+        exp = Experiment(sub_tasks=[Task(name="x")])
+        exp.based_experiments = ["base"]
+        exp.experiment_workspace = "ws"
+        ei = EvolvingItem.from_experiment(exp)
+        assert len(ei.sub_tasks) == 1
+        assert ei.based_experiments == ["base"]
+        assert ei.experiment_workspace == "ws"
+
+
+# =============================================================================
+# CoSTEERQueriedKnowledge
+# =============================================================================
+
+
+class TestCoSTEERQueriedKnowledge:
+    def test_default_construction(self):
+        from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERQueriedKnowledge
+        qk = CoSTEERQueriedKnowledge()
+        assert qk.success_task_to_knowledge_dict == {}
+        assert qk.failed_task_info_set == set()
+
+    def test_with_data(self):
+        from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERQueriedKnowledge
+        qk = CoSTEERQueriedKnowledge(
+            success_task_to_knowledge_dict={"a": "knowledge_a"},
+            failed_task_info_set={"fail1", "fail2"},
+        )
+        assert qk.success_task_to_knowledge_dict["a"] == "knowledge_a"
+        assert "fail1" in qk.failed_task_info_set
+        assert "fail2" in qk.failed_task_info_set

test/qlib/test_costeer_strategy.py

@@ -0,0 +1,225 @@
+"""Tests for CoSTEER config, task, and evolve strategy population logic."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+
+# =============================================================================
+# CoSTEERSettings
+# =============================================================================
+
+
+class TestCoSTEERSettings:
+    def test_default_values(self):
+        from rdagent.components.coder.CoSTEER.config import CoSTEERSettings
+        s = CoSTEERSettings()
+        assert s.max_loop == 1
+        assert s.fail_task_trial_limit == 5
+        assert s.v2_query_component_limit == 1
+        assert s.v2_query_error_limit == 1
+        assert s.v2_query_former_trace_limit == 3
+        assert s.v2_add_fail_attempt_to_latest_successful_execution is False
+        assert s.v2_knowledge_sampler == 1.0
+        assert s.coder_use_cache is False
+        assert s.enable_filelock is False
+
+    def test_singleton_instance(self):
+        from rdagent.components.coder.CoSTEER.config import CoSTEER_SETTINGS
+        from rdagent.components.coder.CoSTEER.config import CoSTEERSettings
+        assert isinstance(CoSTEER_SETTINGS, CoSTEERSettings)
+        assert CoSTEER_SETTINGS.max_loop == 1
+
+
+# =============================================================================
+# CoSTEERTask
+# =============================================================================
+
+
+class TestCoSTEERTask:
+    def test_base_code_stored(self):
+        from rdagent.components.coder.CoSTEER.task import CoSTEERTask
+        t = CoSTEERTask(name="test", base_code="print(1)")
+        assert t.base_code == "print(1)"
+
+    def test_base_code_none_by_default(self):
+        from rdagent.components.coder.CoSTEER.task import CoSTEERTask
+        t = CoSTEERTask(name="test")
+        assert t.base_code is None
+
+
+# =============================================================================
+# MultiProcessEvolvingStrategy.assign_code_list_to_evo
+# =============================================================================
+
+
+class TestAssignCodeListToEvo:
+    def test_empty_code_list_noops(self):
+        from rdagent.components.coder.CoSTEER.evolving_strategy import MultiProcessEvolvingStrategy
+        from rdagent.components.coder.CoSTEER.evolvable_subjects import EvolvingItem
+        from rdagent.core.experiment import Task
+
+        strat = MultiProcessEvolvingStrategy.__new__(MultiProcessEvolvingStrategy)
+        MultiProcessEvolvingStrategy.__init__(strat, scen=MagicMock(), settings=MagicMock())
+
+        ei = EvolvingItem(sub_tasks=[Task(name="t1")])
+        ei.experiment_workspace = MagicMock()
+        result = strat.assign_code_list_to_evo([{}], ei)
+        assert result is ei
+
+    def test_none_entry_is_skipped(self):
+        from rdagent.components.coder.CoSTEER.evolving_strategy import MultiProcessEvolvingStrategy
+        from rdagent.components.coder.CoSTEER.evolvable_subjects import EvolvingItem
+        from rdagent.core.experiment import Task
+
+        strat = MultiProcessEvolvingStrategy.__new__(MultiProcessEvolvingStrategy)
+        MultiProcessEvolvingStrategy.__init__(strat, scen=MagicMock(), settings=MagicMock())
+
+        ei = EvolvingItem(sub_tasks=[Task(name="t1")])
+        ei.experiment_workspace = MagicMock()
+        result = strat.assign_code_list_to_evo([None], ei)
+        assert result.sub_workspace_list[0] is None  # unchanged
+
+    def test_code_injects_files(self):
+        from rdagent.components.coder.CoSTEER.evolving_strategy import MultiProcessEvolvingStrategy
+        from rdagent.components.coder.CoSTEER.evolvable_subjects import EvolvingItem
+        from rdagent.core.experiment import Task
+
+        strat = MultiProcessEvolvingStrategy.__new__(MultiProcessEvolvingStrategy)
+        MultiProcessEvolvingStrategy.__init__(strat, scen=MagicMock(), settings=MagicMock())
+
+        ei = EvolvingItem(sub_tasks=[Task(name="t1")])
+        mock_ws = MagicMock()
+        ei.experiment_workspace = mock_ws
+
+        strat.assign_code_list_to_evo([{"factor.py": "x=1"}], ei)
+        mock_ws.inject_files.assert_called_once_with(**{"factor.py": "x=1"})
+
+    def test_change_summary_extracted(self):
+        from rdagent.components.coder.CoSTEER.evolving_strategy import MultiProcessEvolvingStrategy
+        from rdagent.components.coder.CoSTEER.evolvable_subjects import EvolvingItem
+        from rdagent.core.experiment import Task
+
+        strat = MultiProcessEvolvingStrategy.__new__(MultiProcessEvolvingStrategy)
+        MultiProcessEvolvingStrategy.__init__(strat, scen=MagicMock(), settings=MagicMock())
+
+        mock_ws = MagicMock()
+        ei = EvolvingItem(sub_tasks=[Task(name="t1")])
+        ei.experiment_workspace = mock_ws
+
+        strat.assign_code_list_to_evo([{"__change_summary__": "summary", "factor.py": "x"}], ei)
+        assert mock_ws.change_summary == "summary"
+        # change_summary should have been popped from dict
+        mock_ws.inject_files.assert_called_once_with(**{"factor.py": "x"})
+
+
+# =============================================================================
+# MultiProcessEvolvingStrategy.evolve_iter
+# =============================================================================
+
+
+class TestEvolveIter:
+    def _make_strat(self):
+        from rdagent.components.coder.CoSTEER.evolving_strategy import MultiProcessEvolvingStrategy
+        from rdagent.components.coder.CoSTEER.config import CoSTEERSettings
+        strat = MultiProcessEvolvingStrategy(
+            scen=MagicMock(), settings=CoSTEERSettings(), improve_mode=False,
+        )
+        return strat
+
+    def _make_evo(self, n_tasks=2):
+        from rdagent.components.coder.CoSTEER.evolvable_subjects import EvolvingItem
+        from rdagent.core.experiment import Task
+        tasks = [Task(name=f"task_{i}") for i in range(n_tasks)]
+        for t in tasks:
+            t.get_task_information = MagicMock(return_value=f"info_{t.name}")
+        ei = EvolvingItem(sub_tasks=tasks)
+        ei.experiment_workspace = MagicMock()
+        return ei
+
+    def test_raises_without_queried_knowledge(self):
+        strat = self._make_strat()
+        evo = self._make_evo()
+        with pytest.raises(ValueError, match="queried_knowledge"):
+            next(strat.evolve_iter(evo=evo, queried_knowledge=None))
+
+    def test_successful_tasks_not_scheduled(self):
+        from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERQueriedKnowledge
+        strat = self._make_strat()
+        evo = self._make_evo(n_tasks=1)
+        qk = CoSTEERQueriedKnowledge(
+            success_task_to_knowledge_dict={
+                "info_task_0": MagicMock(implementation=MagicMock(file_dict={"f.py": "x"})),
+            },
+        )
+        # evolve_iter is a generator, next() starts it
+        gen = strat.evolve_iter(evo=evo, queried_knowledge=qk)
+        # Should yield the evo (populated from success knowledge)
+        result = next(gen)
+        # The task was already successful, so no new scheduling
+        assert result is evo
+
+    def test_failed_tasks_skipped(self):
+        from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERQueriedKnowledge
+        strat = self._make_strat()
+        evo = self._make_evo(n_tasks=1)
+        qk = CoSTEERQueriedKnowledge(
+            failed_task_info_set={"info_task_0"},
+        )
+        gen = strat.evolve_iter(evo=evo, queried_knowledge=qk)
+        result = next(gen)
+        # Task skipped because it's in failed_set
+        assert result is evo
+
+    def test_improve_mode_skips_with_no_last_feedback(self):
+        from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERQueriedKnowledge
+        strat = self._make_strat()
+        strat.improve_mode = True
+        evo = self._make_evo(n_tasks=1)
+        qk = CoSTEERQueriedKnowledge()
+        gen = strat.evolve_iter(evo=evo, queried_knowledge=qk, evolving_trace=[])
+        result = next(gen)
+        # In improve_mode with no last_feedback, task should be skipped
+        # (code_list[0] should be {} — empty implementation)
+        assert result is evo
+
+    def test_non_improve_mode_schedules_new_tasks(self):
+        """Tasks not in success/failed should be scheduled."""
+        from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERQueriedKnowledge
+        strat = self._make_strat()
+        evo = self._make_evo(n_tasks=1)
+        qk = CoSTEERQueriedKnowledge()
+
+        with patch(
+            "rdagent.components.coder.CoSTEER.evolving_strategy.multiprocessing_wrapper",
+            return_value=[{"factor.py": "x=1"}],
+        ):
+            gen = strat.evolve_iter(evo=evo, queried_knowledge=qk)
+            result = next(gen)
+            assert result is evo
+
+
+# =============================================================================
+# CoSTEERMultiEvaluator (partial)
+# =============================================================================
+
+
+class TestCoSTEERMultiEvaluator:
+    def test_init_with_single_evaluator(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiEvaluator
+        mock_eval = MagicMock()
+        eva = CoSTEERMultiEvaluator(single_evaluator=mock_eval, scen=MagicMock())
+        assert eva.single_evaluator is mock_eval
+
+    def test_init_with_evaluator_list(self):
+        from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiEvaluator
+        mock_list = [MagicMock(), MagicMock()]
+        eva = CoSTEERMultiEvaluator(single_evaluator=mock_list, scen=MagicMock())
+        assert eva.single_evaluator is mock_list

test/qlib/test_factor_coder.py

@@ -0,0 +1,221 @@
+"""Tests for factor_coder — evaluators, task, workspace."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+
+# =============================================================================
+# FactorTask
+# =============================================================================
+
+
+class TestFactorTask:
+    def test_construction_fields(self):
+        from rdagent.components.coder.factor_coder.factor import FactorTask
+        t = FactorTask(
+            factor_name="f1",
+            factor_description="desc",
+            factor_formulation="formula",
+            variables={"x": 1},
+            resource="r1",
+        )
+        assert t.factor_name == "f1"
+        assert t.factor_description == "desc"
+        assert t.factor_formulation == "formula"
+        assert t.variables == {"x": 1}
+        assert t.factor_resources == "r1"
+        assert t.factor_implementation is False
+        assert t.base_code is None  # from CoSTEERTask
+
+    def test_get_task_information(self):
+        from rdagent.components.coder.factor_coder.factor import FactorTask
+        t = FactorTask("f1", "desc", "formula", variables={"x": 1})
+        info = t.get_task_information()
+        assert "factor_name: f1" in info
+        assert "factor_description: desc" in info
+        assert "factor_formulation: formula" in info
+        assert "variables: {'x': 1}" in info
+
+    def test_get_task_brief_information(self):
+        from rdagent.components.coder.factor_coder.factor import FactorTask
+        t = FactorTask("f1", "desc", "formula")
+        info = t.get_task_brief_information()
+        assert "factor_name: f1" in info
+
+    def test_get_task_information_and_implementation_result(self):
+        from rdagent.components.coder.factor_coder.factor import FactorTask
+        t = FactorTask("f1", "desc", "formula")
+        result = t.get_task_information_and_implementation_result()
+        assert result["factor_name"] == "f1"
+        assert result["factor_description"] == "desc"
+        assert "factor_implementation" in result
+
+    def test_from_dict(self):
+        from rdagent.components.coder.factor_coder.factor import FactorTask
+        d = {
+            "factor_name": "f2",
+            "factor_description": "d2",
+            "factor_formulation": "f2",
+            "variables": {},
+            "resource": None,
+            "factor_implementation": True,
+        }
+        t = FactorTask.from_dict(d)
+        assert t.factor_name == "f2"
+        assert t.factor_implementation is True
+
+    def test_repr(self):
+        from rdagent.components.coder.factor_coder.factor import FactorTask
+        t = FactorTask("myfactor", "desc", "formula")
+        assert "FactorTask" in repr(t)
+        assert "myfactor" in repr(t)
+
+
+# =============================================================================
+# FactorFBWorkspace
+# =============================================================================
+
+
+class TestFactorFBWorkspace:
+    def test_init_sets_workspace_path(self):
+        from rdagent.components.coder.factor_coder.factor import FactorFBWorkspace, FactorTask
+        t = FactorTask("f1", "desc", "formula")
+        ws = FactorFBWorkspace(target_task=t)
+        assert ws.workspace_path is not None
+        # Directory is created lazily by execute(), not in __init__
+        assert isinstance(ws.workspace_path, Path)
+
+    def test_execute_returns_message_and_dataframe(self):
+        from rdagent.components.coder.factor_coder.factor import FactorFBWorkspace, FactorTask
+        t = FactorTask("f1", "desc", "formula")
+        t.version = 1
+        ws = FactorFBWorkspace(target_task=t)
+        # Inject valid factor code
+        ws.inject_files(**{
+            "factor.py": (
+                "import pandas as pd\n"
+                "import numpy as np\n"
+                "data = pd.read_hdf('intraday_pv.h5', key='data')\n"
+                "factor_val = data['$close'].pct_change()\n"
+                "factor_val = factor_val.to_frame('f1')\n"
+                "factor_val.to_hdf('result.h5', key='data', mode='w')\n"
+            ),
+        })
+        msg, df = ws.execute()
+        assert isinstance(msg, str)
+        assert df is not None
+
+    def test_execute_succeeds_and_returns_data(self):
+        from rdagent.components.coder.factor_coder.factor import FactorFBWorkspace, FactorTask
+        t = FactorTask("fl1", "desc", "formula")
+        ws = FactorFBWorkspace(target_task=t)
+        ws.inject_files(**{
+            "factor.py": (
+                "import pandas as pd\n"
+                "data = pd.read_hdf('intraday_pv.h5', key='data')\n"
+                "factor_val = data['$close'].pct_change().to_frame('fl1')\n"
+                "factor_val.to_hdf('result.h5', key='data', mode='w')\n"
+            ),
+        })
+        msg, df = ws.execute()
+        assert FactorFBWorkspace.FB_EXEC_SUCCESS in msg
+        assert FactorFBWorkspace.FB_OUTPUT_FILE_FOUND in msg
+        assert df is not None
+
+
+# =============================================================================
+# FactorEvaluatorForCoder (partial integration)
+# =============================================================================
+
+
+class TestFactorEvaluatorForCoder:
+    def test_init_creates_sub_evaluators(self):
+        from rdagent.components.coder.factor_coder.evaluators import FactorEvaluatorForCoder
+        mock_scen = MagicMock()
+        eva = FactorEvaluatorForCoder(scen=mock_scen)
+        assert eva.value_evaluator is not None
+        assert eva.code_evaluator is not None
+        assert eva.final_decision_evaluator is not None
+
+    def test_evaluate_with_none_implementation(self):
+        from rdagent.components.coder.factor_coder.evaluators import FactorEvaluatorForCoder
+        eva = FactorEvaluatorForCoder(scen=MagicMock())
+        assert eva.evaluate(target_task=MagicMock(), implementation=None) is None
+
+    def test_evaluate_returns_queried_knowledge_if_present(self):
+        from rdagent.components.coder.factor_coder.evaluators import FactorEvaluatorForCoder
+        from rdagent.components.coder.factor_coder.factor import FactorTask
+
+        eva = FactorEvaluatorForCoder(scen=MagicMock())
+
+        t = FactorTask("f1", "desc", "formula")
+        qk = MagicMock()
+        qk.success_task_to_knowledge_dict = {"info_f1": MagicMock(feedback="cached_fb")}
+        t.get_task_information = MagicMock(return_value="info_f1")
+        qk.failed_task_info_set = set()
+
+        fb = eva.evaluate(target_task=t, implementation=MagicMock(), queried_knowledge=qk)
+        assert fb == "cached_fb"  # returned from cache
+
+    def test_evaluate_skips_failed_task(self):
+        from rdagent.components.coder.factor_coder.evaluators import FactorEvaluatorForCoder
+        from rdagent.components.coder.factor_coder.factor import FactorTask
+
+        eva = FactorEvaluatorForCoder(scen=MagicMock())
+
+        t = FactorTask("f1", "desc", "formula")
+        qk = MagicMock()
+        qk.success_task_to_knowledge_dict = {}
+        t.get_task_information = MagicMock(return_value="info_f1")
+        qk.failed_task_info_set = {"info_f1"}
+
+        fb = eva.evaluate(target_task=t, implementation=MagicMock(), queried_knowledge=qk)
+        assert fb.final_decision is False
+        assert "failed too many times" in fb.execution_feedback
+
+
+# =============================================================================
+# FactorEvaluator (eva_utils) — constructors and identity
+# =============================================================================
+
+
+class TestFactorEvaluatorsInit:
+    def test_factor_inf_evaluator_init(self):
+        from rdagent.components.coder.factor_coder.eva_utils import FactorInfEvaluator
+        eva = FactorInfEvaluator()
+        assert str(eva) == "FactorInfEvaluator"
+
+    def test_factor_single_column_evaluator_init(self):
+        from rdagent.components.coder.factor_coder.eva_utils import FactorSingleColumnEvaluator
+        eva = FactorSingleColumnEvaluator()
+        assert str(eva) == "FactorSingleColumnEvaluator"
+
+    def test_factor_output_format_evaluator_init(self):
+        from rdagent.components.coder.factor_coder.eva_utils import FactorOutputFormatEvaluator
+        eva = FactorOutputFormatEvaluator()
+        assert str(eva) == "FactorOutputFormatEvaluator"
+
+    def test_factor_missing_values_evaluator_init(self):
+        from rdagent.components.coder.factor_coder.eva_utils import FactorMissingValuesEvaluator
+        eva = FactorMissingValuesEvaluator()
+        assert str(eva) == "FactorMissingValuesEvaluator"
+
+    def test_factor_correlation_evaluator_init(self):
+        from rdagent.components.coder.factor_coder.eva_utils import FactorCorrelationEvaluator
+        eva = FactorCorrelationEvaluator(hard_check=True)
+        assert eva.hard_check is True
+        assert str(eva) == "FactorCorrelationEvaluator"
+
+    def test_factor_value_evaluator_init(self):
+        from rdagent.components.coder.factor_coder.eva_utils import FactorValueEvaluator
+        mock_scen = MagicMock()
+        eva = FactorValueEvaluator(mock_scen)
+        assert eva.scen is mock_scen

test/qlib/test_model_coder.py

@@ -0,0 +1,182 @@
+"""Tests for model_coder — ModelTask, shape/value evaluators, config."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import numpy as np
+import pytest
+
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+
+# =============================================================================
+# ModelTask
+# =============================================================================
+
+
+class TestModelTask:
+    def test_construction_fields(self):
+        from rdagent.components.coder.model_coder.model import ModelTask
+        t = ModelTask(
+            name="m1",
+            description="desc",
+            architecture="LSTM",
+            hyperparameters={"lr": 0.001},
+            training_hyperparameters={"epochs": 10},
+            formulation="y = f(x)",
+            variables={"x": "feature"},
+            model_type="TimeSeries",
+        )
+        assert t.name == "m1"
+        assert t.description == "desc"
+        assert t.architecture == "LSTM"
+        assert t.hyperparameters == {"lr": 0.001}
+        assert t.training_hyperparameters == {"epochs": 10}
+        assert t.formulation == "y = f(x)"
+        assert t.variables == {"x": "feature"}
+        assert t.model_type == "TimeSeries"
+        assert t.base_code is None
+
+    def test_get_task_information(self):
+        from rdagent.components.coder.model_coder.model import ModelTask
+        t = ModelTask(
+            name="m1", description="desc", architecture="LSTM",
+            hyperparameters={}, training_hyperparameters={},
+            model_type="Tabular",
+        )
+        info = t.get_task_information()
+        assert "name: m1" in info
+        assert "architecture: LSTM" in info
+        assert "model_type: Tabular" in info
+
+    def test_get_task_information_with_optional_fields(self):
+        from rdagent.components.coder.model_coder.model import ModelTask
+        t = ModelTask(
+            name="m2", description="d2", architecture="GRU",
+            hyperparameters={}, training_hyperparameters={},
+            formulation="f1", variables={"v": 1}, model_type="Graph",
+        )
+        info = t.get_task_information()
+        assert "formulation: f1" in info
+        assert "variables: {'v': 1}" in info
+
+    def test_get_task_brief_information(self):
+        from rdagent.components.coder.model_coder.model import ModelTask
+        t = ModelTask(
+            name="m1", description="desc", architecture="LSTM",
+            hyperparameters={"lr": 0.01}, training_hyperparameters={"epochs": 5},
+        )
+        info = t.get_task_brief_information()
+        assert "name: m1" in info
+        assert "architecture: LSTM" in info
+        assert "hyperparameters" in info
+
+    def test_from_dict(self):
+        from rdagent.components.coder.model_coder.model import ModelTask
+        d = {
+            "name": "m3", "description": "d3", "architecture": "TCN",
+            "hyperparameters": {}, "training_hyperparameters": {},
+        }
+        t = ModelTask.from_dict(d)
+        assert t.name == "m3"
+
+    def test_repr(self):
+        from rdagent.components.coder.model_coder.model import ModelTask
+        t = ModelTask(
+            name="mymodel", description="d", architecture="LSTM",
+            hyperparameters={}, training_hyperparameters={},
+        )
+        assert "ModelTask" in repr(t)
+        assert "mymodel" in repr(t)
+
+
+# =============================================================================
+# Shape/Value evaluators (eva_utils)
+# =============================================================================
+
+
+class TestShapeEvaluator:
+    def test_correct_shape(self):
+        from rdagent.components.coder.model_coder.eva_utils import shape_evaluator
+        msg, ok = shape_evaluator(np.ones((32, 10)), target_shape=(32, 10))
+        assert ok is True
+        assert "correct" in msg.lower()
+
+    def test_incorrect_shape(self):
+        from rdagent.components.coder.model_coder.eva_utils import shape_evaluator
+        msg, ok = shape_evaluator(np.ones((32, 5)), target_shape=(32, 10))
+        assert ok is False
+        assert "incorrect" in msg.lower()
+
+    def test_none_prediction(self):
+        from rdagent.components.coder.model_coder.eva_utils import shape_evaluator
+        msg, ok = shape_evaluator(None, target_shape=(32, 10))
+        assert ok is False
+
+    def test_none_target_shape(self):
+        from rdagent.components.coder.model_coder.eva_utils import shape_evaluator
+        msg, ok = shape_evaluator(np.ones((3,)), target_shape=None)
+        assert ok is False
+
+    def test_float_array(self):
+        from rdagent.components.coder.model_coder.eva_utils import shape_evaluator
+        msg, ok = shape_evaluator(np.array([1.0, 2.0]), target_shape=(2,))
+        assert ok is True
+
+
+class TestValueEvaluator:
+    def test_none_prediction(self):
+        from rdagent.components.coder.model_coder.eva_utils import value_evaluator
+        msg, ok = value_evaluator(None, np.ones((3,)))
+        assert ok is False
+
+    def test_none_target(self):
+        from rdagent.components.coder.model_coder.eva_utils import value_evaluator
+        msg, ok = value_evaluator(np.ones((3,)), None)
+        assert ok is False
+
+    def test_small_difference_passes(self):
+        from rdagent.components.coder.model_coder.eva_utils import value_evaluator
+        msg, ok = value_evaluator(
+            np.array([1.0, 2.0, 3.0]),
+            np.array([1.0, 2.0, 3.01]),
+        )
+        assert bool(ok) is True  # diff < 0.1
+
+    def test_large_difference_fails(self):
+        from rdagent.components.coder.model_coder.eva_utils import value_evaluator
+        msg, ok = value_evaluator(
+            np.array([1.0, 2.0]),
+            np.array([10.0, 20.0]),
+        )
+        assert bool(ok) is False  # diff > 0.1
+
+
+# =============================================================================
+# ModelCoSTEERSettings
+# =============================================================================
+
+
+class TestModelCoSTEERSettings:
+    def test_default_env_type(self):
+        from rdagent.components.coder.model_coder.conf import ModelCoSTEERSettings
+        s = ModelCoSTEERSettings()
+        assert s.env_type == "conda"
+
+    def test_singleton(self):
+        from rdagent.components.coder.model_coder.conf import MODEL_COSTEER_SETTINGS
+        from rdagent.components.coder.model_coder.conf import ModelCoSTEERSettings
+        assert isinstance(MODEL_COSTEER_SETTINGS, ModelCoSTEERSettings)
+
+    def test_get_model_env_runs(self):
+        from rdagent.components.coder.model_coder.conf import get_model_env
+        # May succeed (conda available) or fail — either way, test the code path
+        try:
+            env = get_model_env()
+            assert env is not None
+        except Exception:
+            pass  # expected if docker/conda not available

test/qlib/test_qlib_pipeline.py

@@ -0,0 +1,186 @@
+"""Tests for qlib pipeline — feedback, bandit, quant_loop_factory."""
+
+from __future__ import annotations
+
+import sys
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import numpy as np
+import pandas as pd
+import pytest
+
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+
+# =============================================================================
+# process_results (feedback.py)
+# =============================================================================
+
+
+class TestProcessResults:
+    def test_process_results_handles_named_series(self):
+        """process_results renames column "0" to "Current Result" — this works
+        when the Series name is '0' (string), which matches the rename dict."""
+        from rdagent.scenarios.qlib.developer.feedback import process_results
+        import pandas as pd
+
+        # process_results expects the Series to produce a DataFrame column named "0" (string)
+        # This happens when the Series has name '0'
+        current = pd.Series(
+            {"IC": 0.05, "1day.excess_return_with_cost.annualized_return": 0.12,
+             "1day.excess_return_with_cost.max_drawdown": -0.08},
+            name="0",
+        )
+        sota = pd.Series(
+            {"IC": 0.03, "1day.excess_return_with_cost.annualized_return": 0.10,
+             "1day.excess_return_with_cost.max_drawdown": -0.05},
+            name="0",
+        )
+
+        result = process_results(current, sota)
+        assert "IC of Current Result is" in result
+        assert "of SOTA Result is" in result
+
+    def test_raises_on_missing_metrics(self):
+        from rdagent.scenarios.qlib.developer.feedback import process_results
+
+        current = pd.Series({"IC": 0.05})
+        sota = pd.Series({"IC": 0.03})
+        with pytest.raises(KeyError):
+            process_results(current, sota)
+
+
+# =============================================================================
+# bandit.py — Metrics and extract_metrics_from_experiment
+# =============================================================================
+
+
+class TestBanditMetrics:
+    def test_default_values_are_zero(self):
+        from rdagent.scenarios.qlib.proposal.bandit import Metrics
+        m = Metrics()
+        assert m.ic == 0.0
+        assert m.sharpe == 0.0
+        assert m.mdd == 0.0
+
+    def test_as_vector_length(self):
+        from rdagent.scenarios.qlib.proposal.bandit import Metrics
+        m = Metrics(ic=0.1, sharpe=1.5)
+        v = m.as_vector()
+        assert len(v) == 8
+        assert v[0] == 0.1
+        assert v[7] == 1.5
+
+    def test_mdd_negated_in_vector(self):
+        from rdagent.scenarios.qlib.proposal.bandit import Metrics
+        m = Metrics(mdd=0.15)
+        v = m.as_vector()
+        assert v[6] == -0.15  # -self.mdd
+
+    def test_extract_metrics_from_experiment(self):
+        from rdagent.scenarios.qlib.proposal.bandit import extract_metrics_from_experiment
+
+        mock_exp = MagicMock()
+        mock_exp.result = {
+            "IC": 0.04,
+            "ICIR": 0.5,
+            "Rank IC": 0.03,
+            "Rank ICIR": 0.4,
+            "1day.excess_return_with_cost.annualized_return ": 0.10,
+            "1day.excess_return_with_cost.information_ratio": 0.6,
+            "1day.excess_return_with_cost.max_drawdown": -0.12,
+        }
+        m = extract_metrics_from_experiment(mock_exp)
+        assert m.ic == 0.04
+        assert m.rank_ic == 0.03
+        assert m.mdd == -0.12
+
+    def test_extract_metrics_returns_default_on_error(self):
+        from rdagent.scenarios.qlib.proposal.bandit import extract_metrics_from_experiment
+
+        mock_exp = MagicMock()
+        mock_exp.result = None  # Will cause AttributeError
+        m = extract_metrics_from_experiment(mock_exp)
+        assert m.ic == 0.0
+        assert m.sharpe == 0.0
+
+    def test_sharpe_computation(self):
+        from rdagent.scenarios.qlib.proposal.bandit import extract_metrics_from_experiment
+
+        mock_exp = MagicMock()
+        mock_exp.result = {
+            "IC": 0.0, "ICIR": 0.0, "Rank IC": 0.0, "Rank ICIR": 0.0,
+            "1day.excess_return_with_cost.annualized_return ": 0.15,
+            "1day.excess_return_with_cost.information_ratio": 0.0,
+            "1day.excess_return_with_cost.max_drawdown": -0.10,
+        }
+        m = extract_metrics_from_experiment(mock_exp)
+        assert m.sharpe == pytest.approx(1.5)  # 0.15 / 0.10
+
+
+# =============================================================================
+# LinearThompsonTwoArm
+# =============================================================================
+
+
+class TestLinearThompsonTwoArm:
+    def test_initialization(self):
+        from rdagent.scenarios.qlib.proposal.bandit import LinearThompsonTwoArm
+        bandit = LinearThompsonTwoArm(dim=5)
+        assert bandit.dim == 5
+        assert bandit.noise_var == 1.0
+        assert bandit.mean["factor"].shape == (5,)
+        assert bandit.mean["model"].shape == (5,)
+        assert bandit.precision["factor"].shape == (5, 5)
+
+    def test_sample_reward_returns_float(self):
+        from rdagent.scenarios.qlib.proposal.bandit import LinearThompsonTwoArm
+        bandit = LinearThompsonTwoArm(dim=3)
+        x = np.ones(3)
+        reward = bandit.sample_reward("factor", x)
+        assert isinstance(reward, float)
+
+    def test_arms_are_initialized_identically(self):
+        from rdagent.scenarios.qlib.proposal.bandit import LinearThompsonTwoArm
+        bandit = LinearThompsonTwoArm(dim=4)
+        assert np.array_equal(bandit.mean["factor"], bandit.mean["model"])
+        assert np.array_equal(bandit.precision["factor"], bandit.precision["model"])
+
+
+# =============================================================================
+# quant_loop_factory.py
+# =============================================================================
+
+
+class TestHasLocalComponents:
+    def test_returns_bool(self):
+        from rdagent.scenarios.qlib.quant_loop_factory import has_local_components
+        result = has_local_components()
+        assert isinstance(result, bool)
+
+    def test_returns_false_with_no_local_dir(self, monkeypatch):
+        from rdagent.scenarios.qlib import quant_loop_factory
+        monkeypatch.setattr(quant_loop_factory.Path, "exists", lambda self: False)
+        assert quant_loop_factory.has_local_components() is False
+
+
+class TestCountValidFactors:
+    def test_returns_zero_when_no_dir(self):
+        from rdagent.scenarios.qlib.quant_loop_factory import count_valid_factors
+        with patch("rdagent.scenarios.qlib.quant_loop_factory.Path.exists", return_value=False):
+            assert count_valid_factors() == 0
+
+    def test_returns_int(self):
+        from rdagent.scenarios.qlib.quant_loop_factory import count_valid_factors
+        result = count_valid_factors()
+        assert isinstance(result, int)
+        assert result >= 0
+
+
+class TestAdvancedLoopThreshold:
+    def test_constant_is_defined(self):
+        from rdagent.scenarios.qlib.quant_loop_factory import ADVANCED_LOOP_FACTOR_THRESHOLD
+        assert ADVANCED_LOOP_FACTOR_THRESHOLD == 5000