mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-27 23:47:46 +00:00
test: 441 deep property-based tests across CoSTEER, workflow, core, LLM utils, and formatting
- costeer_deep: 112 tests (knowledge base, feedback, evaluators, auto-fixer) - workflow_deep: 84 tests (RDLoop, proposals, traces, hypothesis/pickle) - core_deep: 74 tests (developer, evaluator, exceptions, experiment, scenario) - llm_utils_deep: 49 tests (embeddings, APIBackend, edge cases, Unicode/NaN) - utils_deep: 122 tests (shrink_text, templates, md5_hash, property-based, stress)
This commit is contained in:
TPTBusiness
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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"
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user