mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-27 23:47:46 +00:00
TPTBusiness
a469692141
- costeer_deep: 112 tests (knowledge base, feedback, evaluators, auto-fixer) - workflow_deep: 84 tests (RDLoop, proposals, traces, hypothesis/pickle) - core_deep: 74 tests (developer, evaluator, exceptions, experiment, scenario) - llm_utils_deep: 49 tests (embeddings, APIBackend, edge cases, Unicode/NaN) - utils_deep: 122 tests (shrink_text, templates, md5_hash, property-based, stress)
894 lines
42 KiB
Python
894 lines
42 KiB
Python
"""Deep tests for CoSTEER components: knowledge_management, evaluators, eva_utils, evolving_strategy, auto_fixer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import pickle
|
|
import sys
|
|
import tempfile
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
from typing import Any
|
|
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))
|
|
|
|
|
|
def _make_mock_fb_workspace(codes: str = "print('hello')") -> Any:
|
|
ws = MagicMock()
|
|
ws.all_codes = codes
|
|
ws.copy.return_value = ws
|
|
return ws
|
|
|
|
|
|
def _make_mock_task(task_info: str = "factor task info") -> Any:
|
|
t = MagicMock()
|
|
t.get_task_information.return_value = task_info
|
|
return t
|
|
|
|
|
|
# =============================================================================
|
|
# Import safety
|
|
# =============================================================================
|
|
|
|
COSTEER_MODULES = [
|
|
"rdagent.components.coder.CoSTEER.knowledge_management",
|
|
"rdagent.components.coder.CoSTEER.evaluators",
|
|
"rdagent.components.coder.CoSTEER.evolvable_subjects",
|
|
"rdagent.components.coder.CoSTEER.evolving_strategy",
|
|
"rdagent.components.coder.CoSTEER.config",
|
|
"rdagent.components.coder.factor_coder.evolving_strategy",
|
|
"rdagent.components.coder.factor_coder.eva_utils",
|
|
"rdagent.components.coder.factor_coder.auto_fixer",
|
|
"rdagent.components.coder.factor_coder.factor",
|
|
"rdagent.components.coder.factor_coder.config",
|
|
"rdagent.components.knowledge_management.graph",
|
|
]
|
|
|
|
|
|
class TestCosteerImports:
|
|
@pytest.mark.parametrize("mod_name", COSTEER_MODULES)
|
|
def test_module_is_importable(self, mod_name: str) -> None:
|
|
"""Verify each CoSTEER submodule can be imported without error."""
|
|
import importlib
|
|
mod = importlib.import_module(mod_name)
|
|
assert mod is not None
|
|
|
|
|
|
# =============================================================================
|
|
# CoSTEERKnowledge
|
|
# =============================================================================
|
|
|
|
|
|
class TestCoSTEERKnowledge:
|
|
def test_construction_stores_task_implementation_feedback(self) -> None:
|
|
"""Knowledge stores target_task, implementation, and feedback."""
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERKnowledge
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
task = _make_mock_task()
|
|
ws = _make_mock_fb_workspace("def f(): pass")
|
|
fb = CoSTEERSingleFeedback(execution="OK", return_checking="pass", code="good", final_decision=True)
|
|
k = CoSTEERKnowledge(target_task=task, implementation=ws, feedback=fb)
|
|
assert k.target_task is task
|
|
assert k.implementation is ws
|
|
assert k.feedback is fb
|
|
|
|
def test_get_implementation_and_feedback_str_contains_code(self) -> None:
|
|
"""The formatted string includes implementation code and feedback."""
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERKnowledge
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
task = _make_mock_task()
|
|
ws = _make_mock_fb_workspace("def my_factor(): return df")
|
|
fb = CoSTEERSingleFeedback(execution="ran", return_checking="ok", code="fine", final_decision=True)
|
|
k = CoSTEERKnowledge(target_task=task, implementation=ws, feedback=fb)
|
|
s = k.get_implementation_and_feedback_str()
|
|
assert "def my_factor" in s
|
|
|
|
def test_copy_implementation_is_called(self) -> None:
|
|
"""Knowledge copies the implementation workspace in __init__."""
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERKnowledge
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
task = _make_mock_task()
|
|
ws = _make_mock_fb_workspace("code")
|
|
ws._copy_called = False
|
|
def side_effect():
|
|
ws._copy_called = True
|
|
return ws
|
|
ws.copy = MagicMock(side_effect=side_effect)
|
|
fb = CoSTEERSingleFeedback(execution="x", return_checking="x", code="x", final_decision=False)
|
|
CoSTEERKnowledge(target_task=task, implementation=ws, feedback=fb)
|
|
assert ws._copy_called
|
|
|
|
|
|
# =============================================================================
|
|
# CoSTEERRAGStrategy — load, init, dump
|
|
# =============================================================================
|
|
|
|
|
|
class TestCoSTEERRAGStrategy:
|
|
def test_load_or_init_creates_v2_when_no_file(self) -> None:
|
|
"""Creates a fresh CoSTEERKnowledgeBaseV2 when no former file exists."""
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import (
|
|
CoSTEERRAGStrategyV2,
|
|
CoSTEERKnowledgeBaseV2,
|
|
)
|
|
strategy = CoSTEERRAGStrategyV2(settings=MagicMock(), dump_knowledge_base_path=Path("/nonexistent_12345.pkl"))
|
|
kb = strategy.load_or_init_knowledge_base(former_knowledge_base_path=None, evolving_version=2)
|
|
assert isinstance(kb, CoSTEERKnowledgeBaseV2)
|
|
|
|
def test_load_or_init_creates_v1_when_no_file(self) -> None:
|
|
"""Creates a fresh CoSTEERKnowledgeBaseV1 when no former file exists."""
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import (
|
|
CoSTEERRAGStrategyV1,
|
|
CoSTEERKnowledgeBaseV1,
|
|
)
|
|
strategy = CoSTEERRAGStrategyV1(settings=MagicMock(), dump_knowledge_base_path=None)
|
|
kb = strategy.load_or_init_knowledge_base(former_knowledge_base_path=None, evolving_version=1)
|
|
assert isinstance(kb, CoSTEERKnowledgeBaseV1)
|
|
|
|
def test_dump_knowledge_base_creates_dir_and_file(self) -> None:
|
|
"""dump_knowledge_base writes pickle file when path is set."""
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import (
|
|
CoSTEERRAGStrategyV2,
|
|
CoSTEERKnowledgeBaseV2,
|
|
)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
dump_path = Path(tmpdir) / "sub" / "kb.pkl"
|
|
strategy = CoSTEERRAGStrategyV2(settings=MagicMock(), dump_knowledge_base_path=dump_path)
|
|
strategy.knowledgebase = CoSTEERKnowledgeBaseV2()
|
|
strategy.dump_knowledge_base()
|
|
assert dump_path.exists()
|
|
|
|
def test_dump_knowledge_base_skips_when_path_is_none(self) -> None:
|
|
"""No error when dump path is None."""
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERRAGStrategyV2
|
|
strategy = CoSTEERRAGStrategyV2(settings=MagicMock(), dump_knowledge_base_path=None)
|
|
strategy.dump_knowledge_base()
|
|
|
|
def test_load_dumped_knowledge_base_restores(self) -> None:
|
|
"""Loading from a dumped file restores the knowledge base."""
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import (
|
|
CoSTEERRAGStrategyV2,
|
|
CoSTEERKnowledgeBaseV2,
|
|
)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
dump_path = Path(tmpdir) / "kb.pkl"
|
|
s1 = CoSTEERRAGStrategyV2(settings=MagicMock(), dump_knowledge_base_path=dump_path)
|
|
s1.knowledgebase = CoSTEERKnowledgeBaseV2()
|
|
s1.knowledgebase.success_task_to_knowledge_dict["k"] = "v"
|
|
s1.dump_knowledge_base()
|
|
s2 = CoSTEERRAGStrategyV2(settings=MagicMock(), dump_knowledge_base_path=dump_path)
|
|
s2.load_dumped_knowledge_base()
|
|
assert s2.knowledgebase is not None
|
|
assert s2.knowledgebase.success_task_to_knowledge_dict["k"] == "v"
|
|
|
|
|
|
# =============================================================================
|
|
# CoSTEERQueriedKnowledge and variants
|
|
# =============================================================================
|
|
|
|
|
|
class TestCoSTEERQueriedKnowledge:
|
|
def test_default_construction_empty_dicts(self) -> None:
|
|
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_construction_with_data(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERQueriedKnowledge
|
|
qk = CoSTEERQueriedKnowledge(
|
|
success_task_to_knowledge_dict={"a": 1}, failed_task_info_set={"b"},
|
|
)
|
|
assert qk.success_task_to_knowledge_dict == {"a": 1}
|
|
assert qk.failed_task_info_set == {"b"}
|
|
|
|
@pytest.mark.parametrize("dict_val", [{}, {"k": None}, {"a": 1, "b": 2}])
|
|
def test_success_task_to_knowledge_dict_variants(self, dict_val: dict) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERQueriedKnowledge
|
|
qk = CoSTEERQueriedKnowledge(success_task_to_knowledge_dict=dict_val)
|
|
assert qk.success_task_to_knowledge_dict == dict_val
|
|
|
|
@pytest.mark.parametrize("set_val", [set(), {"x"}, {"a", "b", "c"}])
|
|
def test_failed_task_info_set_variants(self, set_val: set) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERQueriedKnowledge
|
|
qk = CoSTEERQueriedKnowledge(failed_task_info_set=set_val)
|
|
assert qk.failed_task_info_set == set_val
|
|
|
|
|
|
class TestCoSTEERQueriedKnowledgeV1:
|
|
def test_extra_fields_default_to_empty(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERQueriedKnowledgeV1
|
|
qk = CoSTEERQueriedKnowledgeV1()
|
|
assert qk.task_to_former_failed_traces == {}
|
|
assert qk.task_to_similar_task_successful_knowledge == {}
|
|
|
|
def test_custom_extra_fields(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERQueriedKnowledgeV1
|
|
qk = CoSTEERQueriedKnowledgeV1(
|
|
task_to_former_failed_traces={"t": []}, task_to_similar_task_successful_knowledge={"t": ["k"]},
|
|
)
|
|
assert qk.task_to_former_failed_traces == {"t": []}
|
|
|
|
def test_inherits_from_base_queried_knowledge(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import (
|
|
CoSTEERQueriedKnowledge, CoSTEERQueriedKnowledgeV1,
|
|
)
|
|
qk = CoSTEERQueriedKnowledgeV1()
|
|
assert isinstance(qk, CoSTEERQueriedKnowledge)
|
|
|
|
|
|
class TestCoSTEERQueriedKnowledgeV2:
|
|
def test_extra_field_defaults_to_empty(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERQueriedKnowledgeV2
|
|
qk = CoSTEERQueriedKnowledgeV2()
|
|
assert qk.task_to_similar_error_successful_knowledge == {}
|
|
|
|
def test_inherits_from_v1(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import (
|
|
CoSTEERQueriedKnowledgeV1, CoSTEERQueriedKnowledgeV2,
|
|
)
|
|
qk = CoSTEERQueriedKnowledgeV2()
|
|
assert isinstance(qk, CoSTEERQueriedKnowledgeV1)
|
|
|
|
|
|
# =============================================================================
|
|
# CoSTEERKnowledgeBaseV1
|
|
# =============================================================================
|
|
|
|
|
|
class TestCoSTEERKnowledgeBaseV1:
|
|
def test_default_construction(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERKnowledgeBaseV1
|
|
kb = CoSTEERKnowledgeBaseV1()
|
|
assert kb.implementation_trace == {}
|
|
assert kb.success_task_info_set == set()
|
|
assert kb.task_to_embedding == {}
|
|
|
|
def test_query_raises_not_implemented(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERKnowledgeBaseV1
|
|
kb = CoSTEERKnowledgeBaseV1()
|
|
with pytest.raises(NotImplementedError):
|
|
kb.query()
|
|
|
|
|
|
# =============================================================================
|
|
# CoSTEERKnowledgeBaseV2
|
|
# =============================================================================
|
|
|
|
|
|
class TestCoSTEERKnowledgeBaseV2:
|
|
def test_default_construction_has_attributes(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERKnowledgeBaseV2
|
|
kb = CoSTEERKnowledgeBaseV2()
|
|
assert kb.working_trace_knowledge == {}
|
|
assert kb.working_trace_error_analysis == {}
|
|
assert kb.success_task_to_knowledge_dict == {}
|
|
assert kb.node_to_implementation_knowledge_dict == {}
|
|
assert kb.task_to_component_nodes == {}
|
|
|
|
def test_v2_importable_and_instantiable(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERKnowledgeBaseV2
|
|
kb = CoSTEERKnowledgeBaseV2()
|
|
assert kb is not None
|
|
assert kb.working_trace_knowledge == {}
|
|
assert kb.success_task_to_knowledge_dict == {}
|
|
|
|
def test_has_update_success_task_method(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERKnowledgeBaseV2
|
|
kb = CoSTEERKnowledgeBaseV2()
|
|
assert hasattr(kb, "update_success_task")
|
|
|
|
def test_has_graph_query_methods(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.knowledge_management import CoSTEERKnowledgeBaseV2
|
|
assert hasattr(CoSTEERKnowledgeBaseV2, "graph_query_by_content")
|
|
assert hasattr(CoSTEERKnowledgeBaseV2, "graph_query_by_node")
|
|
assert hasattr(CoSTEERKnowledgeBaseV2, "graph_query_by_intersection")
|
|
|
|
|
|
# =============================================================================
|
|
# CoSTEERSingleFeedback
|
|
# =============================================================================
|
|
|
|
|
|
class TestCoSTEERSingleFeedback:
|
|
def test_construction_with_all_fields(self) -> None:
|
|
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_construction_default_final_decision_none(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
fb = CoSTEERSingleFeedback(execution="x", return_checking="x", code="x")
|
|
assert fb.final_decision is None
|
|
|
|
def test_val_and_update_init_dict_converts_false_string(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
for s in ("false", "False"):
|
|
result = CoSTEERSingleFeedback.val_and_update_init_dict({
|
|
"execution": "x", "return_checking": "x", "code": "x", "final_decision": s,
|
|
})
|
|
assert result["final_decision"] is False
|
|
|
|
def test_val_and_update_init_dict_converts_true_string(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
for s in ("true", "True"):
|
|
result = CoSTEERSingleFeedback.val_and_update_init_dict({
|
|
"execution": "x", "return_checking": "x", "code": "x", "final_decision": s,
|
|
})
|
|
assert result["final_decision"] is True
|
|
|
|
def test_val_and_update_init_dict_raises_on_missing_final_decision(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
with pytest.raises(ValueError):
|
|
CoSTEERSingleFeedback.val_and_update_init_dict({
|
|
"execution": "x", "return_checking": "x", "code": "x",
|
|
})
|
|
|
|
def test_val_and_update_init_dict_raises_on_invalid_type(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
with pytest.raises(ValueError):
|
|
CoSTEERSingleFeedback.val_and_update_init_dict({
|
|
"execution": "x", "return_checking": "x", "code": "x", "final_decision": 1,
|
|
})
|
|
|
|
def test_val_and_update_init_dict_jsonifies_non_string_attrs(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
result = CoSTEERSingleFeedback.val_and_update_init_dict({
|
|
"execution": ["line1"], "return_checking": {"status": "ok"}, "code": ["def f(): pass"],
|
|
"final_decision": True,
|
|
})
|
|
assert isinstance(result["execution"], str)
|
|
|
|
def test_val_and_update_init_dict_preserves_none_attrs(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
result = CoSTEERSingleFeedback.val_and_update_init_dict({
|
|
"execution": "x", "return_checking": None, "code": "x", "final_decision": False,
|
|
})
|
|
assert result["return_checking"] is None
|
|
|
|
def test_merge_all_true_makes_true(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
fb1 = CoSTEERSingleFeedback(execution="a", return_checking="a", code="a", final_decision=True)
|
|
fb2 = CoSTEERSingleFeedback(execution="b", return_checking="b", code="b", final_decision=True)
|
|
merged = CoSTEERSingleFeedback.merge([fb1, fb2])
|
|
assert merged.final_decision is True
|
|
|
|
def test_merge_one_false_makes_false(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
fb1 = CoSTEERSingleFeedback(execution="a", return_checking="a", code="a", final_decision=True)
|
|
fb2 = CoSTEERSingleFeedback(execution="b", return_checking="b", code="b", final_decision=False)
|
|
merged = CoSTEERSingleFeedback.merge([fb1, fb2])
|
|
assert merged.final_decision is False
|
|
|
|
def test_merge_concatenates_execution_strings(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
fb1 = CoSTEERSingleFeedback(execution="ex1", return_checking="r1", code="c1", final_decision=False)
|
|
fb2 = CoSTEERSingleFeedback(execution="ex2", return_checking="r2", code="c2", final_decision=False)
|
|
merged = CoSTEERSingleFeedback.merge([fb1, fb2])
|
|
assert "ex1\n\nex2" in merged.execution
|
|
|
|
def test_merge_preserves_source_feedback(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
fb1 = CoSTEERSingleFeedback(execution="a", return_checking="a", code="a", final_decision=True, source_feedback={"e1": True})
|
|
fb2 = CoSTEERSingleFeedback(execution="b", return_checking="b", code="b", final_decision=True, source_feedback={"e2": False})
|
|
merged = CoSTEERSingleFeedback.merge([fb1, fb2])
|
|
assert merged.source_feedback["e1"] is True
|
|
assert merged.source_feedback["e2"] is False
|
|
|
|
def test_str_contains_success_on_true(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
fb = CoSTEERSingleFeedback(execution="x", return_checking="y", code="z", final_decision=True)
|
|
assert "SUCCESS" in str(fb)
|
|
|
|
def test_str_contains_fail_on_false(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
fb = CoSTEERSingleFeedback(execution="x", return_checking="y", code="z", final_decision=False)
|
|
assert "FAIL" in str(fb)
|
|
|
|
def test_str_no_return_checking_when_none(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
fb = CoSTEERSingleFeedback(execution="x", return_checking=None, code="z", final_decision=False)
|
|
assert "No return checking" in str(fb)
|
|
|
|
def test_bool_returns_final_decision(self) -> None:
|
|
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_source_feedback_defaults_to_empty_dict(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
fb = CoSTEERSingleFeedback(execution="x", return_checking="x", code="x", final_decision=True)
|
|
assert fb.source_feedback == {}
|
|
|
|
def test_raw_execution_default_empty_str(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
fb = CoSTEERSingleFeedback(execution="x", return_checking="x", code="x", final_decision=True)
|
|
assert fb.raw_execution == ""
|
|
|
|
def test_pickle_safety(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
fb = CoSTEERSingleFeedback(execution="exec", return_checking="ret", code="code", final_decision=True, source_feedback={"src": True})
|
|
data = pickle.dumps(fb)
|
|
fb2 = pickle.loads(data)
|
|
assert fb2.execution == "exec"
|
|
assert fb2.final_decision is True
|
|
|
|
def test_merge_single_item_is_deepcopy(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
fb = CoSTEERSingleFeedback(execution="ex", return_checking="rc", code="cd", final_decision=True)
|
|
merged = CoSTEERSingleFeedback.merge([fb])
|
|
assert merged is not fb
|
|
|
|
def test_final_decision_bool_conversion(self) -> None:
|
|
"""True/False boolean is passed through unchanged."""
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedback
|
|
result = CoSTEERSingleFeedback.val_and_update_init_dict({
|
|
"execution": "x", "return_checking": "x", "code": "x", "final_decision": True,
|
|
})
|
|
assert result["final_decision"] is True
|
|
|
|
|
|
# =============================================================================
|
|
# CoSTEERSingleFeedbackDeprecated
|
|
# =============================================================================
|
|
|
|
|
|
class TestCoSTEERSingleFeedbackDeprecated:
|
|
def test_construction_with_all_kwargs(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedbackDeprecated
|
|
fb = CoSTEERSingleFeedbackDeprecated(execution_feedback="e", shape_feedback="s", code_feedback="c",
|
|
value_feedback="v", final_decision=True, final_feedback="f",
|
|
value_generated_flag=True, final_decision_based_on_gt=True, source_feedback={"src": True})
|
|
assert fb.execution_feedback == "e"
|
|
|
|
def test_execution_property_returns_execution_feedback(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedbackDeprecated
|
|
fb = CoSTEERSingleFeedbackDeprecated(execution_feedback="hello")
|
|
assert fb.execution == "hello"
|
|
|
|
def test_execution_setter_sets_execution_feedback(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedbackDeprecated
|
|
fb = CoSTEERSingleFeedbackDeprecated()
|
|
fb.execution = "world"
|
|
assert fb.execution_feedback == "world"
|
|
|
|
def test_return_checking_returns_feedback_when_generated(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedbackDeprecated
|
|
fb = CoSTEERSingleFeedbackDeprecated(value_feedback="val ok", shape_feedback="shape ok", value_generated_flag=True)
|
|
rc = fb.return_checking
|
|
assert "val ok" in rc
|
|
|
|
def test_return_checking_returns_none_when_not_generated(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedbackDeprecated
|
|
fb = CoSTEERSingleFeedbackDeprecated(value_generated_flag=False)
|
|
assert fb.return_checking is None
|
|
|
|
def test_code_property_returns_code_feedback(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedbackDeprecated
|
|
fb = CoSTEERSingleFeedbackDeprecated(code_feedback="my code")
|
|
assert fb.code == "my code"
|
|
|
|
def test_code_setter_sets_code_feedback(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedbackDeprecated
|
|
fb = CoSTEERSingleFeedbackDeprecated()
|
|
fb.code = "new code"
|
|
assert fb.code_feedback == "new code"
|
|
|
|
def test_str_contains_all_sections(self) -> None:
|
|
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)
|
|
assert "exec" in s
|
|
assert "SUCCESS" in s
|
|
|
|
def test_default_values_are_none(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERSingleFeedbackDeprecated
|
|
fb = CoSTEERSingleFeedbackDeprecated()
|
|
assert fb.execution_feedback is None
|
|
assert fb.final_decision is None
|
|
|
|
|
|
# =============================================================================
|
|
# CoSTEERMultiFeedback
|
|
# =============================================================================
|
|
|
|
|
|
class TestCoSTEERMultiFeedback:
|
|
def test_empty_construction(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback
|
|
mf = CoSTEERMultiFeedback([])
|
|
assert len(mf) == 0
|
|
|
|
def test_getitem_returns_single_feedback(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback, CoSTEERSingleFeedback
|
|
fb = CoSTEERSingleFeedback(execution="x", return_checking="x", code="x", final_decision=True)
|
|
mf = CoSTEERMultiFeedback([fb])
|
|
assert mf[0] is fb
|
|
|
|
def test_append_adds_feedback(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback, CoSTEERSingleFeedback
|
|
mf = CoSTEERMultiFeedback([])
|
|
fb = CoSTEERSingleFeedback(execution="x", return_checking="x", code="x", final_decision=True)
|
|
mf.append(fb)
|
|
assert len(mf) == 1
|
|
|
|
def test_iter_yields_all_feedbacks(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback, CoSTEERSingleFeedback
|
|
fbs = [
|
|
CoSTEERSingleFeedback(execution="a", return_checking="a", code="a", final_decision=True),
|
|
CoSTEERSingleFeedback(execution="b", return_checking="b", code="b", final_decision=False),
|
|
]
|
|
mf = CoSTEERMultiFeedback(fbs)
|
|
assert list(mf) == fbs
|
|
|
|
def test_is_acceptable_all_true(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback, CoSTEERSingleFeedback
|
|
fbs = [
|
|
CoSTEERSingleFeedback(execution="a", return_checking="a", code="a", final_decision=True),
|
|
CoSTEERSingleFeedback(execution="b", return_checking="b", code="b", final_decision=True),
|
|
]
|
|
assert CoSTEERMultiFeedback(fbs).is_acceptable()
|
|
|
|
def test_is_acceptable_any_false(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback, CoSTEERSingleFeedback
|
|
fbs = [
|
|
CoSTEERSingleFeedback(execution="a", return_checking="a", code="a", final_decision=True),
|
|
CoSTEERSingleFeedback(execution="b", return_checking="b", code="b", final_decision=False),
|
|
]
|
|
assert not CoSTEERMultiFeedback(fbs).is_acceptable()
|
|
|
|
def test_finished_succeeds_with_none_feedbacks(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback, CoSTEERSingleFeedback
|
|
fbs = [
|
|
CoSTEERSingleFeedback(execution="a", return_checking="a", code="a", final_decision=True),
|
|
None,
|
|
]
|
|
assert CoSTEERMultiFeedback(fbs).finished()
|
|
|
|
def test_bool_all_true(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiFeedback, CoSTEERSingleFeedback
|
|
fbs = [CoSTEERSingleFeedback(execution="a", return_checking="a", code="a", final_decision=True)]
|
|
assert bool(CoSTEERMultiFeedback(fbs))
|
|
|
|
|
|
# =============================================================================
|
|
# CoSTEERMultiEvaluator
|
|
# =============================================================================
|
|
|
|
|
|
class TestCoSTEERMultiEvaluator:
|
|
def test_initialization_with_single_evaluator(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiEvaluator
|
|
mock_eval = MagicMock()
|
|
evaluator = CoSTEERMultiEvaluator(single_evaluator=mock_eval, scen=MagicMock())
|
|
assert evaluator.single_evaluator is mock_eval
|
|
|
|
def test_initialization_with_list_of_evaluators(self) -> None:
|
|
from rdagent.components.coder.CoSTEER.evaluators import CoSTEERMultiEvaluator
|
|
mock_evals = [MagicMock(), MagicMock()]
|
|
evaluator = CoSTEERMultiEvaluator(single_evaluator=mock_evals, scen=MagicMock())
|
|
assert evaluator.single_evaluator == mock_evals
|
|
|
|
|
|
# =============================================================================
|
|
# Factor Evaluators
|
|
# =============================================================================
|
|
|
|
|
|
class TestFactorInfEvaluator:
|
|
def test_no_inf_values_returns_true(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorInfEvaluator
|
|
evaluator = FactorInfEvaluator()
|
|
imp = MagicMock()
|
|
df = pd.DataFrame({"f": [1.0, 2.0]}, index=pd.MultiIndex.from_tuples(
|
|
[("2020-01-01", "EUR"), ("2020-01-02", "EUR")], names=["datetime", "instrument"]))
|
|
imp.execute.return_value = (None, df)
|
|
_, result_bool = evaluator.evaluate(implementation=imp, gt_implementation=None)
|
|
assert result_bool is True
|
|
|
|
def test_with_inf_values_returns_false(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorInfEvaluator
|
|
evaluator = FactorInfEvaluator()
|
|
imp = MagicMock()
|
|
df = pd.DataFrame({"f": [float("inf"), 2.0]}, index=pd.MultiIndex.from_tuples(
|
|
[("2020-01-01", "EUR"), ("2020-01-02", "EUR")], names=["datetime", "instrument"]))
|
|
imp.execute.return_value = (None, df)
|
|
_, result_bool = evaluator.evaluate(implementation=imp, gt_implementation=None)
|
|
assert result_bool is False
|
|
|
|
def test_none_dataframe_returns_false(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorInfEvaluator
|
|
evaluator = FactorInfEvaluator()
|
|
imp = MagicMock()
|
|
imp.execute.return_value = (None, None)
|
|
_, result_bool = evaluator.evaluate(implementation=imp, gt_implementation=None)
|
|
assert result_bool is False
|
|
|
|
|
|
class TestFactorSingleColumnEvaluator:
|
|
def test_single_column_returns_true(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorSingleColumnEvaluator
|
|
evaluator = FactorSingleColumnEvaluator()
|
|
imp = MagicMock()
|
|
df = pd.DataFrame({"col": [1]}, index=pd.MultiIndex.from_tuples(
|
|
[("2020-01-01", "EUR")], names=["datetime", "instrument"]))
|
|
imp.execute.return_value = (None, df)
|
|
_, result = evaluator.evaluate(implementation=imp, gt_implementation=None)
|
|
assert result is True
|
|
|
|
def test_multi_column_returns_false(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorSingleColumnEvaluator
|
|
evaluator = FactorSingleColumnEvaluator()
|
|
imp = MagicMock()
|
|
df = pd.DataFrame({"a": [1], "b": [2]}, index=pd.MultiIndex.from_tuples(
|
|
[("2020-01-01", "EUR")], names=["datetime", "instrument"]))
|
|
imp.execute.return_value = (None, df)
|
|
_, result = evaluator.evaluate(implementation=imp, gt_implementation=None)
|
|
assert result is False
|
|
|
|
|
|
class TestFactorRowCountEvaluator:
|
|
def test_equal_row_count_returns_ratio_one(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorRowCountEvaluator
|
|
evaluator = FactorRowCountEvaluator()
|
|
imp = MagicMock()
|
|
gt = MagicMock()
|
|
idx = pd.MultiIndex.from_tuples([("2020-01-01", "EUR")], names=["datetime", "instrument"])
|
|
imp.execute.return_value = (None, pd.DataFrame({"f": [1]}, index=idx))
|
|
gt.execute.return_value = (None, pd.DataFrame({"f": [2]}, index=idx))
|
|
_, ratio = evaluator.evaluate(implementation=imp, gt_implementation=gt)
|
|
assert ratio == 1.0
|
|
|
|
def test_different_row_count_returns_ratio_below_one(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorRowCountEvaluator
|
|
evaluator = FactorRowCountEvaluator()
|
|
imp = MagicMock()
|
|
gt = MagicMock()
|
|
idx_a = pd.MultiIndex.from_tuples([("2020-01-01", "EUR")], names=["datetime", "instrument"])
|
|
idx_b = pd.MultiIndex.from_tuples(
|
|
[("2020-01-01", "EUR"), ("2020-01-02", "EUR")], names=["datetime", "instrument"])
|
|
imp.execute.return_value = (None, pd.DataFrame({"f": [1]}, index=idx_a))
|
|
gt.execute.return_value = (None, pd.DataFrame({"f": [2, 3]}, index=idx_b))
|
|
_, ratio = evaluator.evaluate(implementation=imp, gt_implementation=gt)
|
|
assert ratio < 1.0
|
|
|
|
@pytest.mark.parametrize("gen_rows,gt_rows,expected", [
|
|
(5, 5, 1.0),
|
|
(100, 5, 0.05),
|
|
(500, 500, 1.0),
|
|
])
|
|
def test_row_count_variants(self, gen_rows: int, gt_rows: int, expected: float) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorRowCountEvaluator
|
|
evaluator = FactorRowCountEvaluator()
|
|
imp = MagicMock()
|
|
gt = MagicMock()
|
|
gen_idx = pd.MultiIndex.from_tuples(
|
|
[(f"2020-01-{i+1:02d}", "EUR") for i in range(max(gen_rows, 1))],
|
|
names=["datetime", "instrument"])
|
|
gt_idx = pd.MultiIndex.from_tuples(
|
|
[(f"2020-01-{i+1:02d}", "EUR") for i in range(max(gt_rows, 1))],
|
|
names=["datetime", "instrument"])
|
|
imp.execute.return_value = (None, pd.DataFrame({"f": list(range(len(gen_idx)))}, index=gen_idx))
|
|
gt.execute.return_value = (None, pd.DataFrame({"f": list(range(len(gt_idx)))}, index=gt_idx))
|
|
_, ratio = evaluator.evaluate(implementation=imp, gt_implementation=gt)
|
|
assert ratio == pytest.approx(expected)
|
|
|
|
|
|
class TestFactorIndexEvaluator:
|
|
def test_identical_index_returns_one(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorIndexEvaluator
|
|
evaluator = FactorIndexEvaluator()
|
|
imp = MagicMock()
|
|
gt = MagicMock()
|
|
idx = pd.MultiIndex.from_tuples(
|
|
[("2020-01-01", "EUR"), ("2020-01-02", "EUR")], names=["datetime", "instrument"])
|
|
imp.execute.return_value = (None, pd.DataFrame({"f": [1, 2]}, index=idx))
|
|
gt.execute.return_value = (None, pd.DataFrame({"f": [3, 4]}, index=idx))
|
|
_, sim = evaluator.evaluate(implementation=imp, gt_implementation=gt)
|
|
assert sim == 1.0
|
|
|
|
def test_disjoint_index_returns_zero(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorIndexEvaluator
|
|
evaluator = FactorIndexEvaluator()
|
|
imp = MagicMock()
|
|
gt = MagicMock()
|
|
idx_a = pd.MultiIndex.from_tuples([("2020-01-01", "EUR")], names=["datetime", "instrument"])
|
|
idx_b = pd.MultiIndex.from_tuples([("2020-01-02", "GBP")], names=["datetime", "instrument"])
|
|
imp.execute.return_value = (None, pd.DataFrame({"f": [1]}, index=idx_a))
|
|
gt.execute.return_value = (None, pd.DataFrame({"f": [2]}, index=idx_b))
|
|
_, sim = evaluator.evaluate(implementation=imp, gt_implementation=gt)
|
|
assert sim == 0.0
|
|
|
|
|
|
class TestFactorEqualValueRatioEvaluator:
|
|
def test_identical_values_return_accuracy_one(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorEqualValueRatioEvaluator
|
|
evaluator = FactorEqualValueRatioEvaluator()
|
|
imp = MagicMock()
|
|
gt = MagicMock()
|
|
idx = pd.MultiIndex.from_tuples(
|
|
[("2020-01-01", "EUR"), ("2020-01-02", "EUR")], names=["datetime", "instrument"])
|
|
df = pd.DataFrame({"f": [1.0, 2.0]}, index=idx)
|
|
imp.execute.return_value = (None, df)
|
|
gt.execute.return_value = (None, df.copy())
|
|
_, acc = evaluator.evaluate(implementation=imp, gt_implementation=gt)
|
|
assert acc == 1.0
|
|
|
|
def test_different_values_return_lower_accuracy(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorEqualValueRatioEvaluator
|
|
evaluator = FactorEqualValueRatioEvaluator()
|
|
imp = MagicMock()
|
|
gt = MagicMock()
|
|
idx = pd.MultiIndex.from_tuples(
|
|
[("2020-01-01", "EUR"), ("2020-01-02", "EUR")], names=["datetime", "instrument"])
|
|
imp.execute.return_value = (None, pd.DataFrame({"f": [1.0, 2.0]}, index=idx))
|
|
gt.execute.return_value = (None, pd.DataFrame({"f": [1.0, 3.0]}, index=idx))
|
|
_, acc = evaluator.evaluate(implementation=imp, gt_implementation=gt)
|
|
assert acc < 1.0
|
|
|
|
def test_none_dataframe_returns_negative_one(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorEqualValueRatioEvaluator
|
|
evaluator = FactorEqualValueRatioEvaluator()
|
|
imp = MagicMock()
|
|
imp.execute.return_value = (None, None)
|
|
_, acc = evaluator.evaluate(implementation=imp, gt_implementation=None)
|
|
assert acc == -1
|
|
|
|
|
|
class TestFactorCorrelationEvaluator:
|
|
def test_is_constructible(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorCorrelationEvaluator
|
|
ev = FactorCorrelationEvaluator(hard_check=True)
|
|
assert ev.hard_check is True
|
|
|
|
def test_none_dataframe_returns_false(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorCorrelationEvaluator
|
|
evaluator = FactorCorrelationEvaluator(hard_check=False)
|
|
imp = MagicMock()
|
|
imp.execute.return_value = (None, None)
|
|
_, result = evaluator.evaluate(implementation=imp, gt_implementation=None)
|
|
assert result is False
|
|
|
|
|
|
class TestFactorDatetimeDailyEvaluator:
|
|
def test_valid_datetime_index_returns_true(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorDatetimeDailyEvaluator
|
|
evaluator = FactorDatetimeDailyEvaluator()
|
|
imp = MagicMock()
|
|
idx = pd.MultiIndex.from_tuples(
|
|
[("2020-01-01 09:00:00", "EUR"), ("2020-01-01 10:00:00", "EUR")],
|
|
names=["datetime", "instrument"])
|
|
imp.execute.return_value = (None, pd.DataFrame({"f": [1, 2]}, index=idx))
|
|
_, result = evaluator.evaluate(implementation=imp, gt_implementation=None)
|
|
assert result is True
|
|
|
|
def test_no_datetime_index_returns_false(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorDatetimeDailyEvaluator
|
|
evaluator = FactorDatetimeDailyEvaluator()
|
|
imp = MagicMock()
|
|
imp.execute.return_value = (None, pd.DataFrame({"f": [1]}, index=[0]))
|
|
_, result = evaluator.evaluate(implementation=imp, gt_implementation=None)
|
|
assert result is False
|
|
|
|
|
|
class TestFactorValueEvaluator:
|
|
def test_evaluator_is_importable(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorValueEvaluator
|
|
assert FactorValueEvaluator is not None
|
|
|
|
def test_evaluate_method_exists(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorValueEvaluator
|
|
assert hasattr(FactorValueEvaluator, "evaluate")
|
|
|
|
|
|
class TestFactorFinalDecisionEvaluator:
|
|
def test_evaluator_is_importable(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorFinalDecisionEvaluator
|
|
assert FactorFinalDecisionEvaluator is not None
|
|
|
|
|
|
# =============================================================================
|
|
# FactorEvaluator base class
|
|
# =============================================================================
|
|
|
|
|
|
class TestFactorEvaluatorBase:
|
|
def test_str_returns_class_name(self) -> None:
|
|
from rdagent.components.coder.factor_coder.eva_utils import FactorInfEvaluator
|
|
ev = FactorInfEvaluator()
|
|
assert str(ev) == "FactorInfEvaluator"
|
|
|
|
|
|
# =============================================================================
|
|
# Auto-fixer (extends existing tests)
|
|
# =============================================================================
|
|
|
|
|
|
class TestAutoFixerEdgeCases:
|
|
@pytest.fixture
|
|
def fixer(self):
|
|
from rdagent.components.coder.factor_coder.auto_fixer import FactorAutoFixer
|
|
return FactorAutoFixer()
|
|
|
|
def test_empty_code_returns_empty(self, fixer) -> None:
|
|
result = fixer.fix("")
|
|
assert result == ""
|
|
|
|
def test_whitespace_only_code_preserved(self, fixer) -> None:
|
|
result = fixer.fix(" \n \n ")
|
|
assert " " in result
|
|
|
|
def test_none_task_info_does_not_crash(self, fixer) -> None:
|
|
result = fixer.fix("x = 1", factor_task_info=None)
|
|
assert "x = 1" in result
|
|
|
|
def test_very_long_code_handled(self, fixer) -> None:
|
|
long_code = "x = 1\n" * 100 + "df['x'] = df.groupby(level=1)['y'].mean()\n" + "y = 2\n" * 100
|
|
result = fixer.fix(long_code)
|
|
assert "groupby" in result
|
|
|
|
def test_convenience_function_returns_string(self, fixer) -> None:
|
|
from rdagent.components.coder.factor_coder.auto_fixer import auto_fix_factor_code
|
|
result = auto_fix_factor_code("x = 1")
|
|
assert isinstance(result, str)
|
|
|
|
def test_fixes_applied_list_tracks_changes(self, fixer) -> None:
|
|
code = "df.groupby(['instrument'])['x'].mean()"
|
|
fixer.fix(code)
|
|
assert len(fixer.fixes_applied) > 0
|
|
|
|
|
|
# =============================================================================
|
|
# FactorMultiProcessEvolvingStrategy
|
|
# =============================================================================
|
|
|
|
|
|
class TestFactorMultiProcessEvolvingStrategy:
|
|
def test_strategy_is_importable(self) -> None:
|
|
from rdagent.components.coder.factor_coder.evolving_strategy import (
|
|
FactorMultiProcessEvolvingStrategy,
|
|
)
|
|
assert FactorMultiProcessEvolvingStrategy is not None
|
|
|
|
|
|
# =============================================================================
|
|
# FactorFBWorkspace / FactorTask
|
|
# =============================================================================
|
|
|
|
|
|
class TestFactorWorkspaceImport:
|
|
def test_factor_fb_workspace_importable(self) -> None:
|
|
from rdagent.components.coder.factor_coder.factor import FactorFBWorkspace
|
|
assert FactorFBWorkspace is not None
|
|
|
|
def test_factor_task_importable(self) -> None:
|
|
from rdagent.components.coder.factor_coder.factor import FactorTask
|
|
assert FactorTask is not None
|
|
|
|
|
|
# =============================================================================
|
|
# UndirectedGraph / UndirectedNode
|
|
# =============================================================================
|
|
|
|
|
|
class TestUndirectedGraphIntegration:
|
|
def test_undirected_graph_importable(self) -> None:
|
|
from rdagent.components.knowledge_management.graph import UndirectedGraph
|
|
assert UndirectedGraph is not None
|
|
|
|
def test_undirected_node_importable(self) -> None:
|
|
from rdagent.components.knowledge_management.graph import UndirectedNode
|
|
assert UndirectedNode is not None
|