test(cli): refactor test_components_path to use async thread and free port (#8748)

* Refactor test_components_path to run server in thread

The test now starts the CLI server in a separate thread on a free port, allowing asynchronous startup and avoiding port conflicts. This change improves test reliability and better simulates real server behavior.

* Update src/backend/tests/unit/test_cli.py

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

* Improve error handling in run_flow CLI test helper

The run_flow function now raises a RuntimeError with a detailed message if the CLI invocation fails, making test failures easier to diagnose.

---------

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
This commit is contained in:
Edwin Jose 2025-06-26 14:04:27 -05:00 committed by GitHub
commit 3b7b701813
No known key found for this signature in database
GPG key ID: B5690EEEBB952194

View file

@ -1,3 +1,7 @@
import socket
import threading
import time
import pytest
from langflow.__main__ import app
from langflow.services import deps
@ -11,16 +15,44 @@ def default_settings():
]
def get_free_port():
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("", 0))
return s.getsockname()[1]
def run_flow(runner, port, components_path, default_settings):
args = [
"run",
"--port",
str(port),
"--components-path",
str(components_path),
*default_settings,
]
result = runner.invoke(app, args)
if result.exit_code != 0:
msg = f"CLI failed with exit code {result.exit_code}: {result.output}"
raise RuntimeError(msg)
def test_components_path(runner, default_settings, tmp_path):
# create a "components" folder
temp_dir = tmp_path / "components"
temp_dir.mkdir(exist_ok=True)
result = runner.invoke(
app,
["run", "--components-path", str(temp_dir), *default_settings],
port = get_free_port()
thread = threading.Thread(
target=run_flow,
args=(runner, port, temp_dir, default_settings),
daemon=True,
)
assert result.exit_code == 0, result.stdout
thread.start()
# Give the server some time to start
time.sleep(5)
settings_service = deps.get_settings_service()
assert str(temp_dir) in settings_service.settings.components_path