diff --git a/test/qlib/test_knowledge_graph.py b/test/qlib/test_knowledge_graph.py new file mode 100644 index 00000000..edcc5938 --- /dev/null +++ b/test/qlib/test_knowledge_graph.py @@ -0,0 +1,112 @@ +"""Tests for knowledge graph and task_loader.""" + +from __future__ import annotations + +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +PROJECT_ROOT = Path(__file__).parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + + +class TestUndirectedNode: + def test_init(self): + from rdagent.components.knowledge_management.graph import UndirectedNode + n = UndirectedNode(content="test", label="component") + assert n.content == "test" + assert n.label == "component" + assert n.id is not None + assert isinstance(n.neighbors, set) + assert len(n.neighbors) == 0 + + def test_add_neighbor_bidirectional(self): + from rdagent.components.knowledge_management.graph import UndirectedNode + a = UndirectedNode(content="a") + b = UndirectedNode(content="b") + a.add_neighbor(b) + assert b in a.neighbors + assert a in b.neighbors + + def test_remove_neighbor(self): + from rdagent.components.knowledge_management.graph import UndirectedNode + a = UndirectedNode(content="a") + b = UndirectedNode(content="b") + a.add_neighbor(b) + a.remove_neighbor(b) + assert b not in a.neighbors + assert a not in b.neighbors + + def test_remove_nonexistent_noop(self): + from rdagent.components.knowledge_management.graph import UndirectedNode + a = UndirectedNode(content="a") + b = UndirectedNode(content="b") + a.remove_neighbor(b) # should not raise + + def test_get_neighbors(self): + from rdagent.components.knowledge_management.graph import UndirectedNode + a = UndirectedNode(content="a") + b = UndirectedNode(content="b") + c = UndirectedNode(content="c") + a.add_neighbor(b) + a.add_neighbor(c) + assert a.get_neighbors() == {b, c} + + def test_rejects_non_string_content(self): + from rdagent.components.knowledge_management.graph import UndirectedNode + with pytest.raises(TypeError, match="string"): + UndirectedNode(content=123) + + def test_string_representation(self): + from rdagent.components.knowledge_management.graph import UndirectedNode + n = UndirectedNode(content="hello", label="test") + s = str(n) + assert "UndirectedNode" in s + assert "hello" in s + assert "test" in s + + +class TestGraphBase: + def test_init_empty(self): + from rdagent.components.knowledge_management.graph import Graph + g = Graph() + assert g.size() == 0 + assert g.get_all_nodes() == [] + + def test_get_node_nonexistent(self): + from rdagent.components.knowledge_management.graph import Graph + g = Graph() + assert g.get_node("nonexistent") is None + + def test_get_all_nodes_by_label_list(self, tmp_path): + from rdagent.components.knowledge_management.graph import Graph, UndirectedNode + g = Graph() + + # Add nodes directly to internal dict + n1 = UndirectedNode(content="a", label="component") + n2 = UndirectedNode(content="b", label="error") + n3 = UndirectedNode(content="c", label="component") + g.nodes[n1.id] = n1 + g.nodes[n2.id] = n2 + g.nodes[n3.id] = n3 + + components = g.get_all_nodes_by_label_list(["component"]) + assert len(components) == 2 + labels = [n.label for n in components] + assert all(l == "component" for l in labels) + + +class TestVectorBase: + def test_cosine_distance_identical_is_zero(self): + from rdagent.components.knowledge_management.vector_base import cosine + import numpy as np + dist = cosine(np.array([1.0, 0.0]), np.array([1.0, 0.0])) + assert dist == pytest.approx(0.0) # cosine distance, not similarity + + def test_cosine_distance_orthogonal(self): + from rdagent.components.knowledge_management.vector_base import cosine + import numpy as np + dist = cosine(np.array([1.0, 0.0]), np.array([0.0, 1.0])) + assert dist == pytest.approx(1.0) diff --git a/test/qlib/test_remaining.py b/test/qlib/test_remaining.py new file mode 100644 index 00000000..1c74ec40 --- /dev/null +++ b/test/qlib/test_remaining.py @@ -0,0 +1,253 @@ +"""Tests for remaining untested modules: conf, knowledge_base, interactor, utils/fmt, llm_utils, experiment utils.""" + +from __future__ import annotations + +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +PROJECT_ROOT = Path(__file__).parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + + +# ============================================================================= +# RDAgentSettings (core/conf.py) +# ============================================================================= + + +class TestRDAgentSettings: + def test_defaults(self): + from rdagent.core.conf import RDAgentSettings + s = RDAgentSettings() + assert s.multi_proc_n == 1 + assert s.cache_with_pickle is True + assert s.use_file_lock is True + assert s.enable_mlflow is False + assert s.step_semaphore == 1 + assert s.subproc_step is False + + def test_get_max_parallel_int(self): + from rdagent.core.conf import RDAgentSettings + s = RDAgentSettings() + s.step_semaphore = 5 + assert s.get_max_parallel() == 5 + + def test_get_max_parallel_dict(self): + from rdagent.core.conf import RDAgentSettings + s = RDAgentSettings() + s.step_semaphore = {"coding": 3, "running": 2} + assert s.get_max_parallel() == 3 + + def test_is_force_subproc_subproc_step(self): + from rdagent.core.conf import RDAgentSettings + s = RDAgentSettings() + s.subproc_step = True + assert s.is_force_subproc() is True + + def test_is_force_subproc_parallel(self): + from rdagent.core.conf import RDAgentSettings + s = RDAgentSettings() + s.subproc_step = False + s.step_semaphore = 4 + assert s.is_force_subproc() is True + + def test_singleton(self): + from rdagent.core.conf import RD_AGENT_SETTINGS, RDAgentSettings + assert isinstance(RD_AGENT_SETTINGS, RDAgentSettings) + + def test_workspace_path_is_path(self): + from rdagent.core.conf import RDAgentSettings + s = RDAgentSettings() + assert isinstance(s.workspace_path, Path) + + def test_env_prefix(self, monkeypatch): + from rdagent.core.conf import RDAgentSettings + monkeypatch.setenv("multi_proc_n", "4") + s = RDAgentSettings() + assert s.multi_proc_n == 4 + + +# ============================================================================= +# KnowledgeBase (core/knowledge_base.py) +# ============================================================================= + + +class TestKnowledgeBase: + def test_init_without_path(self): + from rdagent.core.knowledge_base import KnowledgeBase + kb = KnowledgeBase() + assert kb.path is None + + def test_init_with_path(self, tmp_path): + from rdagent.core.knowledge_base import KnowledgeBase + p = tmp_path / "kb.pkl" + kb = KnowledgeBase(path=p) + assert kb.path == p + + def test_dump_creates_file(self, tmp_path): + from rdagent.core.knowledge_base import KnowledgeBase + p = tmp_path / "kb.pkl" + kb = KnowledgeBase(path=p) + kb.foo = "bar" + kb.dump() + assert p.exists() + + def test_load_restores_state(self, tmp_path): + from rdagent.core.knowledge_base import KnowledgeBase + p = tmp_path / "kb.pkl" + kb1 = KnowledgeBase(path=p) + kb1.foo = "hello" + kb1.dump() + kb2 = KnowledgeBase(path=p) + assert kb2.foo == "hello" + + +# ============================================================================= +# Interactor (core/interactor.py) +# ============================================================================= + + +class TestInteractor: + def test_abstract_class(self): + from rdagent.core.interactor import Interactor + + class MyInteractor(Interactor): + def interact(self, exp, trace=None): + return exp + + i = MyInteractor(scen=MagicMock()) + assert i.scen is not None + + +# ============================================================================= +# shrink_text (utils/fmt.py) +# ============================================================================= + + +class TestShrinkText: + def test_short_text_unchanged(self): + from rdagent.utils.fmt import shrink_text + result = shrink_text("hello", context_lines=10, line_len=100) + assert result == "hello" + + def test_long_lines_shrunk(self): + from rdagent.utils.fmt import shrink_text + result = shrink_text("a" * 100, context_lines=10, line_len=20) + assert "chars are hidden" in result + + def test_many_lines_shrunk(self): + 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=1000) + assert "lines are hidden" in result + + def test_row_shrink_disabled(self): + 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=1000, row_shrink=False) + assert "lines are hidden" not in result + + +# ============================================================================= +# md5_hash (utils/__init__.py) +# ============================================================================= + + +class TestMD5Hash: + def test_returns_string(self): + from rdagent.utils import md5_hash + h = md5_hash("hello") + assert isinstance(h, str) + assert len(h) == 64 # actually SHA-256, name is historical + + def test_deterministic(self): + from rdagent.utils import md5_hash + assert md5_hash("hello") == md5_hash("hello") + + def test_different_inputs(self): + from rdagent.utils import md5_hash + assert md5_hash("a") != md5_hash("b") + + +# ============================================================================= +# convert2bool (utils/__init__.py) +# ============================================================================= + + +class TestConvert2Bool: + def test_true_strings(self): + from rdagent.utils import convert2bool + for v in ("yes", "true", "True", "YES", "ok"): + assert convert2bool(v) is True + + def test_false_strings(self): + from rdagent.utils import convert2bool + for v in ("no", "false", "False", "NO"): + assert convert2bool(v) is False + + def test_boolean_passthrough(self): + from rdagent.utils import convert2bool + assert convert2bool(True) is True + assert convert2bool(False) is False + + def test_invalid_raises(self): + from rdagent.utils import convert2bool + with pytest.raises(ValueError): + convert2bool("maybe") + + +# ============================================================================= +# calculate_embedding_distance (llm_utils.py) +# ============================================================================= + + +class TestEmbeddingDistance: + def test_empty_lists_returns_empty(self): + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + result = calculate_embedding_distance_between_str_list([], []) + assert result == [[]] + + def test_one_empty_list(self): + from rdagent.oai.llm_utils import calculate_embedding_distance_between_str_list + result = calculate_embedding_distance_between_str_list(["a"], []) + assert result == [[]] + + +# ============================================================================= +# get_file_desc (experiment/utils.py) — pure logic +# ============================================================================= + + +class TestGetFileDesc: + def test_md_file(self, tmp_path): + from rdagent.scenarios.qlib.experiment.utils import get_file_desc + md_file = tmp_path / "test.md" + md_file.write_text("# Hello\ncontent") + desc = get_file_desc(md_file) + assert "Markdown Documentation" in desc + assert "Hello" in desc + + def test_unsupported_extension_raises(self, tmp_path): + from rdagent.scenarios.qlib.experiment.utils import get_file_desc + f = tmp_path / "test.txt" + f.write_text("hello") + with pytest.raises(NotImplementedError): + get_file_desc(f) + + +# ============================================================================= +# workflow/conf.py +# ============================================================================= + + +class TestBasePropSetting: + def test_importable(self): + from rdagent.components.workflow.conf import BasePropSetting + assert BasePropSetting is not None + + def test_has_evolving_n(self): + from rdagent.components.workflow.conf import BasePropSetting + s = BasePropSetting() + assert hasattr(s, "evolving_n")