From a7b392fdc2f1d25ff110f86132fb4cddfd5b1716 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 21 Oct 2026 14:58:56 +0000 Subject: [PATCH 1/2] test: move OpenAI-backed decorator tests into live_provider suite Co-authored-by: Hassieb Pakzad --- tests/e2e/test_decorators.py | 184 --------------------- tests/live_provider/test_decorators.py | 220 +++++++++++++++++++++++++ 2 files changed, 220 insertions(+), 184 deletions(-) create mode 100644 tests/live_provider/test_decorators.py diff --git a/tests/e2e/test_decorators.py b/tests/e2e/test_decorators.py index c11aabaa2..ea40ca048 100644 --- a/tests/e2e/test_decorators.py +++ b/tests/e2e/test_decorators.py @@ -6,14 +6,11 @@ from typing import Optional import pytest -from langchain_core.prompts import ChatPromptTemplate -from langchain_openai import ChatOpenAI from opentelemetry import trace from langfuse import Langfuse, get_client, observe, propagate_attributes from langfuse._client.environment_variables import LANGFUSE_PUBLIC_KEY from langfuse._client.resource_manager import LangfuseResourceManager -from langfuse.langchain import CallbackHandler from langfuse.media import LangfuseMedia from tests.support.utils import ( get_observations, @@ -389,88 +386,6 @@ def level_1_function(*args, **kwargs): assert level_3_observation.total_cost > 0 -def test_decorators_langchain(): - mock_name = "test_decorators_langchain" - langfuse = get_client() - mock_trace_id = langfuse.create_trace_id() - - @observe() - def langchain_operations(*args, **kwargs): - # Get langfuse callback handler for LangChain - handler = CallbackHandler() - prompt = ChatPromptTemplate.from_template("tell me a short joke about {topic}") - model = ChatOpenAI(temperature=0) - - chain = prompt | model - - return chain.invoke( - {"topic": kwargs["topic"]}, - config={ - "callbacks": [handler], - }, - ) - - @observe() - def level_3_function(*args, **kwargs): - langfuse.update_current_span(metadata=mock_metadata) - langfuse.update_current_span(metadata=mock_deep_metadata) - return langchain_operations(*args, **kwargs) - - @observe() - def level_2_function(*args, **kwargs): - langfuse.update_current_span(metadata=mock_metadata) - - return level_3_function(*args, **kwargs) - - @observe() - def level_1_function(*args, **kwargs): - return level_2_function(*args, **kwargs) - - with propagate_attributes(session_id=mock_session_id, trace_name=mock_name): - level_1_function(topic="socks", langfuse_trace_id=mock_trace_id) - - langfuse.flush() - - trace_data = wait_for_trace_snapshot( - mock_trace_id, - is_result_ready=lambda trace: ( - trace.session_id == mock_session_id - and trace.name == mock_name - and { - "level_1_function", - "level_2_function", - "level_3_function", - "langchain_operations", - "ChatPromptTemplate", - }.issubset({observation.name for observation in trace.observations}) - ), - ) - assert len(trace_data.observations) > 2 - - # trace parameters if set anywhere in the call stack - assert trace_data.session_id == mock_session_id - assert trace_data.name == mock_name - - level_1_observation = _get_observation_by_name(trace_data, "level_1_function") - level_2_observation = _get_observation_by_name(trace_data, "level_2_function") - level_3_observation = _get_observation_by_name(trace_data, "level_3_function") - langchain_observation = _get_observation_by_name(trace_data, "langchain_operations") - prompt_observation = _get_observation_by_name(trace_data, "ChatPromptTemplate") - - assert level_1_observation.name == "level_1_function" - assert _is_descendant(trace_data, level_2_observation.id, level_1_observation.id) - assert level_2_observation.name == "level_2_function" - assert level_2_observation.metadata["key"] == mock_metadata["key"] - assert _is_descendant(trace_data, level_3_observation.id, level_2_observation.id) - assert level_3_observation.name == "level_3_function" - assert level_3_observation.metadata["key"] == mock_deep_metadata["key"] - assert _is_descendant(trace_data, langchain_observation.id, level_3_observation.id) - assert langchain_observation.name == "langchain_operations" - - # Check that LangChain components are captured - assert _is_descendant(trace_data, prompt_observation.id, langchain_observation.id) - - def test_get_current_trace_url(): langfuse = get_client() mock_trace_id = langfuse.create_trace_id() @@ -860,105 +775,6 @@ async def main_async(**kwargs): assert nested_observation.output == "Hello--, async --World!" -@pytest.mark.asyncio -async def test_async_nested_openai_chat_stream(): - from langfuse.openai import AsyncOpenAI - - mock_name = "test_async_nested_openai_chat_stream" - langfuse = get_client() - mock_trace_id = langfuse.create_trace_id() - mock_tags = ["tag1", "tag2"] - mock_session_id = "session-id-1" - mock_user_id = "user-id-1" - - @observe(capture_output=False) - async def level_2_function(): - gen = await AsyncOpenAI().chat.completions.create( - model="gpt-3.5-turbo", - messages=[{"role": "user", "content": "1 + 1 = "}], - temperature=0, - metadata={"someKey": "someResponse"}, - stream=True, - ) - - with propagate_attributes( - session_id=mock_session_id, - user_id=mock_user_id, - tags=mock_tags, - trace_name=mock_name, - ): - async for c in gen: - print(c) - - langfuse.update_current_span(metadata=mock_metadata) - - return "level_2" - - @observe() - async def level_1_function(*args, **kwargs): - await level_2_function() - - return "level_1" - - result = await level_1_function( - *mock_args, **mock_kwargs, langfuse_trace_id=mock_trace_id - ) - langfuse.flush() - - assert result == "level_1" # Wrapped function returns correctly - - # ID setting for span or trace - trace_data = wait_for_trace_snapshot( - mock_trace_id, - is_result_ready=lambda trace: ( - trace.session_id == mock_session_id - and trace.name == mock_name - and { - "level_1_function", - "level_2_function", - "OpenAI-generation", - }.issubset({observation.name for observation in trace.observations}) - ), - ) - assert len(trace_data.observations) >= 3 - - # trace parameters if set anywhere in the call stack - assert trace_data.session_id == mock_session_id - assert trace_data.name == mock_name - - level_1_observation = _get_observation_by_name(trace_data, "level_1_function") - level_2_observation = _get_observation_by_name(trace_data, "level_2_function") - level_3_observation = _get_observation_by_name(trace_data, "OpenAI-generation") - assert _is_descendant(trace_data, level_2_observation.id, level_1_observation.id) - assert _is_descendant(trace_data, level_3_observation.id, level_2_observation.id) - - assert level_2_observation.metadata["key"] == mock_metadata["key"] - - generation = level_3_observation - - assert generation.name == "OpenAI-generation" - assert generation.metadata["someKey"] == "someResponse" - assert generation.input == [{"content": "1 + 1 = ", "role": "user"}] - assert generation.type == "GENERATION" - assert "gpt-3.5-turbo" in generation.model - assert generation.start_time is not None - assert generation.end_time is not None - assert generation.start_time < generation.end_time - assert generation.model_parameters == { - "service_tier": "default", - "temperature": 0, - "top_p": 1, - "frequency_penalty": 0, - "max_tokens": "Infinity", - "presence_penalty": 0, - } - assert generation.usage_details["input"] is not None - assert generation.usage_details["output"] is not None - assert generation.usage_details["total"] is not None - print(generation) - assert generation.output == "2" - - def test_generator_as_function_input(): langfuse = get_client() mock_trace_id = langfuse.create_trace_id() diff --git a/tests/live_provider/test_decorators.py b/tests/live_provider/test_decorators.py new file mode 100644 index 000000000..984371fd9 --- /dev/null +++ b/tests/live_provider/test_decorators.py @@ -0,0 +1,220 @@ +import pytest +from langchain_core.prompts import ChatPromptTemplate +from langchain_openai import ChatOpenAI + +from langfuse import get_client, observe, propagate_attributes +from langfuse.langchain import CallbackHandler +from tests.support.utils import wait_for_trace_snapshot + +mock_metadata = {"key": "metadata"} +mock_deep_metadata = {"key": "mock_deep_metadata"} +mock_session_id = "session-id-1" +mock_args = (1, 2, 3) +mock_kwargs = {"a": 1, "b": 2, "c": 3} + + +def _get_observation_by_name(trace_data, name): + return next( + observation + for observation in trace_data.observations + if observation.name == name + ) + + +def _is_descendant(trace_data, child_id, ancestor_id): + observations_by_id = { + observation.id: observation for observation in trace_data.observations + } + current_id = child_id + + while current_id in observations_by_id: + current = observations_by_id[current_id] + parent_id = current.parent_observation_id + if parent_id == ancestor_id: + return True + if parent_id not in observations_by_id: + return False + current_id = parent_id + + return False + + +def test_decorators_langchain(): + mock_name = "test_decorators_langchain" + langfuse = get_client() + mock_trace_id = langfuse.create_trace_id() + + @observe() + def langchain_operations(*args, **kwargs): + # Get langfuse callback handler for LangChain + handler = CallbackHandler() + prompt = ChatPromptTemplate.from_template("tell me a short joke about {topic}") + model = ChatOpenAI(temperature=0) + + chain = prompt | model + + return chain.invoke( + {"topic": kwargs["topic"]}, + config={ + "callbacks": [handler], + }, + ) + + @observe() + def level_3_function(*args, **kwargs): + langfuse.update_current_span(metadata=mock_metadata) + langfuse.update_current_span(metadata=mock_deep_metadata) + return langchain_operations(*args, **kwargs) + + @observe() + def level_2_function(*args, **kwargs): + langfuse.update_current_span(metadata=mock_metadata) + + return level_3_function(*args, **kwargs) + + @observe() + def level_1_function(*args, **kwargs): + return level_2_function(*args, **kwargs) + + with propagate_attributes(session_id=mock_session_id, trace_name=mock_name): + level_1_function(topic="socks", langfuse_trace_id=mock_trace_id) + + langfuse.flush() + + trace_data = wait_for_trace_snapshot( + mock_trace_id, + is_result_ready=lambda trace: ( + trace.session_id == mock_session_id + and trace.name == mock_name + and { + "level_1_function", + "level_2_function", + "level_3_function", + "langchain_operations", + "ChatPromptTemplate", + }.issubset({observation.name for observation in trace.observations}) + ), + ) + assert len(trace_data.observations) > 2 + + # trace parameters if set anywhere in the call stack + assert trace_data.session_id == mock_session_id + assert trace_data.name == mock_name + + level_1_observation = _get_observation_by_name(trace_data, "level_1_function") + level_2_observation = _get_observation_by_name(trace_data, "level_2_function") + level_3_observation = _get_observation_by_name(trace_data, "level_3_function") + langchain_observation = _get_observation_by_name(trace_data, "langchain_operations") + prompt_observation = _get_observation_by_name(trace_data, "ChatPromptTemplate") + + assert level_1_observation.name == "level_1_function" + assert _is_descendant(trace_data, level_2_observation.id, level_1_observation.id) + assert level_2_observation.name == "level_2_function" + assert level_2_observation.metadata["key"] == mock_metadata["key"] + assert _is_descendant(trace_data, level_3_observation.id, level_2_observation.id) + assert level_3_observation.name == "level_3_function" + assert level_3_observation.metadata["key"] == mock_deep_metadata["key"] + assert _is_descendant(trace_data, langchain_observation.id, level_3_observation.id) + assert langchain_observation.name == "langchain_operations" + + # Check that LangChain components are captured + assert _is_descendant(trace_data, prompt_observation.id, langchain_observation.id) + + +@pytest.mark.asyncio +async def test_async_nested_openai_chat_stream(): + from langfuse.openai import AsyncOpenAI + + mock_name = "test_async_nested_openai_chat_stream" + langfuse = get_client() + mock_trace_id = langfuse.create_trace_id() + mock_tags = ["tag1", "tag2"] + mock_session_id = "session-id-1" + mock_user_id = "user-id-1" + + @observe(capture_output=False) + async def level_2_function(): + gen = await AsyncOpenAI().chat.completions.create( + model="gpt-3.5-turbo", + messages=[{"role": "user", "content": "1 + 1 = "}], + temperature=0, + metadata={"someKey": "someResponse"}, + stream=True, + ) + + with propagate_attributes( + session_id=mock_session_id, + user_id=mock_user_id, + tags=mock_tags, + trace_name=mock_name, + ): + async for c in gen: + print(c) + + langfuse.update_current_span(metadata=mock_metadata) + + return "level_2" + + @observe() + async def level_1_function(*args, **kwargs): + await level_2_function() + + return "level_1" + + result = await level_1_function( + *mock_args, **mock_kwargs, langfuse_trace_id=mock_trace_id + ) + langfuse.flush() + + assert result == "level_1" # Wrapped function returns correctly + + # ID setting for span or trace + trace_data = wait_for_trace_snapshot( + mock_trace_id, + is_result_ready=lambda trace: ( + trace.session_id == mock_session_id + and trace.name == mock_name + and { + "level_1_function", + "level_2_function", + "OpenAI-generation", + }.issubset({observation.name for observation in trace.observations}) + ), + ) + assert len(trace_data.observations) >= 3 + + # trace parameters if set anywhere in the call stack + assert trace_data.session_id == mock_session_id + assert trace_data.name == mock_name + + level_1_observation = _get_observation_by_name(trace_data, "level_1_function") + level_2_observation = _get_observation_by_name(trace_data, "level_2_function") + level_3_observation = _get_observation_by_name(trace_data, "OpenAI-generation") + assert _is_descendant(trace_data, level_2_observation.id, level_1_observation.id) + assert _is_descendant(trace_data, level_3_observation.id, level_2_observation.id) + + assert level_2_observation.metadata["key"] == mock_metadata["key"] + + generation = level_3_observation + + assert generation.name == "OpenAI-generation" + assert generation.metadata["someKey"] == "someResponse" + assert generation.input == [{"content": "1 + 1 = ", "role": "user"}] + assert generation.type == "GENERATION" + assert "gpt-3.5-turbo" in generation.model + assert generation.start_time is not None + assert generation.end_time is not None + assert generation.start_time < generation.end_time + assert generation.model_parameters == { + "service_tier": "default", + "temperature": 0, + "top_p": 1, + "frequency_penalty": 0, + "max_tokens": "Infinity", + "presence_penalty": 0, + } + assert generation.usage_details["input"] is not None + assert generation.usage_details["output"] is not None + assert generation.usage_details["total"] is not None + print(generation) + assert generation.output == "2" From 8b70105139deceab273c66ae5f2018ac523ffbf3 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 21 Oct 2026 14:58:56 +0000 Subject: [PATCH 2/2] ci: skip live-provider tests on fork pull requests Co-authored-by: Hassieb Pakzad --- .github/workflows/ci.yml | 15 ++++++++++++++- AGENTS.md | 5 +++-- CONTRIBUTING.md | 2 ++ 3 files changed, 19 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1504f59d3..e225c48b1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -152,23 +152,34 @@ jobs: # SERPAPI_API_KEY: ${{ secrets.SERPAPI_API_KEY }} HUGGINGFACEHUB_API_TOKEN: ${{ secrets.HUGGINGFACEHUB_API_TOKEN }} ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }} + # GitHub withholds secrets from fork PRs, so provider calls there can only fail. + # The suite still runs on same-repo PRs and on main, where a missing key fails loudly. + SKIP_LIVE_PROVIDER: ${{ matrix.suite == 'live_provider' && github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository }} name: ${{ matrix.job_name }} steps: + - name: Skip live-provider tests on fork PRs + if: ${{ env.SKIP_LIVE_PROVIDER == 'true' }} + run: echo "::notice::Live-provider tests are skipped on fork PRs because provider secrets are unavailable. They run after merge on main." - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + if: ${{ env.SKIP_LIVE_PROVIDER != 'true' }} with: persist-credentials: false - name: Install uv and set Python version + if: ${{ env.SKIP_LIVE_PROVIDER != 'true' }} uses: astral-sh/setup-uv@1c37ad07a6a961277cf70c0d37d6f313000f5884 # v10.3.0 with: version: "0.11.2" python-version: "3.13" enable-cache: true # zizmor: ignore[cache-poisoning] CI-only, no artifacts published - name: Install the project dependencies + if: ${{ env.SKIP_LIVE_PROVIDER != 'true' }} run: uv sync --locked - name: Check uv Python version + if: ${{ env.SKIP_LIVE_PROVIDER != 'true' }} run: uv run --frozen python --version - name: Prepare langfuse server compose + if: ${{ env.SKIP_LIVE_PROVIDER != 'true' }} run: | mkdir -p ./langfuse-server LANGFUSE_SERVER_SHA="$(git ls-remote https://github.com/langfuse/langfuse.git HEAD | cut -f1)" @@ -177,6 +188,7 @@ jobs: echo "${LANGFUSE_SERVER_SHA}" - name: Run langfuse server + if: ${{ env.SKIP_LIVE_PROVIDER != 'true' }} run: | cd ./langfuse-server @@ -194,6 +206,7 @@ jobs: echo "::endgroup::" - name: Health check for langfuse server + if: ${{ env.SKIP_LIVE_PROVIDER != 'true' }} run: | echo "Checking if the langfuse server is up..." retry_count=0 @@ -264,7 +277,7 @@ jobs: fi - name: Run live-provider tests - if: ${{ matrix.suite == 'live_provider' }} + if: ${{ matrix.suite == 'live_provider' && env.SKIP_LIVE_PROVIDER != 'true' }} run: | uv run --frozen python --version uv run --frozen pytest \ diff --git a/AGENTS.md b/AGENTS.md index 950bf3055..42f4612da 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -100,7 +100,8 @@ Minimum verification matrix: ### `tests/live_provider` -- This suite uses real provider calls and always runs as one dedicated CI suite. +- This suite uses real provider calls and runs as one dedicated CI suite. CI skips it on fork PRs, which have no provider secrets. +- Any test that calls a model provider belongs here, never in `tests/e2e`, so fork PRs stay green on the e2e shards. - Do not split or shard `tests/live_provider` into separate smoke and extended jobs unless the team explicitly changes that policy. - Keep assertions focused on stable provider-facing behavior rather than brittle observation counts. @@ -111,7 +112,7 @@ The main CI workflow currently runs: - mypy on Python 3.13 - `tests/unit` on a Python 3.11-3.15 matrix - `tests/e2e` in 2 mechanical shards plus a serial subset inside each shard -- `tests/live_provider` as one always-on suite +- `tests/live_provider` as one suite, skipped on fork PRs because GitHub withholds secrets there - PR title validation for Conventional Commits If you change the e2e split: diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 0aaf910ed..51defd979 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -49,6 +49,8 @@ Live-provider tests make real provider calls and require provider API keys: uv run --frozen pytest -n 4 --dist worksteal tests/live_provider -m "live_provider" ``` +GitHub does not expose repository secrets to pull requests from forks, so CI skips the live-provider suite on those PRs. It runs on same-repo PRs and again after merge on `main`. Any test that calls a model provider belongs in `tests/live_provider`, not `tests/e2e`. + Run a specific test with: ```bash