pengjunjie/langflow
1
0
Fork 1
mirror of https://github.com/langflow-ai/langflow.git synced 2026-07-24 01:22:23 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
6d2e255f5d37dfe39420d2c7bb20e37c1867dc7c
langflow/src/lfx/tests/unit/cli/test_run_command.py
Jordan Frazier 784169cee7 fix: fallback to sync call in lfx run when stream=true (#12906) 
* fix(lfx): unblock streaming flows in lfx run

LCModelComponent._handle_stream tried to persist a partial Message via
send_message whenever the LM was wired to ChatOutput, but the message-
store path requires session_id and the chunk-consumption path requires
an EventManager. lfx run had neither, so flows with stream=True crashed
in astore_message ("session_id, sender, sender_name must be provided").

- run_flow now auto-generates a session_id when none is supplied so the
  message-store validator passes; an explicit value still wins.
- lfx run gains a --session-id flag for memory continuity across runs
  (Memory / MessageHistory components keyed on session_id).
- _handle_stream now also requires an EventManager before taking the
  streaming branch — without one, the chunk iterator would be stored
  but never drained, surfacing as an empty result downstream. Falls
  back to ainvoke and returns the full text instead.

Tests cover the autogen, caller-precedence, and uniqueness cases on
run_flow plus the four _handle_stream branches (no session_id, no
event_manager, both present, not connected to chat output).

* fix(lfx): autogen session_id in CUGA agent when graph has none

Matches the pattern already used by base/agents/agent.py and
base/agents/altk_base_agent.py. Without this, calling the CUGA agent
outside run_flow (which now autogens a session_id) would still fail
astore_message validation.

* [autofix.ci] apply automated fixes

* fix(lfx): plumb session_id through serve /run and /stream

StreamRequest already declared a session_id field but it was never
applied to the graph; RunRequest didn't have the field at all. Both
endpoints called execute_graph_with_capture, which executed against
an empty graph.session_id, so message-store paths skipped storage
silently and Memory components could not maintain continuity.

- Add session_id to RunRequest.
- Have execute_graph_with_capture accept session_id, autogen if empty
  (matches run_flow), and apply it to graph.session_id before
  execution.
- Forward session_id from both /run and /stream handlers.

* fix(lfx): propagate session_id and user_id so memory works on lfx run

The lfx run path uses graph.async_start instead of graph.arun, bypassing
the has_session_id_vertices propagation loop in Graph._run that the
playground hits via build_graph_from_data. Memory/MessageHistory
components reading session_id from their input field would see "" even
when --session-id was passed. Replicate the loop in run_flow and
execute_graph_with_capture, with the same precedence as the playground:
hardcoded values on the component win.

AgentComponent's variable lookup precheck blocks any flow that resolves
variables (e.g. api_key) when graph.user_id is empty. Auto-generate a
ceremonial UUID; lfx's env-fallback variable service ignores user_id, so
this only satisfies the precheck — env vars remain process-global with
no per-user scoping.

Make VariableService.get_variable async to match the langflow call site
(`await variable_service.get_variable(...)` in custom_component.get_variable).
The signature accepts and ignores user_id/field/session kwargs so flows
behave identically under either backend.

Tests cover propagation precedence (empty input filled, hardcoded value
preserved, missing vertex skipped), user_id auto-gen and caller-takes-
precedence, and the async signature with kwarg absorption.

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* [autofix.ci] apply automated fixes (attempt 3/3)

* fix(lfx): plumb fallback_to_env_vars so DatabaseVariableService works in lfx run

The lfx run path uses graph.async_start, which never propagated
fallback_to_env_vars to vertex builds — the flag defaulted to False all
the way down to update_params_with_load_from_db_fields. Result: if a user
swapped lfx's env-fallback VariableService for langflow's
DatabaseVariableService (via lfx.toml), every load_from_db variable
(e.g. api_key=OPENAI_API_KEY) would raise "variable not found" because
the random ceremonial user_id has no DB rows, with no env fallback.

Add fallback_to_env_vars kwarg to async_start and astep (default False,
non-breaking for existing callers). run_flow and execute_graph_with_capture
read settings.fallback_to_env_var (defaults True, settable via
LANGFLOW_FALLBACK_TO_ENV_VAR=false) and pass it through. Mirrors what
processing.process.run_graph_internal does for the langflow API path.

Make memory.stubs.astore_message tolerant of non-UUID flow_ids: the stub
is the no-op fallback when no real database is registered, so it should
not crash on synthetic identifiers (e.g. test fixtures, lfx callers
passing string flow ids). UUID parsing only normalizes format; an
invalid string is preserved verbatim.

Tests:
- TestRunFlowFallbackToEnvVars: confirms run_flow forwards
  fallback_to_env_vars from settings (default and disabled).
- TestGraphExecution: same for execute_graph_with_capture.
- Existing mock_async_start signatures updated to accept **kwargs.
- Test fixtures using flow_id="test-flow-id" replaced with a real UUID
  so the now-active ChatInput storage path doesn't trip stubs.py's
  UUID parse — aligning fixtures with production semantics.

* test(lfx): pass session_id=None when invoking the typer run() directly

The lfx test suite calls ``run(...)`` (the typer command) without going through
typer's CLI parser. Any parameter with a ``typer.Option(...)`` default in the
signature (e.g. ``session_id``) evaluates to a ``typer.models.OptionInfo``
sentinel under that invocation pattern, not None. The session_id propagation
loop then writes that sentinel into ``vertex.raw_params["session_id"]``, which
fails ``MessageTextInput`` validation with
``Invalid value type <class 'typer.models.OptionInfo'>``.

Fix at the call site: pass ``session_id=None`` explicitly. Mirrors how typer
would resolve the option after parsing CLI args. Two tests affected;
test_run_command.py now reflects the constraint that calls bypassing typer
must pass all option-shaped args.

* fix(lfx): harden session_id/user_id handling on lfx run path

Addresses review findings on the streaming-fix PR:

- Reject empty/whitespace --session-id and --user-id up-front so a shell
  quirk or empty env var surfaces a clear error instead of silently
  auto-generating a fresh session and breaking Memory continuity.
- Extract a shared helper (lfx/run/_defaults.py) for session_id/user_id
  auto-gen, vertex propagation, and fallback_to_env_vars resolution; both
  run_flow and execute_graph_with_capture now delegate to it.
- Warn when settings_service is None (silently flipped fallback_to_env_vars
  to False before).
- CUGA agent: wrap uuid.uuid4() with str() to match the rest of the file
  and the codebase's expectation that Message.session_id renders as a hex
  string. Add focused tests with module-level skip when the cuga import
  side-effect (MODELS_METADATA["OpenAI"]) isn't satisfiable.
- Streaming-fallback warning now names the component (display_name + _id)
  so users can identify which model fell back to ainvoke.
- memory/stubs.py: replace contextlib.suppress(ValueError) with explicit
  try/except + warning so malformed flow_ids leave a breadcrumb.
- Improve --session-id help text to explain WHEN to set it.
- Tune session_id auto-gen log level from warning to info (less noisy on
  default CLI runs; visible at -v).
- Add tests: --session-id CLI plumbing, multi-call continuity, empty/
  whitespace rejection, and isinstance(str) on the streaming fallback path.

* [autofix.ci] apply automated fixes

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-04-29 15:22:24 +00:00

541 lines
19 KiB
Python
Raw Blame History

"""Unit tests for the run command functionality."""
import contextlib
import json
import tempfile
from pathlib import Path
from unittest.mock import AsyncMock, patch
import pytest
import typer
from lfx.cli.run import run
class TestRunCommand:
"""Unit tests for run command internal functionality."""
@pytest.fixture
def simple_chat_script(self, tmp_path):
"""Create a simple chat script for testing."""
script_content = '''"""A simple chat flow example for Langflow.
This script demonstrates how to set up a basic conversational flow using Langflow's ChatInput and ChatOutput components.
Features:
- Configures logging to 'langflow.log' at INFO level
- Connects ChatInput to ChatOutput
- Builds a Graph object for the flow
Usage:
python simple_chat.py
You can use this script as a template for building more complex conversational flows in Langflow.
"""
from pathlib import Path
from lfx.components.input_output import ChatInput, ChatOutput
from lfx.graph import Graph
from lfx.log.logger import LogConfig
log_config = LogConfig(
log_level="INFO",
log_file=Path("langflow.log"),
)
chat_input = ChatInput()
chat_output = ChatOutput().set(input_value=chat_input.message_response)
graph = Graph(chat_input, chat_output, log_config=log_config)
'''
script_path = tmp_path / "simple_chat.py"
script_path.write_text(script_content)
return script_path
@pytest.fixture
def invalid_script(self, tmp_path):
"""Create a script without a graph variable."""
script_content = '''"""Invalid script without graph variable."""
from lfx.components.input_output import ChatInput
chat_input = ChatInput()
# Missing graph variable
'''
script_path = tmp_path / "invalid_script.py"
script_path.write_text(script_content)
return script_path
@pytest.fixture
def syntax_error_script(self, tmp_path):
"""Create a script with syntax errors."""
script_content = '''"""Script with syntax errors."""
from lfx.components.input_output import ChatInput
# Syntax error - missing closing parenthesis
chat_input = ChatInput(
'''
script_path = tmp_path / "syntax_error.py"
script_path.write_text(script_content)
return script_path
@pytest.fixture
def simple_json_flow(self):
"""Create a simple JSON flow for testing."""
return {
"data": {
"nodes": [
{
"id": "ChatInput-1",
"type": "ChatInput",
"position": {"x": 100, "y": 100},
"data": {"display_name": "Chat Input"},
},
{
"id": "ChatOutput-1",
"type": "ChatOutput",
"position": {"x": 400, "y": 100},
"data": {"display_name": "Chat Output"},
},
],
"edges": [
{
"id": "edge-1",
"source": "ChatInput-1",
"target": "ChatOutput-1",
"sourceHandle": "message_response",
"targetHandle": "input_value",
}
],
}
}
def test_execute_input_validation_no_sources(self):
"""Test that execute raises exit code 1 when no input source is provided."""
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=None,
input_value=None,
input_value_option=None,
verbose=False,
output_format="json",
flow_json=None,
stdin=False,
)
assert exc_info.value.exit_code == 1
def test_execute_input_validation_multiple_sources(self, simple_chat_script):
"""Test that execute raises exit code 1 when multiple input sources are provided."""
# Test script_path + flow_json
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=simple_chat_script,
input_value=None,
input_value_option=None,
verbose=False,
output_format="json",
flow_json='{"data": {"nodes": []}}',
stdin=False,
)
assert exc_info.value.exit_code == 1
# Test flow_json + stdin
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=None,
input_value=None,
input_value_option=None,
verbose=False,
output_format="json",
flow_json='{"data": {"nodes": []}}',
stdin=True,
)
assert exc_info.value.exit_code == 1
def test_execute_python_script_success(self, simple_chat_script, capsys):
"""Test executing a valid Python script."""
# Test that Python script execution either succeeds or fails gracefully
# ``run`` is a typer command, so any parameter with a ``typer.Option(...)`` default
# (e.g. session_id) needs to be passed explicitly when invoking outside typer's
# parser — otherwise the default evaluates to a ``typer.models.OptionInfo``
# sentinel and propagates downstream.
with contextlib.suppress(typer.Exit):
run(
script_path=simple_chat_script,
input_value="Hello, world!",
input_value_option=None,
verbose=False,
output_format="json",
flow_json=None,
stdin=False,
session_id=None,
)
# Test passes as long as no unhandled exceptions occur
# Check that output was produced
captured = capsys.readouterr()
if captured.out:
# Should be valid JSON when successful
# Output should always be valid JSON when verbose=False
output_data = json.loads(captured.out)
assert isinstance(output_data, dict)
# Either success with result or error with error field
assert "result" in output_data or "error" in output_data
def test_execute_python_script_verbose(self, simple_chat_script, capsys):
"""Test executing a Python script with verbose output."""
# Test that verbose mode execution either succeeds or fails gracefully
# See note in ``test_execute_python_script_success`` re: session_id=None.
with contextlib.suppress(typer.Exit):
run(
script_path=simple_chat_script,
input_value="Hello, world!",
input_value_option=None,
verbose=True,
output_format="json",
flow_json=None,
stdin=False,
session_id=None,
)
# Test passes as long as no unhandled exceptions occur
# In verbose mode, there should be diagnostic output
captured = capsys.readouterr()
# Verbose mode should show diagnostic messages in stderr
assert len(captured.out + captured.err) > 0
def test_execute_python_script_different_formats(self, simple_chat_script):
"""Test executing a Python script with different output formats."""
formats = ["json", "text", "message", "result"]
for output_format in formats:
# Test that each format either succeeds or fails gracefully
with contextlib.suppress(typer.Exit):
run(
script_path=simple_chat_script,
input_value="Test input",
input_value_option=None,
verbose=False,
output_format=output_format,
flow_json=None,
stdin=False,
)
# Test passes as long as no unhandled exceptions occur
def test_execute_file_not_exists(self, tmp_path):
"""Test execute with non-existent file raises exit code 1."""
non_existent_file = tmp_path / "does_not_exist.py"
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=non_existent_file,
input_value=None,
input_value_option=None,
verbose=False,
output_format="json",
flow_json=None,
stdin=False,
)
assert exc_info.value.exit_code == 1
def test_execute_invalid_file_extension(self, tmp_path):
"""Test execute with invalid file extension raises exit code 1."""
txt_file = tmp_path / "test.txt"
txt_file.write_text("not a script")
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=txt_file,
input_value=None,
input_value_option=None,
verbose=False,
output_format="json",
flow_json=None,
stdin=False,
)
assert exc_info.value.exit_code == 1
def test_execute_python_script_no_graph_variable(self, invalid_script):
"""Test execute with Python script that has no graph variable."""
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=invalid_script,
input_value=None,
input_value_option=None,
verbose=False,
output_format="json",
flow_json=None,
stdin=False,
)
assert exc_info.value.exit_code == 1
def test_execute_python_script_syntax_error(self, syntax_error_script):
"""Test execute with Python script that has syntax errors."""
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=syntax_error_script,
input_value=None,
input_value_option=None,
verbose=False,
output_format="json",
flow_json=None,
stdin=False,
)
assert exc_info.value.exit_code == 1
def test_execute_flow_json_valid(self, simple_json_flow):
"""Test execute with valid flow_json."""
flow_json_str = json.dumps(simple_json_flow)
# Test that JSON flow execution either succeeds or fails gracefully
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=None,
input_value="Hello JSON!",
input_value_option=None,
verbose=False,
output_format="json",
flow_json=flow_json_str,
stdin=False,
)
# The function should exit cleanly (either success or expected failure)
assert exc_info.value.exit_code in [0, 1]
def test_execute_flow_json_invalid(self):
"""Test execute with invalid flow_json raises exit code 1."""
invalid_json = '{"nodes": [invalid json'
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=None,
input_value=None,
input_value_option=None,
verbose=False,
output_format="json",
flow_json=invalid_json,
stdin=False,
)
assert exc_info.value.exit_code == 1
@patch("sys.stdin")
def test_execute_stdin_valid(self, mock_stdin, simple_json_flow):
"""Test execute with valid stdin input."""
flow_json_str = json.dumps(simple_json_flow)
mock_stdin.read.return_value = flow_json_str
# Test that stdin execution either succeeds or fails gracefully.
# See note in ``test_execute_python_script_success`` re: session_id=None.
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=None,
input_value="Hello stdin!",
input_value_option=None,
verbose=False,
output_format="json",
flow_json=None,
stdin=True,
session_id=None,
)
# Check that stdin was read and function exited cleanly
mock_stdin.read.assert_called_once()
assert exc_info.value.exit_code in [0, 1]
@patch("sys.stdin")
def test_execute_stdin_empty(self, mock_stdin):
"""Test execute with empty stdin raises exit code 1."""
mock_stdin.read.return_value = ""
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=None,
input_value=None,
input_value_option=None,
verbose=False,
output_format="json",
flow_json=None,
stdin=True,
)
assert exc_info.value.exit_code == 1
@patch("sys.stdin")
def test_execute_stdin_invalid(self, mock_stdin):
"""Test execute with invalid stdin JSON raises exit code 1."""
mock_stdin.read.return_value = '{"nodes": [invalid json'
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=None,
input_value=None,
input_value_option=None,
verbose=False,
output_format="json",
flow_json=None,
stdin=True,
)
assert exc_info.value.exit_code == 1
def test_execute_input_value_precedence(self, simple_chat_script, capsys):
"""Test that positional input_value takes precedence over --input-value option."""
# Test that input precedence works and execution either succeeds or fails gracefully
with contextlib.suppress(typer.Exit):
run(
script_path=simple_chat_script,
input_value="positional_value",
input_value_option="option_value",
verbose=False,
output_format="json",
flow_json=None,
stdin=False,
)
# Test passes as long as no unhandled exceptions occur
# If successful, verify that positional value was used
captured = capsys.readouterr()
if captured.out and "positional_value" in captured.out:
# Positional value was used correctly
assert True
def test_execute_directory_instead_of_file(self, tmp_path):
"""Test execute with directory instead of file raises exit code 1."""
directory = tmp_path / "test_dir"
directory.mkdir()
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=directory,
input_value=None,
input_value_option=None,
verbose=False,
output_format="json",
flow_json=None,
stdin=False,
)
assert exc_info.value.exit_code == 1
def test_execute_json_flow_with_temporary_file_cleanup(self, simple_json_flow):
"""Test that temporary files are cleaned up when using flow_json."""
flow_json_str = json.dumps(simple_json_flow)
# Count temporary files before
temp_dir = Path(tempfile.gettempdir())
temp_files_before = list(temp_dir.glob("*.json"))
with contextlib.suppress(typer.Exit):
run(
script_path=None,
input_value="Test cleanup",
input_value_option=None,
verbose=False,
output_format="json",
flow_json=flow_json_str,
stdin=False,
)
# Count temporary files after
temp_files_after = list(temp_dir.glob("*.json"))
# Should not have more temp files than before (cleanup working)
assert len(temp_files_after) <= len(temp_files_before) + 1 # Allow for one potential leftover
def test_execute_verbose_error_output(self, invalid_script, capsys):
"""Test that verbose mode shows error details."""
# See note in ``test_execute_python_script_success`` re: session_id=None.
with pytest.raises(typer.Exit) as exc_info:
run(
script_path=invalid_script,
input_value=None,
input_value_option=None,
verbose=True,
output_format="json",
flow_json=None,
stdin=False,
session_id=None,
)
assert exc_info.value.exit_code == 1
captured = capsys.readouterr()
# Verbose mode should show error details
error_output = captured.out + captured.err
assert "graph" in error_output.lower() or "variable" in error_output.lower()
def test_session_id_cli_flag_plumbs_through_to_run_flow(self, simple_chat_script):
"""--session-id is wired from typer through `run` into `run_flow(session_id=...)`.
Locks in the typer Option name so a rename or wiring typo at lfx.cli.run.run
is caught here (the unit tests on run_flow itself only cover the function
contract, not this CLI layer).
"""
captured = {}
async def fake_run_flow(**kwargs):
captured.update(kwargs)
return {"success": True, "result": "ok", "logs": ""}
with patch("lfx.cli.run.run_flow", new=AsyncMock(side_effect=fake_run_flow)):
run(
script_path=simple_chat_script,
input_value="hi",
input_value_option=None,
verbose=False,
output_format="json",
flow_json=None,
stdin=False,
session_id="my-fixed-session",
)
assert captured.get("session_id") == "my-fixed-session"
def test_session_id_cli_flag_omitted_passes_none(self, simple_chat_script):
"""No --session-id => `run_flow` receives session_id=None (auto-gen happens downstream)."""
captured = {}
async def fake_run_flow(**kwargs):
captured.update(kwargs)
return {"success": True, "result": "ok", "logs": ""}
with patch("lfx.cli.run.run_flow", new=AsyncMock(side_effect=fake_run_flow)):
run(
script_path=simple_chat_script,
input_value="hi",
input_value_option=None,
verbose=False,
output_format="json",
flow_json=None,
stdin=False,
session_id=None,
)
assert captured.get("session_id") is None
def test_execute_without_input_value(self, simple_chat_script, capsys):
"""Test executing without providing input value."""
# Test that execution without input either succeeds or fails gracefully
with contextlib.suppress(typer.Exit):
run(
script_path=simple_chat_script,
input_value=None,
input_value_option=None,
verbose=False,
output_format="json",
flow_json=None,
stdin=False,
)
# Test passes as long as no unhandled exceptions occur
# Check that output was produced
captured = capsys.readouterr()
if captured.out:
# Should be valid JSON when successful
try:
output_data = json.loads(captured.out)
assert isinstance(output_data, dict)
except json.JSONDecodeError:
assert len(captured.out.strip()) >= 0
Reference in New Issue View Git Blame Copy Permalink