diff --git a/test/oai/test_llm_utils_deep.py b/test/oai/test_llm_utils_deep.py new file mode 100644 index 00000000..fb4ad9ee --- /dev/null +++ b/test/oai/test_llm_utils_deep.py @@ -0,0 +1,516 @@ +"""Deep tests for rdagent.oai.llm_utils: embedding distance, APIBackend, and edge cases.""" + +from __future__ import annotations + +import pickle +import sys +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import numpy as np +import pytest + +PROJECT_ROOT = Path(__file__).parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + + +# ============================================================================= +# Import safety +# ============================================================================= + +LLM_MODULES = [ + "rdagent.oai.llm_utils", + "rdagent.oai.llm_conf", + "rdagent.oai.backend.base", + "rdagent.utils", +] + + +class TestLLMImports: + @pytest.mark.parametrize("module_name", LLM_MODULES) + def test_module_importable(self, module_name: str) -> None: + """Each LLM utility module imports without error.""" + import importlib + mod = importlib.import_module(module_name) + assert mod is not None + + +# ============================================================================= +# calculate_embedding_distance_between_str_list +# ============================================================================= + + +class TestEmbeddingDistance: + """Tests for calculate_embedding_distance_between_str_list.""" + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_empty_source_returns_empty(self, mock_api: MagicMock) -> None: + """Empty source list returns nested empty list.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [] + result = calculate_embedding_distance_between_str_list([], ["target"]) + assert result == [[]] + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_empty_target_returns_empty(self, mock_api: MagicMock) -> None: + """Empty target list returns nested empty list.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [] + result = calculate_embedding_distance_between_str_list(["source"], []) + assert result == [[]] + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_both_empty_returns_empty(self, mock_api: MagicMock) -> None: + """Both lists empty returns nested empty list.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [] + result = calculate_embedding_distance_between_str_list([], []) + assert result == [[]] + + def test_both_empty_no_api_call(self) -> None: + """Empty inputs return [[]] without any API call.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + result = calculate_embedding_distance_between_str_list([], []) + assert result == [[]] + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_single_source_single_target(self, mock_api: MagicMock) -> None: + """Single source and target return 1x1 matrix.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [ + [0.5, 0.5], # source embedding + [0.5, 0.5], # target embedding + ] + result = calculate_embedding_distance_between_str_list(["s1"], ["t1"]) + assert len(result) == 1 + assert len(result[0]) == 1 + assert isinstance(result[0][0], float) + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_multiple_sources_single_target(self, mock_api: MagicMock) -> None: + """Multiple sources, single target returns n x 1 matrix.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [ + [1.0, 0.0], + [0.0, 1.0], + [0.5, 0.5], + ] + result = calculate_embedding_distance_between_str_list(["s1", "s2"], ["t1"]) + assert len(result) == 2 + assert len(result[0]) == 1 + assert len(result[1]) == 1 + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_similarity_range(self, mock_api: MagicMock) -> None: + """Similarity values should be in [-1, 1] range after normalization.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [ + [1.0, 0.0, 0.0], + [0.0, 1.0, 0.0], + [-1.0, 0.0, 0.0], + [0.7, 0.3, 0.1], + ] + result = calculate_embedding_distance_between_str_list( + ["s1", "s2", "s3"], ["t1"], + ) + for row in result: + for val in row: + assert -1.0 - 1e-9 <= val <= 1.0 + 1e-9 + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_identical_embedding_produces_one(self, mock_api: MagicMock) -> None: + """Identical embeddings produce similarity of 1.0.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [ + [3.0, 4.0], # source (norm=5, unit=[0.6, 0.8]) + [3.0, 4.0], # target (norm=5, unit=[0.6, 0.8]) + ] + result = calculate_embedding_distance_between_str_list(["s1"], ["t1"]) + assert result[0][0] == pytest.approx(1.0, abs=1e-9) + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_orthogonal_embedding_produces_zero(self, mock_api: MagicMock) -> None: + """Orthogonal embeddings produce similarity of 0.0.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [ + [1.0, 0.0], + [0.0, 1.0], + ] + result = calculate_embedding_distance_between_str_list(["s1"], ["t1"]) + assert result[0][0] == pytest.approx(0.0, abs=1e-9) + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_opposite_embedding_produces_negative_one(self, mock_api: MagicMock) -> None: + """Opposite embeddings produce similarity of -1.0.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [ + [1.0, 0.0], + [-1.0, 0.0], + ] + result = calculate_embedding_distance_between_str_list(["s1"], ["t1"]) + assert result[0][0] == pytest.approx(-1.0, abs=1e-9) + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_zero_vector_embedding(self, mock_api: MagicMock) -> None: + """Zero vector embedding should be handled (division by zero).""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [ + [0.0, 0.0], + [1.0, 0.0], + ] + # After normalization, zero vector becomes NaN, dot produces NaN + result = calculate_embedding_distance_between_str_list(["s1"], ["t1"]) + assert isinstance(result[0][0], float) + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_large_embedding_values(self, mock_api: MagicMock) -> None: + """Large-magnitude embeddings are correctly normalized.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [ + [1e5, 0.0], + [0.0, 1e5], + ] + result = calculate_embedding_distance_between_str_list(["s1"], ["t1"]) + assert isinstance(result[0][0], float) + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_return_type_is_list_of_lists_of_floats(self, mock_api: MagicMock) -> None: + """Return type is List[List[float]].""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [ + [1.0], + [1.0], + ] + result = calculate_embedding_distance_between_str_list(["a"], ["b"]) + assert isinstance(result, list) + assert isinstance(result[0], list) + assert isinstance(result[0][0], float) + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_matrix_shape_matches_input_counts(self, mock_api: MagicMock) -> None: + """Output matrix has shape (len(sources), len(targets)).""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + n_sources, n_targets = 3, 5 + # Create embeddings for all strings + emb_dim = 128 + embeddings = [ + list(np.random.randn(emb_dim)) + for _ in range(n_sources + n_targets) + ] + mock_api.return_value.create_embedding.return_value = embeddings + + sources = [f"s{i}" for i in range(n_sources)] + targets = [f"t{i}" for i in range(n_targets)] + result = calculate_embedding_distance_between_str_list(sources, targets) + assert len(result) == n_sources + assert all(len(row) == n_targets for row in result) + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_unicode_strings(self, mock_api: MagicMock) -> None: + """Unicode/emoji strings are handled.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [ + [0.5, 0.5], + [0.5, 0.5], + ] + result = calculate_embedding_distance_between_str_list(["日本語"], ["🌟"]) + assert len(result) == 1 + assert len(result[0]) == 1 + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_real_calculate_embedding_via_mock(self, mock_api: MagicMock) -> None: + """Full calculation path works via mocked API.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [ + [1.0, 2.0, 3.0], + [2.0, 3.0, 4.0], + [4.0, 2.0, 0.0], + [4.0, 1.0, 0.0], + ] + result = calculate_embedding_distance_between_str_list( + ["task_info_1", "task_info_2"], + ["target_1", "target_2"], + ) + assert len(result) == 2 + assert len(result[0]) == 2 + + +# ============================================================================= +# APIBackend +# ============================================================================= + + +class TestAPIBackend: + """Tests for APIBackend (alias for get_api_backend).""" + + def test_api_backend_is_callable_fn(self) -> None: + """APIBackend resolves to a callable class factory.""" + from rdagent.oai.llm_utils import APIBackend + assert callable(APIBackend) + + def test_get_api_backend_is_importable(self) -> None: + """get_api_backend is importable.""" + from rdagent.oai.llm_utils import get_api_backend + assert callable(get_api_backend) + + @patch("rdagent.oai.llm_utils.import_class") + def test_get_api_backend_calls_import_class(self, mock_import: MagicMock) -> None: + """get_api_backend uses import_class to resolve backend class.""" + from rdagent.oai.llm_utils import get_api_backend + mock_cls = MagicMock() + mock_cls.return_value = MagicMock() + mock_import.return_value = mock_cls + + backend = get_api_backend(cache_enabled=False) + assert backend is not None + mock_import.assert_called_once() + + @patch("rdagent.oai.llm_utils.import_class") + def test_api_backend_passes_args(self, mock_import: MagicMock) -> None: + """APIBackend passes args to the backend constructor.""" + from rdagent.oai.llm_utils import get_api_backend + mock_cls = MagicMock() + mock_import.return_value = mock_cls + + get_api_backend(use_chat_cache=True, json_mode=True) + mock_cls.assert_called_once_with(use_chat_cache=True, json_mode=True) + + def test_api_backend_reference_equality(self) -> None: + """APIBackend and get_api_backend are the same object.""" + from rdagent.oai.llm_utils import APIBackend, get_api_backend + assert APIBackend is get_api_backend + + +# ============================================================================= +# LLM settings +# ============================================================================= + + +class TestLLMSettings: + """Tests for LLM settings module.""" + + def test_llm_settings_is_importable(self) -> None: + """LLM_SETTINGS is importable.""" + from rdagent.oai.llm_conf import LLM_SETTINGS + assert LLM_SETTINGS is not None + + def test_llm_settings_has_backend(self) -> None: + """LLM_SETTINGS has backend attribute.""" + from rdagent.oai.llm_conf import LLM_SETTINGS + assert hasattr(LLM_SETTINGS, "backend") + + def test_llm_settings_backend_is_string(self) -> None: + """LLM_SETTINGS.backend is a string class path.""" + from rdagent.oai.llm_conf import LLM_SETTINGS + assert isinstance(LLM_SETTINGS.backend, str) + + +# ============================================================================= +# md5_hash utility +# ============================================================================= + + +class TestMd5Hash: + """Tests for md5_hash utility.""" + + def test_md5_hash_is_importable(self) -> None: + """md5_hash is importable.""" + from rdagent.utils import md5_hash + assert callable(md5_hash) + + def test_md5_hash_returns_string(self) -> None: + """md5_hash returns a hex digest string.""" + from rdagent.utils import md5_hash + result = md5_hash("test input") + assert isinstance(result, str) + assert len(result) == 64 # SHA256 hex digest (named md5 but uses sha256) + + def test_md5_hash_deterministic(self) -> None: + """md5_hash is deterministic.""" + from rdagent.utils import md5_hash + a = md5_hash("hello") + b = md5_hash("hello") + assert a == b + + def test_md5_hash_different_inputs(self) -> None: + """Different inputs produce different hashes.""" + from rdagent.utils import md5_hash + a = md5_hash("hello") + b = md5_hash("world") + assert a != b + + @pytest.mark.parametrize("input_val", [ + "", "a", "abc" * 1000, "unicode_日本語", "emoji_🌟", "multi\nline\nstring", + ]) + def test_md5_hash_various_inputs(self, input_val: str) -> None: + """Various input types produce valid hashes.""" + from rdagent.utils import md5_hash + result = md5_hash(input_val) + assert isinstance(result, str) + assert len(result) == 64 + + +# ============================================================================= +# Integration tests — end-to-end mocked embedding pipeline +# ============================================================================= + + +class TestEmbeddingPipeline: + """Integration-style tests for the embedding pipeline (mocked).""" + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_knowledge_base_typical_usage(self, mock_api: MagicMock) -> None: + """Typical usage pattern: query similarity of task vs known successes.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + + target_task = "Calculate rolling z-score of $close for EURUSD" + success_tasks = [ + "Calculate SMA of $close", + "Calculate volatility of returns", + "Compute volume-weighted average price", + ] + + # Mock embeddings: first target, then three successes + mock_api.return_value.create_embedding.return_value = [ + [0.3, 0.7, 0.1, 0.5], + [0.4, 0.6, 0.2, 0.4], + [0.1, 0.8, 0.0, 0.5], + [0.2, 0.9, 0.1, 0.3], + ] + + similarity = calculate_embedding_distance_between_str_list( + [target_task], success_tasks, + ) + assert len(similarity) == 1 + assert len(similarity[0]) == 3 + + # Sort by similarity descending + similar_indexes = sorted( + range(len(similarity[0])), + key=lambda i: similarity[0][i], + reverse=True, + ) + assert len(similar_indexes) == 3 + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_embedding_concatenation_order(self, mock_api: MagicMock) -> None: + """Source embeddings are first, then target embeddings.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + + mock_api.return_value.create_embedding.return_value = [ + [1.0, 0.0], # source + [0.0, 1.0], # target + ] + + result = calculate_embedding_distance_between_str_list(["s"], ["t"]) + assert result[0][0] == pytest.approx(0.0, abs=1e-9) + + +# ============================================================================= +# Edge cases — NaN, inf, extreme values in embedding vectors +# ============================================================================= + + +class TestEmbeddingEdgeCases: + """Edge case tests for embedding distance calculation.""" + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_nan_in_embeddings(self, mock_api: MagicMock) -> None: + """NaN values in embeddings produce NaN in similarity.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [ + [float("nan"), 1.0], + [1.0, 0.0], + ] + result = calculate_embedding_distance_between_str_list(["s"], ["t"]) + assert isinstance(result[0][0], float) + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_inf_in_embeddings(self, mock_api: MagicMock) -> None: + """Inf values in embeddings produce NaN or inf in similarity.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + mock_api.return_value.create_embedding.return_value = [ + [float("inf"), 0.0], + [1.0, 0.0], + ] + result = calculate_embedding_distance_between_str_list(["s"], ["t"]) + assert isinstance(result[0][0], float) + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_very_high_dimensional_embedding(self, mock_api: MagicMock) -> None: + """High-dimensional embeddings (1536 dims) work.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + dim = 1536 + mock_api.return_value.create_embedding.return_value = [ + list(np.random.randn(dim)), + list(np.random.randn(dim)), + ] + result = calculate_embedding_distance_between_str_list(["s"], ["t"]) + assert len(result[0]) == 1 + assert -1.0 <= result[0][0] <= 1.0 + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_many_targets(self, mock_api: MagicMock) -> None: + """Large number of targets works correctly.""" + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + n_targets = 100 + mock_api.return_value.create_embedding.return_value = [ + list(np.random.randn(128)) + for _ in range(1 + n_targets) + ] + sources = ["s"] + targets = [f"t{i}" for i in range(n_targets)] + result = calculate_embedding_distance_between_str_list(sources, targets) + assert len(result) == 1 + assert len(result[0]) == n_targets + + +# ============================================================================= +# Backend base class +# ============================================================================= + + +class TestBackendBase: + """Tests for the backend base class.""" + + def test_base_api_backend_is_importable(self) -> None: + """BaseAPIBackend is importable.""" + from rdagent.oai.backend.base import APIBackend + assert APIBackend is not None + + def test_base_api_backend_is_a_class(self) -> None: + """BaseAPIBackend is a class.""" + from rdagent.oai.backend.base import APIBackend + assert isinstance(APIBackend, type) + + +# ============================================================================= +# Pickle safety for LLM-related objects +# ============================================================================= + + +class TestLLMPickleSafety: + """Pickle safety tests for LLM utility objects.""" + + def test_similarity_matrix_pickle(self) -> None: + """Similarity matrix (list of lists) survives pickle.""" + matrix = [[0.5, 0.8], [0.3, 0.1]] + data = pickle.dumps(matrix) + loaded = pickle.loads(data) + assert loaded == matrix + + def test_embedding_list_pickle(self) -> None: + """Embedding vector list survives pickle.""" + emb = [0.1, 0.2, 0.3, 0.4] + data = pickle.dumps(emb) + loaded = pickle.loads(data) + assert loaded == emb + + @patch("rdagent.oai.llm_utils.APIBackend") + def test_mocked_api_result_pickle(self, mock_api: MagicMock) -> None: + """Mocked API result (list of floats) survives pickle.""" + mock_result = [[0.1, 0.2], [0.3, 0.4]] + data = pickle.dumps(mock_result) + loaded = pickle.loads(data) + assert loaded == mock_result diff --git a/test/qlib/test_core_deep.py b/test/qlib/test_core_deep.py new file mode 100644 index 00000000..26a8dbab --- /dev/null +++ b/test/qlib/test_core_deep.py @@ -0,0 +1,520 @@ +"""Deep tests for rdagent.core: developer.py, evaluation.py, and related abstractions.""" + +from __future__ import annotations + +import pickle +import sys +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +PROJECT_ROOT = Path(__file__).parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + + +# ============================================================================= +# Import safety +# ============================================================================= + +CORE_MODULES = [ + "rdagent.core.developer", + "rdagent.core.evaluation", + "rdagent.core.experiment", + "rdagent.core.proposal", + "rdagent.core.scenario", + "rdagent.core.evolving_framework", + "rdagent.core.evolving_agent", + "rdagent.core.conf", + "rdagent.core.exception", + "rdagent.core.utils", +] + + +class TestCoreModuleImports: + @pytest.mark.parametrize("module_name", CORE_MODULES) + def test_module_importable(self, module_name: str) -> None: + import importlib + mod = importlib.import_module(module_name) + assert mod is not None + + +# ============================================================================= +# Feedback +# ============================================================================= + + +class TestFeedback: + def test_default_is_acceptable_returns_true(self) -> None: + from rdagent.core.evaluation import Feedback + fb = Feedback() + assert fb.is_acceptable() is True + + def test_default_finished_returns_true(self) -> None: + from rdagent.core.evaluation import Feedback + fb = Feedback() + assert fb.finished() is True + + def test_default_bool_is_true(self) -> None: + from rdagent.core.evaluation import Feedback + fb = Feedback() + assert bool(fb) is True + + def test_is_acceptable_calls_bool(self) -> None: + from rdagent.core.evaluation import Feedback + + class FalseFeedback(Feedback): + def __bool__(self) -> bool: + return False + + fb = FalseFeedback() + assert fb.is_acceptable() is False + + def test_finished_can_be_overridden(self) -> None: + from rdagent.core.evaluation import Feedback + + class CustomFinish(Feedback): + def __bool__(self) -> bool: + return False + + def finished(self) -> bool: + return True + + fb = CustomFinish() + assert fb.finished() is True + assert bool(fb) is False + + def test_pickle_safety(self) -> None: + from rdagent.core.evaluation import Feedback + fb = Feedback() + data = pickle.dumps(fb) + fb2 = pickle.loads(data) + assert isinstance(fb2, Feedback) + assert bool(fb2) is True + + +# ============================================================================= +# Evaluator / EvaluableObj +# ============================================================================= + + +class TestEvaluator: + def test_evaluator_is_abstract(self) -> None: + from rdagent.core.evaluation import Evaluator + assert hasattr(Evaluator, "evaluate") + + def test_concrete_evaluator_must_implement_evaluate(self) -> None: + from rdagent.core.evaluation import Evaluator, Feedback + + class ConcreteEvaluator(Evaluator): + def evaluate(self, eo) -> Feedback: + return Feedback() + + ev = ConcreteEvaluator() + result = ev.evaluate(None) + assert isinstance(result, Feedback) + + @pytest.mark.parametrize("input_eo", [None, "string", {"key": "value"}, [1, 2, 3]]) + def test_concrete_evaluator_accepts_any_input(self, input_eo: Any) -> None: + from rdagent.core.evaluation import Evaluator, Feedback + + class FlexibleEvaluator(Evaluator): + def evaluate(self, eo) -> Feedback: + return Feedback() + + ev = FlexibleEvaluator() + result = ev.evaluate(input_eo) + assert isinstance(result, Feedback) + + +class TestEvaluableObj: + def test_evaluable_obj_is_importable(self) -> None: + from rdagent.core.evaluation import EvaluableObj + assert EvaluableObj is not None + + def test_evaluable_obj_can_be_instantiated(self) -> None: + from rdagent.core.evaluation import EvaluableObj + obj = EvaluableObj() + assert isinstance(obj, EvaluableObj) + + +# ============================================================================= +# Developer +# ============================================================================= + + +class TestDeveloperBase: + def test_developer_is_importable(self) -> None: + from rdagent.core.developer import Developer + assert Developer is not None + + def test_developer_stores_scenario(self) -> None: + from rdagent.core.developer import Developer + from rdagent.core.experiment import ASpecificExp + + class ConcreteDev(Developer[ASpecificExp]): + def develop(self, exp: ASpecificExp) -> ASpecificExp: + return exp + + scen = MagicMock() + dev = ConcreteDev(scen) + assert dev.scen is scen + + def test_develop_modifies_in_place(self) -> None: + from rdagent.core.developer import Developer + from rdagent.core.experiment import ASpecificExp + + class ModifyingDev(Developer[ASpecificExp]): + def develop(self, exp: ASpecificExp) -> ASpecificExp: + exp._modified = True + return exp + + dev = ModifyingDev(MagicMock()) + exp = MagicMock() + result = dev.develop(exp) + assert result._modified is True + + +# ============================================================================= +# Experiment classes +# ============================================================================= + + +class TestExperiment: + def test_task_has_get_task_information(self) -> None: + from rdagent.core.experiment import Task + assert hasattr(Task, "get_task_information") + + def test_workspace_has_execute(self) -> None: + from rdagent.core.experiment import Workspace + assert hasattr(Workspace, "execute") + + def test_fb_workspace_inject_files(self) -> None: + from rdagent.core.experiment import FBWorkspace + ws = FBWorkspace() + ws.inject_files(**{"factor.py": "def calc(): pass"}) + code = ws.all_codes + assert "def calc" in code + + def test_fb_workspace_copy_returns_new_instance(self) -> None: + from rdagent.core.experiment import FBWorkspace + ws = FBWorkspace() + ws.inject_files(**{"test.py": "x=1"}) + ws2 = ws.copy() + assert ws2 is not ws + assert ws2.all_codes == ws.all_codes + + def test_fb_workspace_pickle_safety(self) -> None: + from rdagent.core.experiment import FBWorkspace + ws = FBWorkspace() + ws.inject_files(**{"factor.py": "x=1", "utils.py": "y=2"}) + data = pickle.dumps(ws) + ws2 = pickle.loads(data) + assert isinstance(ws2, FBWorkspace) + + @pytest.mark.parametrize("files", [ + {}, + {"a.py": ""}, + {"a.py": "x=1", "b.py": "y=2", "c.py": "z=3"}, + ]) + def test_fb_workspace_file_variants(self, files: dict) -> None: + from rdagent.core.experiment import FBWorkspace + ws = FBWorkspace() + ws.inject_files(**files) + assert isinstance(ws.all_codes, str) + + def test_aspecific_exp_is_importable(self) -> None: + from rdagent.core.experiment import ASpecificExp + assert ASpecificExp is not None + + +# ============================================================================= +# EvoStep +# ============================================================================= + + +class TestEvoStep: + def test_default_construction(self) -> None: + from rdagent.core.evolving_framework import EvoStep + es = EvoStep(evolvable_subjects="evo") + assert es.evolvable_subjects == "evo" + assert es.queried_knowledge is None + assert es.feedback is None + + def test_full_construction(self) -> None: + from rdagent.core.evolving_framework import EvoStep, QueriedKnowledge + qk = QueriedKnowledge() + es = EvoStep(evolvable_subjects="subj", queried_knowledge=qk, feedback="fb") + assert es.queried_knowledge is qk + assert es.feedback == "fb" + + def test_equality_by_reference(self) -> None: + from rdagent.core.evolving_framework import EvoStep + es1 = EvoStep(evolvable_subjects="a") + es2 = EvoStep(evolvable_subjects="a") + assert es1 == es2 + + def test_pickle_safety(self) -> None: + from rdagent.core.evolving_framework import EvoStep + es = EvoStep(evolvable_subjects="subj", feedback="good") + data = pickle.dumps(es) + es2 = pickle.loads(data) + assert es2.evolvable_subjects == "subj" + assert es2.feedback == "good" + + def test_with_none_values(self) -> None: + from rdagent.core.evolving_framework import EvoStep + es = EvoStep(evolvable_subjects=None, queried_knowledge=None, feedback=None) + assert es.evolvable_subjects is None + + +# ============================================================================= +# Knowledge and QueriedKnowledge +# ============================================================================= + + +class TestKnowledge: + def test_knowledge_is_importable(self) -> None: + from rdagent.core.evolving_framework import Knowledge + assert Knowledge is not None + + def test_knowledge_can_be_instantiated(self) -> None: + from rdagent.core.evolving_framework import Knowledge + k = Knowledge() + assert isinstance(k, Knowledge) + + def test_knowledge_pickle_safety(self) -> None: + from rdagent.core.evolving_framework import Knowledge + k = Knowledge() + data = pickle.dumps(k) + k2 = pickle.loads(data) + assert isinstance(k2, Knowledge) + + +class TestQueriedKnowledge: + def test_default_construction(self) -> None: + from rdagent.core.evolving_framework import QueriedKnowledge + qk = QueriedKnowledge() + assert isinstance(qk, QueriedKnowledge) + + def test_pickle_safety(self) -> None: + from rdagent.core.evolving_framework import QueriedKnowledge + qk = QueriedKnowledge() + data = pickle.dumps(qk) + qk2 = pickle.loads(data) + assert isinstance(qk2, QueriedKnowledge) + + +# ============================================================================= +# RAGStrategy / RAGEvaluator +# ============================================================================= + + +class TestRAGStrategy: + def test_rag_strategy_has_methods(self) -> None: + from rdagent.core.evolving_framework import RAGStrategy + assert hasattr(RAGStrategy, "generate_knowledge") + assert hasattr(RAGStrategy, "query") + + +class TestRAGEvaluator: + def test_rage_evaluator_is_importable(self) -> None: + from rdagent.core.evolving_agent import RAGEvaluator + assert RAGEvaluator is not None + + +# ============================================================================= +# EvolvingKnowledgeBase +# ============================================================================= + + +class TestEvolvingKnowledgeBase: + def test_has_query_method(self) -> None: + from rdagent.core.evolving_framework import EvolvingKnowledgeBase + assert hasattr(EvolvingKnowledgeBase, "query") + + def test_takes_optional_path_argument(self) -> None: + from rdagent.core.evolving_framework import EvolvingKnowledgeBase + kb = EvolvingKnowledgeBase(path=Path("/tmp/test")) + assert kb.path == Path("/tmp/test") + + +# ============================================================================= +# EvolvableSubjects +# ============================================================================= + + +class TestEvolvableSubjects: + def test_evolvable_subjects_is_importable(self) -> None: + from rdagent.core.evolving_framework import EvolvableSubjects + assert EvolvableSubjects is not None + + def test_evolvable_subjects_has_clone_method(self) -> None: + from rdagent.core.evolving_framework import EvolvableSubjects + assert hasattr(EvolvableSubjects, "clone") + + def test_evolvable_subjects_is_instantiable(self) -> None: + from rdagent.core.evolving_framework import EvolvableSubjects + es = EvolvableSubjects() + assert es is not None + + +# ============================================================================= +# Scenario +# ============================================================================= + + +class TestScenario: + def test_scenario_is_abstract(self) -> None: + from rdagent.core.scenario import Scenario + assert hasattr(Scenario, "__abstractmethods__") + + def test_scenario_has_rich_style_description(self) -> None: + from rdagent.core.scenario import Scenario + assert hasattr(Scenario, "rich_style_description") + + def test_scenario_has_background(self) -> None: + from rdagent.core.scenario import Scenario + assert hasattr(Scenario, "background") + + def test_scenario_source_data_default(self) -> None: + from rdagent.core.scenario import Scenario + + class NoDataScen(Scenario): + @property + def background(self) -> str: return "bg" + @property + def rich_style_description(self) -> str: return "rd" + def get_scenario_all_desc(self, **kw) -> str: return "ad" + def get_runtime_environment(self) -> str: return "re" + + scen = NoDataScen() + assert scen.source_data == "" + + +# ============================================================================= +# Exception classes +# ============================================================================= + + +class TestExceptionClasses: + def test_llm_unavailable_error_is_exception(self) -> None: + from rdagent.core.exception import LLMUnavailableError + assert issubclass(LLMUnavailableError, Exception) + + def test_llm_unavailable_error_string_message(self) -> None: + from rdagent.core.exception import LLMUnavailableError + with pytest.raises(LLMUnavailableError, match="test message"): + raise LLMUnavailableError("test message") + + def test_llm_unavailable_error_pickle(self) -> None: + from rdagent.core.exception import LLMUnavailableError + e = LLMUnavailableError("pickle me") + data = pickle.dumps(e) + e2 = pickle.loads(data) + assert str(e2) == "pickle me" + + @pytest.mark.parametrize("message", ["", "short", "multi\nline\nmessage"]) + def test_llm_unavailable_error_message_variants(self, message: str) -> None: + from rdagent.core.exception import LLMUnavailableError + e = LLMUnavailableError(message) + assert str(e) == message + + +# ============================================================================= +# Conf module +# ============================================================================= + + +class TestConfModule: + def test_rd_agent_settings_is_importable(self) -> None: + from rdagent.core.conf import RD_AGENT_SETTINGS + assert RD_AGENT_SETTINGS is not None + + def test_rd_agent_settings_has_multi_proc_n(self) -> None: + from rdagent.core.conf import RD_AGENT_SETTINGS + assert hasattr(RD_AGENT_SETTINGS, "multi_proc_n") + + def test_rd_agent_settings_get_max_parallel(self) -> None: + from rdagent.core.conf import RD_AGENT_SETTINGS + result = RD_AGENT_SETTINGS.get_max_parallel() + assert isinstance(result, int) + + +# ============================================================================= +# Utils module +# ============================================================================= + + +class TestCoreUtils: + def test_import_class_with_valid_path(self) -> None: + from rdagent.core.utils import import_class + cls = import_class("rdagent.core.evaluation.Feedback") + from rdagent.core.evaluation import Feedback + assert cls is Feedback + + def test_import_class_raises_on_invalid_path(self) -> None: + from rdagent.core.utils import import_class + with pytest.raises((ValueError, ImportError, ModuleNotFoundError)): + import_class("rdagent.nonexistent.Class") + + +# ============================================================================= +# Pickle safety +# ============================================================================= + + +class TestPickleSafety: + def test_feedback_list_picklable(self) -> None: + from rdagent.core.evaluation import Feedback + items = [Feedback(), Feedback(), Feedback()] + data = pickle.dumps(items) + loaded = pickle.loads(data) + assert [bool(x) for x in loaded] == [True, True, True] + + def test_evo_step_with_none_feedback_pickle(self) -> None: + from rdagent.core.evolving_framework import EvoStep + es = EvoStep(evolvable_subjects="s", queried_knowledge=None, feedback=None) + data = pickle.dumps(es) + es2 = pickle.loads(data) + assert es2.feedback is None + + +# ============================================================================= +# Edge cases +# ============================================================================= + + +class TestEdgeCases: + def test_evo_step_with_all_none(self) -> None: + from rdagent.core.evolving_framework import EvoStep + es = EvoStep(evolvable_subjects=None, queried_knowledge=None, feedback=None) + assert es.evolvable_subjects is None + + def test_feedback_bool_edge_cases(self) -> None: + from rdagent.core.evaluation import Feedback + + class AlwaysTrue(Feedback): + def __bool__(self): return True + + class AlwaysFalse(Feedback): + def __bool__(self): return False + + assert bool(AlwaysTrue()) is True + assert bool(AlwaysFalse()) is False + + +# ============================================================================= +# Generic types +# ============================================================================= + + +class TestGenericTypes: + def test_aspecific_exp_importable(self) -> None: + from rdagent.core.experiment import ASpecificExp + assert ASpecificExp is not None + + def test_developer_is_importable(self) -> None: + from rdagent.core.developer import Developer + assert Developer is not None diff --git a/test/qlib/test_costeer_deep.py b/test/qlib/test_costeer_deep.py new file mode 100644 index 00000000..7afae149 --- /dev/null +++ b/test/qlib/test_costeer_deep.py @@ -0,0 +1,893 @@ +"""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 diff --git a/test/qlib/test_workflow_deep.py b/test/qlib/test_workflow_deep.py new file mode 100644 index 00000000..5e0b254b --- /dev/null +++ b/test/qlib/test_workflow_deep.py @@ -0,0 +1,718 @@ +"""Deep tests for workflow components: rd_loop.py, proposal, trace, hypothesis systems.""" + +from __future__ import annotations + +import asyncio +import pickle +import sys +from multiprocessing import Queue +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +PROJECT_ROOT = Path(__file__).parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + + +def _make_mock_prop_setting(**overrides: Any) -> Any: + ps = MagicMock() + ps.scen = "rdagent.scenarios.qlib.scenario.QlibQuantScenario" + for k, v in overrides.items(): + setattr(ps, k, v) + ps.model_dump.return_value = {} + return ps + + +# ============================================================================= +# Import safety +# ============================================================================= + +WORKFLOW_MODULES = [ + "rdagent.components.workflow.rd_loop", + "rdagent.components.workflow.conf", + "rdagent.core.proposal", + "rdagent.core.developer", + "rdagent.core.experiment", + "rdagent.core.scenario", + "rdagent.core.evolving_framework", + "rdagent.core.evolving_agent", + "rdagent.core.utils", + "rdagent.utils.workflow", + "rdagent.utils.qlib", +] + + +class TestWorkflowImports: + @pytest.mark.parametrize("module_name", WORKFLOW_MODULES) + def test_module_importable(self, module_name: str) -> None: + import importlib + mod = importlib.import_module(module_name) + assert mod is not None + + +# ============================================================================= +# LoopBase and LoopMeta +# ============================================================================= + + +class TestLoopBase: + def test_loop_base_is_importable(self) -> None: + from rdagent.utils.workflow import LoopBase + assert LoopBase is not None + + def test_loop_meta_is_importable(self) -> None: + from rdagent.utils.workflow import LoopMeta + assert LoopMeta is not None + + def test_loop_base_can_be_instantiated(self) -> None: + from rdagent.utils.workflow import LoopBase + loop = LoopBase() + assert loop is not None + + +# ============================================================================= +# RDLoop — construction +# ============================================================================= + + +class TestRDLoopConstruction: + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_init_imports_scenario(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + mock_scen = MagicMock() + mock_import.return_value = mock_scen + props = _make_mock_prop_setting() + loop = RDLoop(props) + assert loop.trace is not None + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_init_creates_trace(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + mock_import.return_value = MagicMock() + props = _make_mock_prop_setting() + loop = RDLoop(props) + assert hasattr(loop, "trace") + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_init_sets_experiment_plan(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + mock_import.return_value = MagicMock() + props = _make_mock_prop_setting() + loop = RDLoop(props) + assert "features" in loop.plan + assert "feature_codes" in loop.plan + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_init_with_hypothesis_gen_setting(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + mock_import.return_value = MagicMock() + props = _make_mock_prop_setting(hypothesis_gen="some.path.ClassName") + loop = RDLoop(props) + assert loop.hypothesis_gen is not None + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_init_without_hypothesis_gen_setting(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + mock_import.return_value = MagicMock() + props = _make_mock_prop_setting() + props.hypothesis_gen = None + loop = RDLoop(props) + assert loop.hypothesis_gen is None + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_init_with_coder_setting(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + mock_import.return_value = MagicMock() + props = _make_mock_prop_setting(coder="some.path.Coder") + loop = RDLoop(props) + assert loop.coder is not None + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_init_with_runner_setting(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + mock_import.return_value = MagicMock() + props = _make_mock_prop_setting(runner="some.path.Runner") + loop = RDLoop(props) + assert loop.runner is not None + + +# ============================================================================= +# RDLoop — step methods +# ============================================================================= + + +class TestRDLoopPropose: + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_propose_returns_hypothesis(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + from rdagent.core.proposal import Hypothesis + mock_scen = MagicMock() + mock_import.return_value = mock_scen + props = _make_mock_prop_setting(hypothesis_gen="some.path") + loop = RDLoop(props) + mock_hypo = Hypothesis(hypothesis="test", reason="test", + concise_reason="cr", concise_observation="co", + concise_justification="cj", concise_knowledge="ck") + loop.hypothesis_gen = MagicMock() + loop.hypothesis_gen.gen.return_value = mock_hypo + result = loop._propose() + assert result == mock_hypo + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_propose_raises_loop_resume_on_llm_error(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + from rdagent.core.exception import LLMUnavailableError + mock_scen = MagicMock() + mock_import.return_value = mock_scen + props = _make_mock_prop_setting(hypothesis_gen="some.path") + loop = RDLoop(props) + loop.hypothesis_gen = MagicMock() + loop.hypothesis_gen.gen.side_effect = LLMUnavailableError("timeout") + with pytest.raises(loop.LoopResumeError): + loop._propose() + + +class TestRDLoopExpGen: + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_exp_gen_returns_experiment(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + from rdagent.core.proposal import Hypothesis + mock_scen = MagicMock() + mock_import.return_value = mock_scen + props = _make_mock_prop_setting(hypothesis2experiment="some.path") + loop = RDLoop(props) + mock_exp = MagicMock() + loop.hypothesis2experiment = MagicMock() + loop.hypothesis2experiment.convert.return_value = mock_exp + hypo = Hypothesis(hypothesis="h", reason="r", + concise_reason="cr", concise_observation="co", + concise_justification="cj", concise_knowledge="ck") + result = loop._exp_gen(hypo) + assert result == mock_exp + + +class TestRDLoopSteps: + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_direct_exp_gen_yields_dict(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + mock_scen = MagicMock() + mock_scen.patcher = None + mock_import.return_value = mock_scen + props = _make_mock_prop_setting(hypothesis_gen="p.HG", hypothesis2experiment="p.H2E") + loop = RDLoop(props) + mock_hypo = MagicMock() + mock_hypo.action = "factor" + loop.hypothesis_gen = MagicMock() + loop.hypothesis_gen.gen.return_value = mock_hypo + loop.hypothesis2experiment = MagicMock() + mock_exp = MagicMock() + mock_exp.sub_tasks = [] + mock_exp.based_experiments = None + loop.hypothesis2experiment.convert.return_value = mock_exp + result = asyncio.run(loop.direct_exp_gen({})) + assert "propose" in result + assert "exp_gen" in result + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_coding_calls_coder_develop(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + mock_scen = MagicMock() + mock_import.return_value = mock_scen + props = _make_mock_prop_setting(coder="p.Coder") + loop = RDLoop(props) + loop.coder = MagicMock() + loop.coder.develop.return_value = MagicMock() + prev_out = {"direct_exp_gen": {"exp_gen": MagicMock()}} + loop.coding(prev_out) + assert loop.coder.develop.called + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_running_calls_runner_develop(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + mock_scen = MagicMock() + mock_import.return_value = mock_scen + props = _make_mock_prop_setting(runner="p.Runner") + loop = RDLoop(props) + loop.runner = MagicMock() + loop.runner.develop.return_value = MagicMock() + prev_out = {"coding": MagicMock()} + loop.running(prev_out) + assert loop.runner.develop.called + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_feedback_on_exception_returns_reject_feedback(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + from rdagent.core.proposal import HypothesisFeedback + mock_scen = MagicMock() + mock_import.return_value = mock_scen + props = _make_mock_prop_setting(summarizer="p.Summarizer") + loop = RDLoop(props) + prev_out = {loop.EXCEPTION_KEY: "test error"} + result = loop.feedback(prev_out) + assert isinstance(result, HypothesisFeedback) + assert result.decision is False + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_feedback_normal_path_calls_summarizer(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + from rdagent.core.proposal import HypothesisFeedback + mock_scen = MagicMock() + mock_import.return_value = mock_scen + props = _make_mock_prop_setting(summarizer="p.Summarizer") + loop = RDLoop(props) + loop.summarizer = MagicMock() + loop.summarizer.generate_feedback.return_value = HypothesisFeedback( + reason="ok", decision=True, code_change_summary="done", acceptable=True) + prev_out = {"running": MagicMock()} + result = loop.feedback(prev_out) + assert isinstance(result, HypothesisFeedback) + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_record_syncs_trace_dag(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + mock_scen = MagicMock() + mock_import.return_value = mock_scen + props = _make_mock_prop_setting() + loop = RDLoop(props) + loop.trace = MagicMock() + mock_exp = MagicMock() + mock_exp.hypothesis = "hypo" + mock_fb = MagicMock() + prev_out = {"feedback": mock_fb, "running": mock_exp, loop.LOOP_IDX_KEY: 0} + loop.record(prev_out) + loop.trace.sync_dag_parent_and_hist.assert_called_once() + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_record_with_none_exp_does_not_crash(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + mock_scen = MagicMock() + mock_import.return_value = mock_scen + props = _make_mock_prop_setting() + loop = RDLoop(props) + loop.trace = MagicMock() + prev_out = {"feedback": MagicMock(), "running": MagicMock(hypothesis=None), loop.LOOP_IDX_KEY: 0} + loop.record(prev_out) + loop.trace.sync_dag_parent_and_hist.assert_not_called() + + +# ============================================================================= +# RDLoop — interaction methods +# ============================================================================= + + +class TestRDLoopInteractions: + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_set_interactor_stores_queues(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + mock_scen = MagicMock() + mock_import.return_value = mock_scen + props = _make_mock_prop_setting() + loop = RDLoop(props) + q1, q2 = Queue(), Queue() + loop._set_interactor(q1, q2) + assert loop.user_request_q is q1 + assert loop.user_response_q is q2 + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_interact_hypo_no_queues_returns_original(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + from rdagent.core.proposal import Hypothesis + mock_scen = MagicMock() + mock_import.return_value = mock_scen + props = _make_mock_prop_setting() + loop = RDLoop(props) + hypo = Hypothesis(hypothesis="h", reason="r", + concise_reason="cr", concise_observation="co", + concise_justification="cj", concise_knowledge="ck") + result = loop._interact_hypo(hypo) + assert result is hypo + + @patch("rdagent.components.workflow.rd_loop.logger.log_object") + @patch("rdagent.components.workflow.rd_loop.import_class") + def test_interact_feedback_no_queues_returns_original(self, mock_import: MagicMock, mock_log: MagicMock) -> None: + from rdagent.components.workflow.rd_loop import RDLoop + from rdagent.core.proposal import HypothesisFeedback + mock_scen = MagicMock() + mock_import.return_value = mock_scen + props = _make_mock_prop_setting() + loop = RDLoop(props) + fb = HypothesisFeedback(reason="r", decision=True, code_change_summary="ok", acceptable=True) + result = loop._interact_feedback(fb) + assert result is fb + + +# ============================================================================= +# BasePropSetting +# ============================================================================= + + +class TestBasePropSetting: + def test_base_prop_setting_is_pydantic_model(self) -> None: + from rdagent.components.workflow.conf import BasePropSetting + from pydantic import BaseModel + assert issubclass(BasePropSetting, BaseModel) + + +# ============================================================================= +# Hypothesis and HypothesisFeedback +# ============================================================================= + + +class TestHypothesis: + def test_construction_with_minimal_fields(self) -> None: + from rdagent.core.proposal import Hypothesis + h = Hypothesis(hypothesis="test", reason="because", + concise_reason="cr", concise_observation="co", + concise_justification="cj", concise_knowledge="ck") + assert h.hypothesis == "test" + assert h.reason == "because" + + def test_has_conciseness_fields(self) -> None: + from rdagent.core.proposal import Hypothesis + h = Hypothesis(hypothesis="h", reason="r", + concise_reason="cr", concise_observation="co", + concise_justification="cj", concise_knowledge="ck") + assert h.concise_reason == "cr" + assert h.concise_observation == "co" + + def test_pickle_safety(self) -> None: + from rdagent.core.proposal import Hypothesis + h = Hypothesis(hypothesis="h", reason="r", + concise_reason="cr", concise_observation="co", + concise_justification="cj", concise_knowledge="ck") + data = pickle.dumps(h) + h2 = pickle.loads(data) + assert h2.hypothesis == "h" + assert h2.concise_knowledge == "ck" + + def test_dict_conversion(self) -> None: + from rdagent.core.proposal import Hypothesis + h = Hypothesis(hypothesis="h", reason="r", + concise_reason="cr", concise_observation="co", + concise_justification="cj", concise_knowledge="ck") + d = h.__dict__ + h2 = type(h)(**d) + assert h2.hypothesis == h.hypothesis + + +class TestHypothesisFeedback: + def test_construction_with_all_fields(self) -> None: + from rdagent.core.proposal import HypothesisFeedback + fb = HypothesisFeedback(reason="good", decision=True, code_change_summary="fixed", acceptable=True) + assert fb.reason == "good" + assert fb.decision is True + + def test_default_values(self) -> None: + from rdagent.core.proposal import HypothesisFeedback + fb = HypothesisFeedback(reason="reason", decision=False) + assert fb.decision is False + + def test_pickle_safety(self) -> None: + from rdagent.core.proposal import HypothesisFeedback + fb = HypothesisFeedback(reason="r", decision=True, code_change_summary="c", acceptable=True) + data = pickle.dumps(fb) + fb2 = pickle.loads(data) + assert fb2.decision is True + + +# ============================================================================= +# Trace +# ============================================================================= + + +class TestTrace: + def test_trace_construction(self) -> None: + from rdagent.core.proposal import Trace + trace = Trace(scen=None) + assert trace is not None + + def test_trace_has_hist_attribute(self) -> None: + from rdagent.core.proposal import Trace + trace = Trace(scen=None) + assert hasattr(trace, "hist") + assert isinstance(trace.hist, list) + + def test_trace_sync_dag_parent_and_hist(self) -> None: + from rdagent.core.proposal import Trace + trace = Trace(scen=None) + exp = MagicMock() + exp.based_experiments = [] + exp.hypothesis = "hypo" + fb = MagicMock() + trace.sync_dag_parent_and_hist((exp, fb), 0) + assert len(trace.hist) > 0 + + def test_trace_pickle_safety(self) -> None: + from rdagent.core.proposal import Trace + trace = Trace(scen=None) + trace.hist = [("entry",)] + data = pickle.dumps(trace) + trace2 = pickle.loads(data) + assert len(trace2.hist) == 1 + + +# ============================================================================= +# HypothesisGen, Hypothesis2Experiment, Experiment2Feedback +# ============================================================================= + + +class TestProposalClasses: + def test_hypothesis_gen_is_importable(self) -> None: + from rdagent.core.proposal import HypothesisGen + assert HypothesisGen is not None + + def test_hypothesis2experiment_is_importable(self) -> None: + from rdagent.core.proposal import Hypothesis2Experiment + assert Hypothesis2Experiment is not None + + def test_experiment2feedback_is_importable(self) -> None: + from rdagent.core.proposal import Experiment2Feedback + assert Experiment2Feedback is not None + + +# ============================================================================= +# Developer +# ============================================================================= + + +class TestDeveloper: + def test_developer_is_importable(self) -> None: + from rdagent.core.developer import Developer + assert Developer is not None + + def test_developer_stores_scenario(self) -> None: + from rdagent.core.developer import Developer + from rdagent.core.experiment import ASpecificExp + class ConcreteDev(Developer[ASpecificExp]): + def develop(self, exp: ASpecificExp) -> ASpecificExp: + return exp + scen = MagicMock() + dev = ConcreteDev(scen) + assert dev.scen is scen + + +# ============================================================================= +# Scenario base class +# ============================================================================= + + +class TestScenarioBase: + def test_scenario_is_abstract(self) -> None: + from rdagent.core.scenario import Scenario + assert hasattr(Scenario, "__abstractmethods__") + + def test_scenario_has_required_properties(self) -> None: + from rdagent.core.scenario import Scenario + assert hasattr(Scenario, "background") + assert hasattr(Scenario, "rich_style_description") + assert hasattr(Scenario, "source_data") + + +# ============================================================================= +# Qlib utilities +# ============================================================================= + + +class TestQlibUtils: + def test_validate_qlib_features_importable(self) -> None: + from rdagent.utils.qlib import validate_qlib_features + assert callable(validate_qlib_features) + + def test_validate_valid_features(self) -> None: + from rdagent.utils.qlib import validate_qlib_features + result = validate_qlib_features(["$close", "$high / $low"]) + assert isinstance(result, bool) + + def test_validate_empty_list(self) -> None: + from rdagent.utils.qlib import validate_qlib_features + result = validate_qlib_features([]) + assert isinstance(result, bool) + + def test_alpha20_importable(self) -> None: + from rdagent.utils.qlib import ALPHA20 + assert isinstance(ALPHA20, dict) + assert len(ALPHA20) > 0 + + @pytest.mark.parametrize("features", [ + ["$close"], ["$open", "$high", "$low", "$close"], ["$close / $open", "$high - $low"], [], + ]) + def test_validate_qlib_features_variants(self, features: list) -> None: + from rdagent.utils.qlib import validate_qlib_features + result = validate_qlib_features(features) + assert isinstance(result, bool) + + +# ============================================================================= +# Experiment classes +# ============================================================================= + + +class TestExperimentClasses: + def test_task_is_importable(self) -> None: + from rdagent.core.experiment import Task + assert Task is not None + + def test_workspace_is_importable(self) -> None: + from rdagent.core.experiment import Workspace + assert Workspace is not None + + def test_fb_workspace_is_importable(self) -> None: + from rdagent.core.experiment import FBWorkspace + assert FBWorkspace is not None + + def test_fb_workspace_inject_files(self) -> None: + from rdagent.core.experiment import FBWorkspace + ws = FBWorkspace() + ws.inject_files(**{"factor.py": "def calc(): pass"}) + code = ws.all_codes + assert "def calc" in code + + def test_fb_workspace_pickle_safety(self) -> None: + from rdagent.core.experiment import FBWorkspace + ws = FBWorkspace() + ws.inject_files(**{"factor.py": "x=1"}) + data = pickle.dumps(ws) + ws2 = pickle.loads(data) + assert isinstance(ws2, FBWorkspace) + + +# ============================================================================= +# Evolving framework imports +# ============================================================================= + + +class TestEvolvingFrameworkImports: + @pytest.mark.parametrize("cls_name,module_path", [ + ("EvolvableSubjects", "rdagent.core.evolving_framework"), + ("EvolvingKnowledgeBase", "rdagent.core.evolving_framework"), + ("EvoStep", "rdagent.core.evolving_framework"), + ("Knowledge", "rdagent.core.evolving_framework"), + ("QueriedKnowledge", "rdagent.core.evolving_framework"), + ("RAGStrategy", "rdagent.core.evolving_framework"), + ("RAGEvaluator", "rdagent.core.evolving_agent"), + ]) + def test_class_importable(self, cls_name: str, module_path: str) -> None: + import importlib + mod = importlib.import_module(module_path) + assert hasattr(mod, cls_name) + + +# ============================================================================= +# EvoStep — dataclass behavior +# ============================================================================= + + +class TestEvoStep: + def test_default_construction(self) -> None: + 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) -> None: + 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) -> None: + from rdagent.core.evolving_framework import EvoStep + es1 = EvoStep(evolvable_subjects="a") + es2 = EvoStep(evolvable_subjects="a") + assert es1 == es2 + + def test_pickle_safety(self) -> None: + from rdagent.core.evolving_framework import EvoStep + es = EvoStep(evolvable_subjects="subj", feedback="good") + data = pickle.dumps(es) + es2 = pickle.loads(data) + assert es2.evolvable_subjects == "subj" + assert es2.feedback == "good" + + +# ============================================================================= +# import_class utility +# ============================================================================= + + +class TestImportClass: + def test_import_class_is_callable(self) -> None: + from rdagent.core.utils import import_class + assert callable(import_class) + + def test_import_class_resolves_known_class(self) -> None: + from rdagent.core.utils import import_class + cls = import_class("rdagent.core.proposal.Hypothesis") + from rdagent.core.proposal import Hypothesis + assert cls is Hypothesis + + def test_import_class_raises_on_bad_path(self) -> None: + from rdagent.core.utils import import_class + with pytest.raises((ValueError, ImportError, ModuleNotFoundError)): + import_class("nonexistent.module.ClassName") + + +# ============================================================================= +# LLMUnavailableError +# ============================================================================= + + +class TestLLMUnavailableError: + def test_is_importable(self) -> None: + from rdagent.core.exception import LLMUnavailableError + assert issubclass(LLMUnavailableError, Exception) + + def test_can_be_raised_and_caught(self) -> None: + from rdagent.core.exception import LLMUnavailableError + with pytest.raises(LLMUnavailableError): + raise LLMUnavailableError("test error") + + def test_pickle_safety(self) -> None: + from rdagent.core.exception import LLMUnavailableError + e = LLMUnavailableError("pickle me") + data = pickle.dumps(e) + e2 = pickle.loads(data) + assert str(e2) == "pickle me" + + +# ============================================================================= +# Pickle safety for combined workflow objects +# ============================================================================= + + +class TestPickleSafetyComposite: + def test_combined_workflow_objects_pickle(self) -> None: + from rdagent.core.proposal import Hypothesis, HypothesisFeedback, Trace + h = Hypothesis(hypothesis="h", reason="r", + concise_reason="cr", concise_observation="co", + concise_justification="cj", concise_knowledge="ck") + fb = HypothesisFeedback(reason="r", decision=True, code_change_summary="ok", acceptable=True) + trace = Trace(scen=None) + trace.hist = [] + bundle = {"hypothesis": h, "feedback": fb, "trace": trace} + data = pickle.dumps(bundle) + bundle2 = pickle.loads(data) + assert bundle2["hypothesis"].hypothesis == "h" diff --git a/test/utils/test_utils_deep.py b/test/utils/test_utils_deep.py new file mode 100644 index 00000000..a40d1a66 --- /dev/null +++ b/test/utils/test_utils_deep.py @@ -0,0 +1,415 @@ +"""Deep tests for rdagent.utils: fmt.py shrink_text and other utility modules.""" + +from __future__ import annotations + +import pickle +import sys +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +PROJECT_ROOT = Path(__file__).parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + + +# ============================================================================= +# Import safety +# ============================================================================= + +UTIL_MODULES = [ + "rdagent.utils", + "rdagent.utils.fmt", + "rdagent.utils.qlib", + "rdagent.utils.env", + "rdagent.utils.workflow", + "rdagent.utils.agent.tpl", +] + + +class TestUtilsImports: + @pytest.mark.parametrize("module_name", UTIL_MODULES) + def test_module_importable(self, module_name: str) -> None: + import importlib + mod = importlib.import_module(module_name) + assert mod is not None + + +# ============================================================================= +# shrink_text +# ============================================================================= + + +class TestShrinkText: + def test_short_text_unchanged(self) -> None: + from rdagent.utils.fmt import shrink_text + result = shrink_text("hello world", context_lines=10, line_len=100) + assert result == "hello world" + + def test_single_line_shorter_than_limit(self) -> None: + from rdagent.utils.fmt import shrink_text + result = shrink_text("abc", context_lines=2, line_len=5) + assert result == "abc" + + def test_multi_line_under_threshold_unchanged(self) -> None: + from rdagent.utils.fmt import shrink_text + text = "line1\nline2\nline3" + result = shrink_text(text, context_lines=5, line_len=50) + assert result == text + + def test_exactly_at_threshold(self) -> None: + from rdagent.utils.fmt import shrink_text + text = "\n".join([f"line{i}" for i in range(4)]) + result = shrink_text(text, context_lines=4, line_len=50) + assert result == text + + def test_more_lines_than_context_shrinks(self) -> None: + from rdagent.utils.fmt import shrink_text + text = "\n".join([f"line{i}" for i in range(100)]) + result = shrink_text(text, context_lines=10, line_len=100) + assert "lines are hidden" in result + + def test_row_shrink_false_preserves_all_lines(self) -> None: + from rdagent.utils.fmt import shrink_text + text = "\n".join([f"line{i}" for i in range(100)]) + result = shrink_text(text, context_lines=5, line_len=100, row_shrink=False) + assert result == text + + def test_col_shrink_long_lines(self) -> None: + from rdagent.utils.fmt import shrink_text + long_line = "x" * 100 + result = shrink_text(long_line, context_lines=5, line_len=20) + assert "chars are hidden" in result + assert len(result) < 100 + + def test_col_shrink_false_preserves_long_lines(self) -> None: + from rdagent.utils.fmt import shrink_text + long_line = "x" * 100 + result = shrink_text(long_line, context_lines=5, line_len=20, col_shrink=False) + assert result == long_line + + def test_both_shrink_disabled(self) -> None: + from rdagent.utils.fmt import shrink_text + text = "x" * 1000 + "\n" + "y" * 1000 + result = shrink_text(text, context_lines=1, line_len=5, row_shrink=False, col_shrink=False) + assert result == text + + def test_first_and_last_lines_preserved(self) -> None: + from rdagent.utils.fmt import shrink_text + text = "\n".join([f"unique_line_{i}" for i in range(100)]) + result = shrink_text(text, context_lines=6, line_len=100) + assert "unique_line_0" in result + assert "unique_line_99" in result + assert "unique_line_50" not in result + + def test_hidden_lines_count_correct(self) -> None: + from rdagent.utils.fmt import shrink_text + total = 100 + ctx = 10 + text = "\n".join([f"L{i}" for i in range(total)]) + result = shrink_text(text, context_lines=ctx, line_len=100) + half = ctx // 2 + hidden = total - half * 2 + assert f"({hidden} lines are hidden)" in result + + def test_empty_string(self) -> None: + from rdagent.utils.fmt import shrink_text + result = shrink_text("", context_lines=5, line_len=10) + assert result == "" + + def test_single_line_with_newline_at_end(self) -> None: + from rdagent.utils.fmt import shrink_text + result = shrink_text("hello\n", context_lines=10, line_len=50) + assert "hello" in result + + def test_all_empty_lines(self) -> None: + from rdagent.utils.fmt import shrink_text + text = "\n".join(["" for _ in range(100)]) + result = shrink_text(text, context_lines=10, line_len=50) + assert isinstance(result, str) + + def test_very_large_context_lines(self) -> None: + from rdagent.utils.fmt import shrink_text + text = "\n".join(["a" for _ in range(50)]) + result = shrink_text(text, context_lines=1000, line_len=10) + assert result == text + + def test_context_lines_of_one(self) -> None: + from rdagent.utils.fmt import shrink_text + text = "line1\nline2\nline3\nline4\nline5" + result = shrink_text(text, context_lines=1, line_len=100) + assert "lines are hidden" in result + + def test_line_len_of_one(self) -> None: + from rdagent.utils.fmt import shrink_text + result = shrink_text("abcdefgh", context_lines=10, line_len=1) + assert "chars are hidden" in result + + def test_line_len_zero(self) -> None: + from rdagent.utils.fmt import shrink_text + result = shrink_text("hello", context_lines=10, line_len=0) + assert "chars are hidden" in result + + def test_returns_string_always(self) -> None: + from rdagent.utils.fmt import shrink_text + for text in ["", "a", "a\nb\nc", "x" * 1000]: + result = shrink_text(text) + assert isinstance(result, str) + + def test_hidden_prefix_format(self) -> None: + from rdagent.utils.fmt import shrink_text + text = "\n".join(["L" for _ in range(100)]) + result = shrink_text(text, context_lines=10, line_len=100) + assert "lines are hidden" in result + assert "..." in result + + @pytest.mark.parametrize("total_lines,ctx", [ + (10, 5), (10, 6), (10, 10), (50, 4), (50, 20), (100, 2), + ]) + def test_various_combinations(self, total_lines: int, ctx: int) -> None: + from rdagent.utils.fmt import shrink_text + text = "\n".join([f"L{i}" for i in range(total_lines)]) + result = shrink_text(text, context_lines=ctx, line_len=100) + assert isinstance(result, str) + assert len(result) > 0 + + @pytest.mark.parametrize("line_len,chars_per_line", [ + (5, 3), (10, 9), (20, 19), (50, 51), + ]) + def test_line_len_vs_chars(self, line_len: int, chars_per_line: int) -> None: + from rdagent.utils.fmt import shrink_text + text = "x" * chars_per_line + result = shrink_text(text, context_lines=5, line_len=line_len) + if chars_per_line > line_len: + assert "chars are hidden" in result + else: + assert result == text + + +# ============================================================================= +# shrink_text — properties +# ============================================================================= + + +class TestShrinkTextProperties: + def test_output_contains_original_when_small(self) -> None: + from rdagent.utils.fmt import shrink_text + lines = ["a", "b", "c", "d", "e"] + text = "\n".join(lines) + result = shrink_text(text, context_lines=len(lines) + 1, line_len=10000) + assert result == text + + def test_shrinking_to_less_lines(self) -> None: + from rdagent.utils.fmt import shrink_text + original = "\n".join([f"line_{i}" for i in range(1000)]) + result = shrink_text(original, context_lines=10, line_len=100) + result_lines = result.split("\n") + assert len(result_lines) < 1000 + + @pytest.mark.parametrize("n_lines", [1, 2, 3, 5, 10]) + def test_various_line_counts(self, n_lines: int) -> None: + from rdagent.utils.fmt import shrink_text + text = "\n".join([f"L{i}" for i in range(n_lines)]) + result = shrink_text(text, context_lines=50, line_len=200) + assert isinstance(result, str) + assert result == text # all fit within context_lines=50 + + +class TestShrinkTextCombinatorial: + @pytest.mark.parametrize("ctx", [0, 1, 2, 5, 10, 50, 100]) + @pytest.mark.parametrize("llen", [0, 1, 5, 10, 50, 200]) + def test_parameter_grid(self, ctx: int, llen: int) -> None: + from rdagent.utils.fmt import shrink_text + text = "x" * 60 + "\n" + "y" * 60 + result = shrink_text(text, context_lines=ctx, line_len=llen) + assert isinstance(result, str) + + @pytest.mark.parametrize("row_shrink", [True, False]) + @pytest.mark.parametrize("col_shrink", [True, False]) + def test_all_shrink_flag_combinations(self, row_shrink: bool, col_shrink: bool) -> None: + from rdagent.utils.fmt import shrink_text + text = "\n".join(["line"] * 200) + result = shrink_text(text, context_lines=5, line_len=50, + row_shrink=row_shrink, col_shrink=col_shrink) + assert isinstance(result, str) + + +# ============================================================================= +# T (template) system +# ============================================================================= + + +class TestTemplateSystem: + def test_t_class_is_importable(self) -> None: + from rdagent.utils.agent.tpl import T + assert T is not None + + def test_t_loads_prompt_template(self) -> None: + from rdagent.utils.agent.tpl import T + tpl = T("scenarios.qlib.prompts:hypothesis_and_feedback") + assert tpl is not None + + def test_t_with_invalid_template_raises(self) -> None: + from rdagent.utils.agent.tpl import T + with pytest.raises(FileNotFoundError): + T("nonexistent.module.path:nonexistent_key") + + @patch("rdagent.utils.agent.tpl.logger") + def test_t_r_method_renders_template(self, mock_logger: MagicMock) -> None: + from rdagent.utils.agent.tpl import T + tpl = T("scenarios.qlib.prompts:hypothesis_and_feedback") + mock_trace = MagicMock() + mock_trace.hist = [] + result = tpl.r(trace=mock_trace) + assert isinstance(result, str) + assert len(result) > 0 + + +# ============================================================================= +# Qlib utilities +# ============================================================================= + + +class TestQlibUtils: + def test_validate_qlib_features_importable(self) -> None: + from rdagent.utils.qlib import validate_qlib_features + assert callable(validate_qlib_features) + + def test_validate_valid_features(self) -> None: + from rdagent.utils.qlib import validate_qlib_features + assert validate_qlib_features(["$close", "$high / $low", "$volume"]) is True + + def test_validate_empty_list(self) -> None: + from rdagent.utils.qlib import validate_qlib_features + result = validate_qlib_features([]) + assert isinstance(result, bool) + + def test_validate_any_expression(self) -> None: + from rdagent.utils.qlib import validate_qlib_features + result = validate_qlib_features(["not_a_real_field_xyz"]) + assert isinstance(result, bool) + + def test_alpha20_importable(self) -> None: + from rdagent.utils.qlib import ALPHA20 + assert isinstance(ALPHA20, dict) + assert len(ALPHA20) > 0 + + @pytest.mark.parametrize("feature_exp", [ + "$close", "$open", "$high", "$low", "$volume", "$vwap", + "$close / $open", "($high - $low) / $open", + ]) + def test_individual_feature_validation(self, feature_exp: str) -> None: + from rdagent.utils.qlib import validate_qlib_features + result = validate_qlib_features([feature_exp]) + assert isinstance(result, bool) + + +# ============================================================================= +# Env utilities +# ============================================================================= + + +class TestEnvUtils: + def test_env_module_is_importable(self) -> None: + from rdagent.utils import env + assert env is not None + + +# ============================================================================= +# md5_hash +# ============================================================================= + + +class TestMd5Hash: + def test_md5_hash_is_function(self) -> None: + from rdagent.utils import md5_hash + assert callable(md5_hash) + + def test_md5_hash_returns_string(self) -> None: + from rdagent.utils import md5_hash + result = md5_hash("test input") + assert isinstance(result, str) + assert len(result) == 64 + + def test_md5_hash_deterministic(self) -> None: + from rdagent.utils import md5_hash + a = md5_hash("hello") + b = md5_hash("hello") + assert a == b + + def test_md5_hash_different_inputs(self) -> None: + from rdagent.utils import md5_hash + a = md5_hash("hello") + b = md5_hash("world") + assert a != b + + @pytest.mark.parametrize("input_val", [ + "", "a", "abc", "multi\nline\nstring", + ]) + def test_md5_hash_various_inputs(self, input_val: str) -> None: + from rdagent.utils import md5_hash + result = md5_hash(input_val) + assert isinstance(result, str) + assert len(result) == 64 + + def test_md5_hash_hex_format(self) -> None: + from rdagent.utils import md5_hash + import re + result = md5_hash("test") + assert re.match(r'^[0-9a-f]{64}$', result) is not None + + +# ============================================================================= +# Workflow utils +# ============================================================================= + + +class TestWorkflowUtils: + def test_loop_base_is_importable(self) -> None: + from rdagent.utils.workflow import LoopBase + assert LoopBase is not None + + def test_loop_meta_is_type(self) -> None: + from rdagent.utils.workflow import LoopMeta + assert isinstance(LoopMeta, type) + + +# ============================================================================= +# Large input stress tests +# ============================================================================= + + +class TestLargeInputs: + def test_ten_thousand_lines(self) -> None: + from rdagent.utils.fmt import shrink_text + text = "\n".join([f"L{i}" for i in range(10000)]) + result = shrink_text(text, context_lines=50, line_len=100) + assert isinstance(result, str) + assert "lines are hidden" in result + + def test_very_long_single_line(self) -> None: + from rdagent.utils.fmt import shrink_text + text = "a" * 100000 + result = shrink_text(text, context_lines=5, line_len=100) + assert "chars are hidden" in result + + +# ============================================================================= +# Pickle safety +# ============================================================================= + + +class TestFmtPickleSafety: + def test_shrunk_text_pickle_safety(self) -> None: + from rdagent.utils.fmt import shrink_text + result = shrink_text("x" * 500, context_lines=5, line_len=10) + data = pickle.dumps(result) + loaded = pickle.loads(data) + assert loaded == result + + def test_alpha20_pickle_safety(self) -> None: + from rdagent.utils.qlib import ALPHA20 + data = pickle.dumps(ALPHA20) + loaded = pickle.loads(data) + assert loaded == ALPHA20