test: add integration tests with in-memory Phoenix (#36)

Author: zurfjereluhmie

.github/workflows/ci.yml

@@ -63,13 +63,37 @@ jobs:
       - name: Test with coverage
         run: uv run pytest tests/ -q --cov=evalwire --cov-report=term-missing --cov-fail-under=85
 
+  test-integration:
+    name: Integration tests
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v8.0.0
+        with:
+          enable-cache: true
+
+      - name: Set up Python
+        uses: actions/setup-python@v6
+        with:
+          python-version-file: .python-version
+
+      - name: Install dependencies
+        run: uv sync --group dev
+
+      - name: Run integration tests
+        run: uv run pytest tests/ -q -m integration
+
   # https://github.com/marketplace/actions/alls-green#why
   ci-success: # This job does nothing and is only used for the branch protection
     name: CI Success
     if: always()
     needs:
       - lint
       - test
+      - test-integration
       - docs-check
     runs-on: ubuntu-latest
     steps:

Makefile

@@ -1,6 +1,6 @@
 .DEFAULT_GOAL := help
 
-.PHONY: help install install-dev install-demo sync lint format typecheck test coverage check fix pre-commit docs docs-serve clean
+.PHONY: help install install-dev install-demo sync lint format typecheck test test-integration coverage check fix pre-commit docs docs-serve clean
 
 help: ## Show this help message
 	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \
@@ -27,9 +27,12 @@ format: ## Run ruff formatter (check only)
 typecheck: ## Run ty type checker
 	uv run ty check
 
-test: ## Run pytest
+test: ## Run pytest (unit tests only)
 	uv run pytest tests/ -q
 
+test-integration: ## Run integration tests (requires Phoenix)
+	uv run pytest tests/ -q -m integration
+
 coverage: ## Run pytest with coverage report
 	uv run pytest tests/ -q --cov=evalwire --cov-report=term-missing --cov-fail-under=85
 

pyproject.toml

@@ -64,6 +64,10 @@ filterwarnings = [
     "ignore::DeprecationWarning:alembic",
     "ignore::sqlalchemy.exc.SAWarning",
 ]
+markers = [
+    "integration: tests that require a running Phoenix instance (deselect with '-m not integration')",
+]
+addopts = "-m 'not integration'"
 
 [dependency-groups]
 dev = [

tests/integration/conftest.py

@@ -0,0 +1,126 @@
+"""Shared fixtures for integration tests.
+
+These tests spin up an in-memory Phoenix server per session and create a
+real ``phoenix.client.Client`` connected to it.  They are gated behind
+the ``integration`` marker so they can be skipped in fast CI runs::
+
+    pytest -m integration          # run only integration tests
+    pytest -m 'not integration'    # skip integration tests (default)
+"""
+
+from __future__ import annotations
+
+import os
+import socket
+import textwrap
+import uuid
+from pathlib import Path
+
+import pytest
+
+
+def _find_free_port() -> int:
+    """Return an available TCP port on localhost."""
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("", 0))
+        return s.getsockname()[1]
+
+
+def _unique_tag() -> str:
+    """Return a short unique suffix for dataset names."""
+    return uuid.uuid4().hex[:8]
+
+
+@pytest.fixture(scope="session")
+def phoenix_server():
+    """Launch an in-memory Phoenix instance for the entire test session.
+
+    Yields the session URL (e.g. ``http://localhost:54321``).
+    Cleans up the server and all data after the session ends.
+    """
+    import phoenix as px
+
+    port = _find_free_port()
+    px.launch_app(run_in_thread=True, use_temp_dir=True, port=port)
+    url = f"http://localhost:{port}"
+
+    yield url
+
+    px.close_app(delete_data=True)
+
+
+@pytest.fixture()
+def phoenix_client(phoenix_server: str):
+    """Return a ``phoenix.client.Client`` connected to the test server.
+
+    Sets ``PHOENIX_COLLECTOR_ENDPOINT`` so that code which creates its own
+    ``Client()`` (like ``evalwire.cli._make_client``) also points at the
+    test instance.
+    """
+    from phoenix.client import Client
+
+    old = os.environ.get("PHOENIX_COLLECTOR_ENDPOINT")
+    os.environ["PHOENIX_COLLECTOR_ENDPOINT"] = phoenix_server
+
+    client = Client()
+
+    yield client
+
+    # Restore original env
+    if old is None:
+        os.environ.pop("PHOENIX_COLLECTOR_ENDPOINT", None)
+    else:
+        os.environ["PHOENIX_COLLECTOR_ENDPOINT"] = old
+
+
+@pytest.fixture()
+def integration_csv(tmp_path: Path) -> tuple[Path, str, str]:
+    """Write a small CSV testset with unique tag names.
+
+    Returns ``(csv_path, search_tag, router_tag)`` so tests can reference
+    the unique dataset names.
+    """
+    tag = _unique_tag()
+    search_tag = f"search_{tag}"
+    router_tag = f"router_{tag}"
+
+    content = textwrap.dedent(f"""\
+        user_query,expected_output,tags
+        "find cycling paths","url-a | url-b","{search_tag} | {router_tag}"
+        "find parks","url-c","{search_tag}"
+        "route me home","home","{router_tag}"
+    """)
+    csv_file = tmp_path / "testset.csv"
+    csv_file.write_text(content)
+    return csv_file, search_tag, router_tag
+
+
+@pytest.fixture()
+def integration_experiments(tmp_path: Path) -> tuple[Path, str]:
+    """Create a minimal experiments/ directory for integration testing.
+
+    Returns ``(experiments_dir, experiment_name)`` where *experiment_name*
+    is the unique dataset-matching directory name.
+    """
+    tag = _unique_tag()
+    exp_name = f"exp_{tag}"
+
+    base = tmp_path / "experiments"
+    base.mkdir()
+
+    exp_dir = base / exp_name
+    exp_dir.mkdir()
+    (exp_dir / "task.py").write_text(
+        textwrap.dedent("""\
+            async def task(example):
+                return example.input.get("user_query", "") + " result"
+        """)
+    )
+    (exp_dir / "check_output.py").write_text(
+        textwrap.dedent("""\
+            def check_output(output, expected):
+                return 1.0 if output else 0.0
+        """)
+    )
+
+    return base, exp_name

tests/integration/test_runner_integration.py

@@ -0,0 +1,186 @@
+"""Integration tests for evalwire.runner.ExperimentRunner.
+
+These tests run against a real in-memory Phoenix instance and verify the
+full experiment lifecycle: dataset lookup, task execution, evaluator scoring,
+and result retrieval.
+"""
+
+from __future__ import annotations
+
+import textwrap
+import uuid
+from pathlib import Path
+
+import pandas as pd
+import pytest
+
+from evalwire.runner import ExperimentRunner
+
+pytestmark = pytest.mark.integration
+
+
+def _create_dataset(phoenix_client, name: str) -> None:
+    """Create a minimal Phoenix dataset with the given name."""
+    df = pd.DataFrame(
+        {
+            "user_query": ["find parks", "find trails"],
+            "expected_output": ["park-url", "trail-url"],
+        }
+    )
+    phoenix_client.datasets.create_dataset(
+        dataframe=df,
+        name=name,
+        input_keys=["user_query"],
+        output_keys=["expected_output"],
+    )
+
+
+_DEFAULT_EVALUATOR = {
+    "check_output": """\
+        def check_output(output, expected):
+            return 1.0 if output else 0.0
+    """,
+}
+
+
+def _make_experiment(
+    base: Path,
+    name: str,
+    task_body: str = "async def task(example): return 'result'",
+    evaluators: dict[str, str] | None = None,
+) -> None:
+    """Create an experiment directory matching a dataset name."""
+    exp_dir = base / name
+    exp_dir.mkdir(parents=True, exist_ok=True)
+    (exp_dir / "task.py").write_text(textwrap.dedent(task_body))
+    for ev_name, ev_body in (evaluators or _DEFAULT_EVALUATOR).items():
+        (exp_dir / f"{ev_name}.py").write_text(textwrap.dedent(ev_body))
+
+
+class TestRunnerLifecycle:
+    """Upload datasets, then run experiments against them."""
+
+    def test_dry_run_completes(self, phoenix_client, tmp_path: Path):
+        name = f"exp_dry_{uuid.uuid4().hex[:8]}"
+        _create_dataset(phoenix_client, name)
+
+        base = tmp_path / "experiments"
+        base.mkdir()
+        _make_experiment(base, name)
+
+        runner = ExperimentRunner(
+            experiments_dir=base,
+            phoenix_client=phoenix_client,
+            dry_run=True,
+        )
+        results = runner.run()
+        assert len(results) == 1
+
+    def test_full_run_returns_experiment_results(self, phoenix_client, tmp_path: Path):
+        name = f"exp_full_{uuid.uuid4().hex[:8]}"
+        _create_dataset(phoenix_client, name)
+
+        base = tmp_path / "experiments"
+        base.mkdir()
+        _make_experiment(base, name)
+
+        runner = ExperimentRunner(
+            experiments_dir=base,
+            phoenix_client=phoenix_client,
+        )
+        results = runner.run()
+        assert len(results) == 1
+
+    def test_run_with_evaluator(self, phoenix_client, tmp_path: Path):
+        name = f"exp_eval_{uuid.uuid4().hex[:8]}"
+        _create_dataset(phoenix_client, name)
+
+        base = tmp_path / "experiments"
+        base.mkdir()
+        _make_experiment(
+            base,
+            name,
+            evaluators={
+                "score": """\
+                    def score(output, expected):
+                        return 0.5
+                """,
+            },
+        )
+
+        runner = ExperimentRunner(
+            experiments_dir=base,
+            phoenix_client=phoenix_client,
+            dry_run=True,
+        )
+        results = runner.run()
+        assert len(results) == 1
+
+    def test_run_with_name_filter(self, phoenix_client, tmp_path: Path):
+        name = f"exp_filter_{uuid.uuid4().hex[:8]}"
+        _create_dataset(phoenix_client, name)
+
+        base = tmp_path / "experiments"
+        base.mkdir()
+        _make_experiment(base, name)
+
+        runner = ExperimentRunner(
+            experiments_dir=base,
+            phoenix_client=phoenix_client,
+            dry_run=True,
+        )
+        results = runner.run(names=[name])
+        assert len(results) == 1
+
+    def test_nonexistent_dataset_raises_system_exit(
+        self, phoenix_client, tmp_path: Path
+    ):
+        name = f"exp_missing_{uuid.uuid4().hex[:8]}"
+
+        base = tmp_path / "experiments"
+        base.mkdir()
+        _make_experiment(base, name)
+
+        runner = ExperimentRunner(
+            experiments_dir=base,
+            phoenix_client=phoenix_client,
+        )
+        with pytest.raises(SystemExit) as exc_info:
+            runner.run()
+        assert exc_info.value.code == 1
+
+    def test_concurrent_run(self, phoenix_client, tmp_path: Path):
+        name = f"exp_conc_{uuid.uuid4().hex[:8]}"
+        _create_dataset(phoenix_client, name)
+
+        base = tmp_path / "experiments"
+        base.mkdir()
+        _make_experiment(base, name)
+
+        runner = ExperimentRunner(
+            experiments_dir=base,
+            phoenix_client=phoenix_client,
+            concurrency=2,
+            dry_run=True,
+        )
+        results = runner.run()
+        assert len(results) == 1
+
+    def test_custom_prefix_and_metadata(self, phoenix_client, tmp_path: Path):
+        name = f"exp_meta_{uuid.uuid4().hex[:8]}"
+        _create_dataset(phoenix_client, name)
+
+        base = tmp_path / "experiments"
+        base.mkdir()
+        _make_experiment(base, name)
+
+        runner = ExperimentRunner(
+            experiments_dir=base,
+            phoenix_client=phoenix_client,
+            dry_run=True,
+        )
+        results = runner.run(
+            experiment_name_prefix="integ",
+            metadata={"ci": True},
+        )
+        assert len(results) == 1