diff --git a/tests/test_build_subgraph.py b/tests/test_build_subgraph.py new file mode 100644 index 0000000..7ed46c1 --- /dev/null +++ b/tests/test_build_subgraph.py @@ -0,0 +1,136 @@ +"""Tests for evalwire.langgraph.build_subgraph.""" + +from __future__ import annotations + +import sys +from dataclasses import dataclass +from typing import Any + +import pytest + + +class TestBuildSubgraph: + """Tests for build_subgraph (requires evalwire[langgraph] extra).""" + + @pytest.fixture(autouse=True) + def _skip_without_langgraph(self): + pytest.importorskip("langgraph", reason="langgraph not installed") + + @pytest.fixture() + def state_cls(self): + @dataclass + class State: + value: str = "" + + return State + + def test_single_node_graph(self, state_cls: type): + from evalwire.langgraph import build_subgraph + + def node_a(state: Any) -> dict: + return {"value": "a"} + + compiled = build_subgraph( + nodes=[("node_a", node_a)], + state_cls=state_cls, + ) + assert compiled is not None + + def test_linear_edge_wiring(self, state_cls: type): + """START -> n1 -> n2 -> n3 -> END.""" + from evalwire.langgraph import build_subgraph + + call_order: list[str] = [] + + def node_1(state: Any) -> dict: + call_order.append("n1") + return {"value": state.value + "1"} + + def node_2(state: Any) -> dict: + call_order.append("n2") + return {"value": state.value + "2"} + + def node_3(state: Any) -> dict: + call_order.append("n3") + return {"value": state.value + "3"} + + compiled = build_subgraph( + nodes=[("n1", node_1), ("n2", node_2), ("n3", node_3)], + state_cls=state_cls, + ) + result = compiled.invoke({"value": ""}) + assert result["value"] == "123" + assert call_order == ["n1", "n2", "n3"] + + def test_with_name_parameter(self, state_cls: type): + from evalwire.langgraph import build_subgraph + + compiled = build_subgraph( + nodes=[("n", lambda state: {"value": "x"})], + state_cls=state_cls, + name="my-graph", + ) + assert compiled.name == "my-graph" + + def test_without_name_parameter(self, state_cls: type): + from evalwire.langgraph import build_subgraph + + compiled = build_subgraph( + nodes=[("n", lambda state: {"value": "x"})], + state_cls=state_cls, + ) + # Default name assigned by LangGraph (not None) + assert compiled.name is not None + + def test_with_input_cls(self): + from evalwire.langgraph import build_subgraph + + @dataclass + class FullState: + query: str = "" + result: str = "" + + @dataclass + class InputState: + query: str = "" + + compiled = build_subgraph( + nodes=[("n", lambda state: {"result": "done"})], + state_cls=FullState, + input_cls=InputState, + ) + result = compiled.invoke({"query": "hello"}) + assert result["result"] == "done" + + def test_with_checkpointer(self, state_cls: type): + mem_mod = pytest.importorskip( + "langgraph.checkpoint.memory", + reason="langgraph checkpoint not installed", + ) + + from evalwire.langgraph import build_subgraph + + compiled = build_subgraph( + nodes=[("n", lambda state: {"value": "x"})], + state_cls=state_cls, + checkpointer=mem_mod.InMemorySaver(), + ) + assert compiled is not None + + def test_empty_nodes_raises(self, state_cls: type): + from evalwire.langgraph import build_subgraph + + with pytest.raises(IndexError): + build_subgraph(nodes=[], state_cls=state_cls) + + +class TestBuildSubgraphImportError: + def test_raises_import_error_without_langgraph(self, monkeypatch): + """build_subgraph raises ImportError if langgraph is not installed.""" + monkeypatch.setitem(sys.modules, "langgraph", None) + monkeypatch.setitem(sys.modules, "langgraph.graph", None) + + from evalwire.langgraph import build_subgraph + + with pytest.raises(ImportError, match="evalwire\\[langgraph\\]"): + build_subgraph(nodes=[("n", lambda s: s)], state_cls=object) diff --git a/tests/test_helpers.py b/tests/test_helpers.py new file mode 100644 index 0000000..713bf38 --- /dev/null +++ b/tests/test_helpers.py @@ -0,0 +1,90 @@ +"""Direct tests for evalwire.evaluators._helpers.""" + +from evalwire.evaluators._helpers import _parse_expected, _zero_value_for + + +class TestParseExpected: + def test_list_input_returned_as_is(self): + result = _parse_expected({"expected_output": ["a", "b"]}) + assert result == ["a", "b"] + + def test_string_literal_list(self): + result = _parse_expected({"expected_output": "['x', 'y']"}) + assert result == ["x", "y"] + + def test_plain_string_wrapped_in_list(self): + result = _parse_expected({"expected_output": "hello"}) + assert result == ["hello"] + + def test_numeric_string_wrapped_in_list(self): + """ast.literal_eval('3.14') -> float 3.14, which gets wrapped.""" + result = _parse_expected({"expected_output": "3.14"}) + assert result == [3.14] + + def test_integer_string_wrapped_in_list(self): + result = _parse_expected({"expected_output": "42"}) + assert result == [42] + + def test_missing_key_returns_empty_list(self): + result = _parse_expected({}) + assert result == [] + + def test_empty_list_input(self): + result = _parse_expected({"expected_output": []}) + assert result == [] + + def test_empty_string_wrapped_in_list(self): + result = _parse_expected({"expected_output": ""}) + # empty string can't be literal_eval'd -> wrapped + assert result == [""] + + def test_tuple_input_converted_to_list(self): + result = _parse_expected({"expected_output": ("a", "b")}) + assert result == ["a", "b"] + + def test_string_literal_tuple(self): + result = _parse_expected({"expected_output": "('a', 'b')"}) + assert result == ["a", "b"] + + def test_nested_list_literal(self): + """ast.literal_eval on a nested list returns the nested structure.""" + result = _parse_expected({"expected_output": "[['a', 'b'], ['c']]"}) + assert result == [["a", "b"], ["c"]] + + def test_url_string_not_parsed(self): + """URLs should not be parsed by ast.literal_eval.""" + result = _parse_expected({"expected_output": "https://example.com"}) + assert result == ["https://example.com"] + + def test_none_value(self): + """None is not a list, not a string -> wrapped as-is.""" + result = _parse_expected({"expected_output": None}) + assert result == [None] + + def test_boolean_string(self): + """'True' is a valid Python literal.""" + result = _parse_expected({"expected_output": "True"}) + assert result == [True] + + +class TestZeroValueFor: + def test_bool_returns_false(self): + assert _zero_value_for(bool) is False + + def test_int_returns_zero_float(self): + assert _zero_value_for(int) == 0.0 + + def test_float_returns_zero_float(self): + assert _zero_value_for(float) == 0.0 + + def test_str_returns_zero_float(self): + assert _zero_value_for(str) == 0.0 + + def test_none_returns_zero_float(self): + assert _zero_value_for(None) == 0.0 + + def test_custom_type_returns_zero_float(self): + class Custom: + pass + + assert _zero_value_for(Custom) == 0.0 diff --git a/tests/test_observability.py b/tests/test_observability.py new file mode 100644 index 0000000..7252469 --- /dev/null +++ b/tests/test_observability.py @@ -0,0 +1,61 @@ +"""Tests for evalwire.observability.setup_observability.""" + +from unittest.mock import MagicMock, patch + + +class TestSetupObservability: + def test_calls_register_with_auto_instrument_true(self): + mock_provider = MagicMock() + with patch( + "phoenix.otel.register", return_value=mock_provider, create=True + ) as mock_register: + from evalwire.observability import setup_observability + + result = setup_observability() + + mock_register.assert_called_once_with(auto_instrument=True) + assert result is mock_provider + + def test_calls_register_with_auto_instrument_false(self): + mock_provider = MagicMock() + with patch( + "phoenix.otel.register", return_value=mock_provider, create=True + ) as mock_register: + from evalwire.observability import setup_observability + + result = setup_observability(auto_instrument=False) + + mock_register.assert_called_once_with(auto_instrument=False) + assert result is mock_provider + + def test_instruments_each_instrumentor(self): + mock_provider = MagicMock() + inst_a = MagicMock() + inst_b = MagicMock() + + with patch("phoenix.otel.register", return_value=mock_provider, create=True): + from evalwire.observability import setup_observability + + setup_observability(instrumentors=[inst_a, inst_b]) + + inst_a.instrument.assert_called_once_with(tracer_provider=mock_provider) + inst_b.instrument.assert_called_once_with(tracer_provider=mock_provider) + + def test_no_instrumentors_none(self): + mock_provider = MagicMock() + with patch("phoenix.otel.register", return_value=mock_provider, create=True): + from evalwire.observability import setup_observability + + # Should not raise when instrumentors is None (default) + result = setup_observability(instrumentors=None) + + assert result is mock_provider + + def test_empty_instrumentors_list(self): + mock_provider = MagicMock() + with patch("phoenix.otel.register", return_value=mock_provider, create=True): + from evalwire.observability import setup_observability + + result = setup_observability(instrumentors=[]) + + assert result is mock_provider diff --git a/tests/test_runner.py b/tests/test_runner.py index 9ffb92e..3403337 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -1,11 +1,12 @@ """Tests for evalwire.runner.ExperimentRunner.""" +import threading from pathlib import Path from unittest.mock import MagicMock import pytest -from evalwire.runner import ExperimentRunner +from evalwire.runner import ExperimentRunner, _get_thread_event_loop def _make_runner( @@ -308,3 +309,71 @@ def _capture_task(**kwargs: object) -> MagicMock: # "RuntimeError: Event loop is closed". assert wrapped(example=object()) == "ok" assert wrapped(example=object()) == "ok" # must not raise + + +class TestConcurrentExecution: + def test_concurrent_experiments_all_complete( + self, experiments_dir: Path, mock_phoenix_client: MagicMock + ): + """With concurrency=2, all experiments should still complete.""" + mock_phoenix_client.experiments.run_experiment.return_value = MagicMock() + runner = _make_runner(experiments_dir, mock_phoenix_client, concurrency=2) + results = runner.run() + assert len(results) == 2 + assert mock_phoenix_client.experiments.run_experiment.call_count == 2 + + def test_concurrent_experiments_with_async_tasks( + self, tmp_path: Path, mock_phoenix_client: MagicMock + ): + """Async tasks run concurrently without event loop conflicts.""" + base = tmp_path / "experiments" + base.mkdir() + + for i in range(3): + exp = base / f"exp_{i}" + exp.mkdir() + (exp / "task.py").write_text( + f"async def task(example): return 'result_{i}'\n" + ) + + mock_phoenix_client.experiments.run_experiment.return_value = MagicMock() + runner = _make_runner(base, mock_phoenix_client, concurrency=3) + results = runner.run() + assert len(results) == 3 + + +class TestGetThreadEventLoop: + def test_returns_event_loop(self): + loop = _get_thread_event_loop() + assert loop is not None + assert not loop.is_closed() + + def test_returns_same_loop_on_repeated_calls(self): + loop1 = _get_thread_event_loop() + loop2 = _get_thread_event_loop() + assert loop1 is loop2 + + def test_creates_new_loop_if_closed(self): + loop1 = _get_thread_event_loop() + loop1.close() + loop2 = _get_thread_event_loop() + assert loop2 is not loop1 + assert not loop2.is_closed() + + def test_different_threads_get_different_loops(self): + loops: dict[str, object] = {} + barrier = threading.Barrier(2) + + def worker(name: str) -> None: + loop = _get_thread_event_loop() + loops[name] = loop + barrier.wait() + + t1 = threading.Thread(target=worker, args=("a",)) + t2 = threading.Thread(target=worker, args=("b",)) + t1.start() + t2.start() + t1.join() + t2.join() + + assert loops["a"] is not loops["b"]