test: add tests for observability, build_subgraph, helpers, and concurrency

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)

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

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

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"]