jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

Merge remote-tracking branch 'origin/dev' into multipart_endpoint

Browse source
This commit is contained in:
Gabriel Luiz Freitas Almeida 2023-06-21 10:07:14 -03:00
commit 0dd16ee794
202 changed files with 21559 additions and 12347 deletions
Download patch file Download diff file Expand all files Collapse all files

42
tests/conftest.py
View file

@ -1,10 +1,14 @@
import json
from pathlib import Path
from typing import AsyncGenerator
from langflow.api.v1.flows import get_session
from langflow.graph.graph.base import Graph
import pytest
from fastapi.testclient import TestClient
from httpx import AsyncClient
from sqlmodel import SQLModel, Session, create_engine
from sqlmodel.pool import StaticPool
def pytest_configure():
@ -46,7 +50,6 @@ def client():
def get_graph(_type="basic"):
"""Get a graph from a json file"""
from langflow.graph.graph import Graph
if _type == "basic":
path = pytest.BASIC_EXAMPLE_PATH
@ -63,6 +66,12 @@ def get_graph(_type="basic"):
return Graph(nodes, edges)
@pytest.fixture
def basic_graph_data():
with open(pytest.BASIC_EXAMPLE_PATH, "r") as f:
return json.load(f)
@pytest.fixture
def basic_graph():
return get_graph()
@ -76,3 +85,34 @@ def complex_graph():
@pytest.fixture
def openapi_graph():
return get_graph("openapi")
@pytest.fixture
def json_flow():
with open(pytest.BASIC_EXAMPLE_PATH, "r") as f:
return f.read()
@pytest.fixture(name="session") #
def session_fixture(): #
engine = create_engine(
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
)
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
yield session
@pytest.fixture(name="client") #
def client_fixture(session: Session): #
def get_session_override(): #
return session
from langflow.main import create_app
app = create_app()
app.dependency_overrides[get_session] = get_session_override #
yield TestClient(app)
app.dependency_overrides.clear() #

510
tests/data/basic_example.json
View file

File diff suppressed because one or more lines are too long

24
tests/data/complex_example.json
View file

@ -197,7 +197,7 @@
"y": 136.29836646158452
},
"data": {
"type": "PythonFunction",
"type": "PythonFunctionTool",
"node": {
"template": {
"code": {
@ -210,6 +210,26 @@
"type": "str",
"list": false
},
"description": {
"required": true,
"placeholder": "",
"show": true,
"multiline": true,
"value": "My description",
"name": "description",
"type": "str",
"list": false
},
"name": {
"required": true,
"placeholder": "",
"show": true,
"multiline": true,
"value": "My Tool",
"name": "name",
"type": "str",
"list": false
},
"_type": "python_function"
},
"description": "Python function to be executed.",
@ -332,7 +352,7 @@
"type": "str",
"list": false
},
"_type": "zero_shot"
"_type": "ZeroShotPrompt"
},
"description": "Prompt template for Zero Shot Agent.",
"base_classes": [

22
tests/test_agents_template.py
View file

@ -1,19 +1,8 @@
from fastapi.testclient import TestClient
from langflow.settings import settings
# check that all agents are in settings.agents
# are in json_response["agents"]
def test_agents_settings(client: TestClient):
response = client.get("/all")
assert response.status_code == 200
json_response = response.json()
agents = json_response["agents"]
assert set(agents.keys()) == set(settings.agents)
def test_zero_shot_agent(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
agents = json_response["agents"]
@ -52,7 +41,7 @@ def test_zero_shot_agent(client: TestClient):
def test_json_agent(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
agents = json_response["agents"]
@ -87,7 +76,7 @@ def test_json_agent(client: TestClient):
def test_csv_agent(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
agents = json_response["agents"]
@ -126,12 +115,12 @@ def test_csv_agent(client: TestClient):
def test_initialize_agent(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
agents = json_response["agents"]
initialize_agent = agents["initialize_agent"]
initialize_agent = agents["AgentInitializer"]
assert initialize_agent["base_classes"] == ["AgentExecutor", "function"]
template = initialize_agent["template"]
@ -147,6 +136,7 @@ def test_initialize_agent(client: TestClient):
"react-docstore",
"self-ask-with-search",
"conversational-react-description",
"openai-functions",
],
"name": "agent",
"type": "str",

17
tests/test_cache.py
View file

@ -1,10 +1,9 @@
import json
from langflow.graph import Graph
import pytest
from langflow.interface.run import (
build_graph,
build_langchain_object_with_caching,
load_or_build_langchain_object,
)
@ -41,18 +40,6 @@ def langchain_objects_are_equal(obj1, obj2):
return str(obj1) == str(obj2)
# Test load_or_build_langchain_object
def test_load_or_build_langchain_object_first_message_true(basic_data_graph):
build_langchain_object_with_caching.clear_cache()
graph = load_or_build_langchain_object(basic_data_graph, is_first_message=True)
assert graph is not None
def test_load_or_build_langchain_object_first_message_false(basic_data_graph):
graph = load_or_build_langchain_object(basic_data_graph, is_first_message=False)
assert graph is not None
# Test build_langchain_object_with_caching
def test_build_langchain_object_with_caching(basic_data_graph):
build_langchain_object_with_caching.clear_cache()
@ -62,7 +49,7 @@ def test_build_langchain_object_with_caching(basic_data_graph):
# Test build_graph
def test_build_graph(basic_data_graph):
graph = build_graph(basic_data_graph)
graph = Graph.from_payload(basic_data_graph)
assert graph is not None
assert len(graph.nodes) == len(basic_data_graph["nodes"])
assert len(graph.edges) == len(basic_data_graph["edges"])

101
tests/test_chains_template.py
View file

@ -3,7 +3,7 @@ from langflow.settings import settings
def test_chains_settings(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
chains = json_response["chains"]
@ -12,20 +12,21 @@ def test_chains_settings(client: TestClient):
# Test the ConversationChain object
def test_conversation_chain(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
chains = json_response["chains"]
chain = chains["ConversationChain"]
# Test the base classes, template, memory, verbose, llm, input_key, output_key, and _type objects
assert set(chain["base_classes"]) == {
"function",
"LLMChain",
"ConversationChain",
"LLMChain",
"Chain",
"Serializable",
"function",
}
template = chain["template"]
assert template["memory"] == {
"required": False,
@ -94,14 +95,20 @@ def test_conversation_chain(client: TestClient):
def test_llm_chain(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
chains = json_response["chains"]
chain = chains["LLMChain"]
# Test the base classes, template, memory, verbose, llm, input_key, output_key, and _type objects
assert set(chain["base_classes"]) == {"function", "LLMChain", "Chain"}
assert set(chain["base_classes"]) == {
"Serializable",
"function",
"LLMChain",
"Chain",
}
template = chain["template"]
assert template["memory"] == {
"required": False,
@ -152,38 +159,21 @@ def test_llm_chain(client: TestClient):
def test_llm_checker_chain(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
chains = json_response["chains"]
chain = chains["LLMCheckerChain"]
# Test the base classes, template, memory, verbose, llm, input_key, output_key, and _type objects
assert set(chain["base_classes"]) == {"function", "LLMCheckerChain", "Chain"}
assert set(chain["base_classes"]) == {
"Serializable",
"function",
"LLMCheckerChain",
"Chain",
}
template = chain["template"]
assert template["memory"] == {
"required": False,
"placeholder": "",
"show": True,
"multiline": False,
"password": False,
"name": "memory",
"type": "BaseMemory",
"list": False,
"advanced": False,
}
assert template["verbose"] == {
"required": False,
"placeholder": "",
"show": True,
"multiline": False,
"value": False,
"password": False,
"name": "verbose",
"type": "bool",
"list": False,
"advanced": True,
}
assert template["llm"] == {
"required": True,
"placeholder": "",
@ -195,48 +185,27 @@ def test_llm_checker_chain(client: TestClient):
"list": False,
"advanced": False,
}
assert template["input_key"] == {
"required": True,
"placeholder": "",
"show": True,
"multiline": False,
"value": "query",
"password": False,
"name": "input_key",
"type": "str",
"list": False,
"advanced": True,
}
assert template["output_key"] == {
"required": True,
"placeholder": "",
"show": True,
"multiline": False,
"value": "result",
"password": False,
"name": "output_key",
"type": "str",
"list": False,
"advanced": True,
}
assert template["_type"] == "LLMCheckerChain"
# Test the description object
assert (
chain["description"] == "Chain for question-answering with self-verification."
)
assert chain["description"] == ""
def test_llm_math_chain(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
chains = json_response["chains"]
chain = chains["LLMMathChain"]
# Test the base classes, template, memory, verbose, llm, input_key, output_key, and _type objects
assert set(chain["base_classes"]) == {"function", "LLMMathChain", "Chain"}
assert set(chain["base_classes"]) == {
"function",
"Serializable",
"LLMMathChain",
"Chain",
}
template = chain["template"]
assert template["memory"] == {
"required": False,
@ -306,7 +275,7 @@ def test_llm_math_chain(client: TestClient):
def test_series_character_chain(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
chains = json_response["chains"]
@ -368,7 +337,7 @@ def test_series_character_chain(client: TestClient):
def test_mid_journey_prompt_chain(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
chains = json_response["chains"]
@ -407,7 +376,7 @@ def test_mid_journey_prompt_chain(client: TestClient):
def test_time_travel_guide_chain(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
chains = json_response["chains"]
@ -450,4 +419,4 @@ def test_time_travel_guide_chain(client: TestClient):
"advanced": False,
}
assert chain["description"] == "Time travel guide chain to be used in the flow."
assert chain["description"] == "Time travel guide chain."

23
tests/test_custom_types.py
View file

@ -1,13 +1,32 @@
# Test this:
from langflow.interface.importing.utils import get_function
import pytest
from langflow.interface.tools.custom import PythonFunction
from langflow.interface.tools.custom import PythonFunctionTool, PythonFunction
from langflow.utils import constants
def test_python_function_tool():
"""Test Python function"""
code = constants.DEFAULT_PYTHON_FUNCTION
func = get_function(code)
func = PythonFunctionTool(name="Test", description="Testing", code=code, func=func)
assert func("text") == "text"
# the tool decorator should raise an error if
# the function is not str -> str
# This raises ValidationError
with pytest.raises(SyntaxError):
code = pytest.CODE_WITH_SYNTAX_ERROR
func = get_function(code)
func = PythonFunctionTool(
name="Test", description="Testing", code=code, func=func
)
def test_python_function():
"""Test Python function"""
func = PythonFunction(code=constants.DEFAULT_PYTHON_FUNCTION)
assert func.get_function()("text") == "text"
assert get_function(func.code)("text") == "text"
# the tool decorator should raise an error if
# the function is not str -> str

340
tests/test_database.py Normal file
View file

@ -0,0 +1,340 @@
import json
import pytest
from uuid import UUID, uuid4
from sqlalchemy.orm import Session
from fastapi.testclient import TestClient
from fastapi.encoders import jsonable_encoder
from langflow.api.v1.schemas import FlowListCreate
from langflow.database.models.flow import Flow, FlowCreate, FlowUpdate
from langflow.database.models.flow_style import (
FlowStyleCreate,
FlowStyleRead,
FlowStyleUpdate,
)
@pytest.fixture(scope="module")
def json_style():
# class FlowStyleBase(SQLModel):
# color: str = Field(index=True)
# emoji: str = Field(index=False)
# flow_id: UUID = Field(default=None, foreign_key="flow.id")
return json.dumps(
{
"color": "red",
"emoji": "👍",
}
)
def test_create_flow(client: TestClient, json_flow: str):
flow = json.loads(json_flow)
data = flow["data"]
flow = FlowCreate(name="Test Flow", description="description", data=data)
response = client.post("api/v1/flows/", json=flow.dict())
assert response.status_code == 201
assert response.json()["name"] == flow.name
assert response.json()["data"] == flow.data
# flow is optional so we can create a flow without a flow
flow = FlowCreate(name="Test Flow")
response = client.post("api/v1/flows/", json=flow.dict(exclude_unset=True))
assert response.status_code == 201
assert response.json()["name"] == flow.name
assert response.json()["data"] == flow.data
def test_read_flows(client: TestClient, json_flow: str):
flow_data = json.loads(json_flow)
data = flow_data["data"]
flow = FlowCreate(name="Test Flow", description="description", data=data)
response = client.post("api/v1/flows/", json=flow.dict())
assert response.status_code == 201
assert response.json()["name"] == flow.name
assert response.json()["data"] == flow.data
flow_style = FlowStyleCreate(color="red", emoji="👍", flow_id=response.json()["id"])
response = client.post(
"api/v1/flow_styles/", json=jsonable_encoder(flow_style.dict())
)
assert response.status_code == 200
assert response.json()["color"] == flow_style.color
assert response.json()["emoji"] == flow_style.emoji
assert response.json()["flow_id"] == str(flow_style.flow_id)
flow = FlowCreate(name="Test Flow", description="description", data=data)
response = client.post("api/v1/flows/", json=flow.dict())
assert response.status_code == 201
assert response.json()["name"] == flow.name
assert response.json()["data"] == flow.data
# Now we need to create FlowStyle objects for each Flow
flow_style = FlowStyleCreate(
color="green", emoji="👍", flow_id=response.json()["id"]
)
response = client.post(
"api/v1/flow_styles/", json=jsonable_encoder(flow_style.dict())
)
assert response.status_code == 200
assert response.json()["color"] == flow_style.color
assert response.json()["emoji"] == flow_style.emoji
assert response.json()["flow_id"] == str(flow_style.flow_id)
response = client.get("api/v1/flows/")
assert response.status_code == 200
assert len(response.json()) > 0
def test_read_flow(client: TestClient, json_flow: str):
flow = json.loads(json_flow)
data = flow["data"]
flow = FlowCreate(name="Test Flow", description="description", data=data)
response = client.post("api/v1/flows/", json=flow.dict())
flow_id = response.json()["id"] # flow_id should be a UUID but is a string
# turn it into a UUID
flow_id = UUID(flow_id)
flow_style = FlowStyleCreate(color="green", emoji="👍", flow_id=flow_id)
response = client.post(
"api/v1/flow_styles/", json=jsonable_encoder(flow_style.dict())
)
assert response.status_code == 200
response_json = response.json()
assert response_json["color"] == flow_style.color
assert response_json["emoji"] == flow_style.emoji
assert response_json["flow_id"] == str(flow_style.flow_id)
response = client.get(f"api/v1/flows/{flow_id}")
assert response.status_code == 200
assert response.json()["name"] == flow.name
assert response.json()["data"] == flow.data
assert response.json()["style"]["color"] == flow_style.color
def test_update_flow(client: TestClient, json_flow: str):
flow = json.loads(json_flow)
data = flow["data"]
flow = FlowCreate(name="Test Flow", description="description", data=data)
response = client.post("api/v1/flows/", json=flow.dict())
flow_id = response.json()["id"]
updated_flow = FlowUpdate(
name="Updated Flow",
description="updated description",
data=data,
)
response = client.patch(f"api/v1/flows/{flow_id}", json=updated_flow.dict())
assert response.status_code == 200
assert response.json()["name"] == updated_flow.name
assert response.json()["description"] == updated_flow.description
# assert response.json()["data"] == updated_flow.data
def test_delete_flow(client: TestClient, json_flow: str):
flow = json.loads(json_flow)
data = flow["data"]
flow = FlowCreate(name="Test Flow", description="description", data=data)
response = client.post("api/v1/flows/", json=flow.dict())
flow_id = response.json()["id"]
response = client.delete(f"api/v1/flows/{flow_id}")
assert response.status_code == 200
assert response.json()["message"] == "Flow deleted successfully"
def test_create_flows(client: TestClient, session: Session, json_flow: str):
flow = json.loads(json_flow)
data = flow["data"]
# Create test data
flow_list = FlowListCreate(
flows=[
FlowCreate(name="Flow 1", description="description", data=data),
FlowCreate(name="Flow 2", description="description", data=data),
]
)
# Make request to endpoint
response = client.post("api/v1/flows/batch/", json=flow_list.dict())
# Check response status code
assert response.status_code == 201
# Check response data
response_data = response.json()
assert len(response_data) == 2
assert response_data[0]["name"] == "Flow 1"
assert response_data[0]["description"] == "description"
assert response_data[0]["data"] == data
assert response_data[1]["name"] == "Flow 2"
assert response_data[1]["description"] == "description"
assert response_data[1]["data"] == data
def test_upload_file(client: TestClient, session: Session, json_flow: str):
flow = json.loads(json_flow)
data = flow["data"]
# Create test data
flow_list = FlowListCreate(
flows=[
FlowCreate(name="Flow 1", description="description", data=data),
FlowCreate(name="Flow 2", description="description", data=data),
]
)
file_contents = json.dumps(flow_list.dict())
response = client.post(
"api/v1/flows/upload/",
files={"file": ("examples.json", file_contents, "application/json")},
)
# Check response status code
assert response.status_code == 201
# Check response data
response_data = response.json()
assert len(response_data) == 2
assert response_data[0]["name"] == "Flow 1"
assert response_data[0]["description"] == "description"
assert response_data[0]["data"] == data
assert response_data[1]["name"] == "Flow 2"
assert response_data[1]["description"] == "description"
assert response_data[1]["data"] == data
def test_download_file(client: TestClient, session: Session, json_flow):
flow = json.loads(json_flow)
data = flow["data"]
# Create test data
flow_list = FlowListCreate(
flows=[
FlowCreate(name="Flow 1", description="description", data=data),
FlowCreate(name="Flow 2", description="description", data=data),
]
)
for flow in flow_list.flows:
db_flow = Flow.from_orm(flow)
session.add(db_flow)
session.commit()
# Make request to endpoint
response = client.get("api/v1/flows/download/")
# Check response status code
assert response.status_code == 200
# Check response data
response_data = response.json()["flows"]
assert len(response_data) == 2
assert response_data[0]["name"] == "Flow 1"
assert response_data[0]["description"] == "description"
assert response_data[0]["data"] == data
assert response_data[1]["name"] == "Flow 2"
assert response_data[1]["description"] == "description"
assert response_data[1]["data"] == data
def test_create_flow_with_invalid_data(client: TestClient):
flow = {"name": "a" * 256, "data": "Invalid flow data"}
response = client.post("api/v1/flows/", json=flow)
assert response.status_code == 422
def test_get_nonexistent_flow(client: TestClient):
uuid = uuid4()
response = client.get(f"api/v1/flows/{uuid}")
assert response.status_code == 404
def test_update_flow_idempotency(client: TestClient, json_flow: str):
flow_data = json.loads(json_flow)
data = flow_data["data"]
flow_data = FlowCreate(name="Test Flow", description="description", data=data)
response = client.post("api/v1/flows/", json=flow_data.dict())
flow_id = response.json()["id"]
updated_flow = FlowCreate(name="Updated Flow", description="description", data=data)
response1 = client.put(f"api/v1/flows/{flow_id}", json=updated_flow.dict())
response2 = client.put(f"api/v1/flows/{flow_id}", json=updated_flow.dict())
assert response1.json() == response2.json()
def test_update_nonexistent_flow(client: TestClient, json_flow: str):
flow_data = json.loads(json_flow)
data = flow_data["data"]
uuid = uuid4()
updated_flow = FlowCreate(
name="Updated Flow",
description="description",
data=data,
)
response = client.patch(f"api/v1/flows/{uuid}", json=updated_flow.dict())
assert response.status_code == 404
def test_delete_nonexistent_flow(client: TestClient):
uuid = uuid4()
response = client.delete(f"api/v1/flows/{uuid}")
assert response.status_code == 404
def test_read_empty_flows(client: TestClient):
response = client.get("api/v1/flows/")
assert response.status_code == 200
assert len(response.json()) == 0
def test_create_flow_style(client: TestClient):
flow_style = FlowStyleCreate(color="red", emoji="🔴")
response = client.post("api/v1/flow_styles/", json=flow_style.dict())
assert response.status_code == 200
created_flow_style = FlowStyleRead(**response.json())
assert created_flow_style.color == flow_style.color
assert created_flow_style.emoji == flow_style.emoji
def test_read_flow_styles(client: TestClient):
response = client.get("api/v1/flow_styles/")
assert response.status_code == 200
flow_styles = [FlowStyleRead(**flow_style) for flow_style in response.json()]
assert not flow_styles
# Create test data
flow_style = FlowStyleCreate(color="red", emoji="🔴")
response = client.post("api/v1/flow_styles/", json=flow_style.dict())
assert response.status_code == 200
# Check response data
response = client.get("api/v1/flow_styles/")
assert response.status_code == 200
flow_styles = [FlowStyleRead(**flow_style) for flow_style in response.json()]
assert len(flow_styles) == 1
assert flow_styles[0].color == flow_style.color
assert flow_styles[0].emoji == flow_style.emoji
def test_read_flow_style(client: TestClient):
flow_style = FlowStyleCreate(color="red", emoji="🔴")
response = client.post("api/v1/flow_styles/", json=flow_style.dict())
created_flow_style = FlowStyleRead(**response.json())
response = client.get(f"api/v1/flow_styles/{created_flow_style.id}")
assert response.status_code == 200
read_flow_style = FlowStyleRead(**response.json())
assert read_flow_style == created_flow_style
def test_update_flow_style(client: TestClient):
flow_style = FlowStyleCreate(color="red", emoji="🔴")
response = client.post("api/v1/flow_styles/", json=flow_style.dict())
created_flow_style = FlowStyleRead(**response.json())
to_update_flow_style = FlowStyleUpdate(color="blue")
response = client.patch(
f"api/v1/flow_styles/{created_flow_style.id}", json=to_update_flow_style.dict()
)
assert response.status_code == 200
updated_flow_style = FlowStyleRead(**response.json())
assert updated_flow_style.color == "blue"
assert updated_flow_style.emoji == flow_style.emoji
def test_delete_flow_style(client: TestClient):
flow_style = FlowStyleCreate(color="red", emoji="🔴")
response = client.post("api/v1/flow_styles/", json=flow_style.dict())
created_flow_style = FlowStyleRead(**response.json())
response = client.delete(f"api/v1/flow_styles/{created_flow_style.id}")
assert response.status_code == 200
assert response.json() == {"message": "FlowStyle deleted successfully"}
response = client.get(f"api/v1/flow_styles/{created_flow_style.id}")
assert response.status_code == 404

20
tests/test_endpoints.py
View file

@ -4,7 +4,7 @@ from langflow.interface.tools.constants import CUSTOM_TOOLS
def test_get_all(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
# We need to test the custom nodes
@ -21,7 +21,7 @@ import math
def square(x):
return x ** 2
"""
response1 = client.post("/validate/code", json={"code": code1})
response1 = client.post("api/v1/validate/code", json={"code": code1})
assert response1.status_code == 200
assert response1.json() == {"imports": {"errors": []}, "function": {"errors": []}}
@ -32,7 +32,7 @@ import non_existent_module
def square(x):
return x ** 2
"""
response2 = client.post("/validate/code", json={"code": code2})
response2 = client.post("api/v1/validate/code", json={"code": code2})
assert response2.status_code == 200
assert response2.json() == {
"imports": {"errors": ["No module named 'non_existent_module'"]},
@ -46,7 +46,7 @@ import math
def square(x)
return x ** 2
"""
response3 = client.post("/validate/code", json={"code": code3})
response3 = client.post("api/v1/validate/code", json={"code": code3})
assert response3.status_code == 200
assert response3.json() == {
"imports": {"errors": []},
@ -54,11 +54,11 @@ def square(x)
}
# Test case with invalid JSON payload
response4 = client.post("/validate/code", json={"invalid_key": code1})
response4 = client.post("api/v1/validate/code", json={"invalid_key": code1})
assert response4.status_code == 422
# Test case with an empty code string
response5 = client.post("/validate/code", json={"code": ""})
response5 = client.post("api/v1/validate/code", json={"code": ""})
assert response5.status_code == 200
assert response5.json() == {"imports": {"errors": []}, "function": {"errors": []}}
@ -69,7 +69,7 @@ import math
def square(x)
return x ** 2
"""
response6 = client.post("/validate/code", json={"code": code6})
response6 = client.post("api/v1/validate/code", json={"code": code6})
assert response6.status_code == 200
assert response6.json() == {
"imports": {"errors": []},
@ -95,13 +95,13 @@ INVALID_PROMPT = "This is an invalid prompt without any input variable."
def test_valid_prompt(client: TestClient):
response = client.post("/validate/prompt", json={"template": VALID_PROMPT})
response = client.post("api/v1/validate/prompt", json={"template": VALID_PROMPT})
assert response.status_code == 200
assert response.json() == {"input_variables": ["product"]}
def test_invalid_prompt(client: TestClient):
response = client.post("/validate/prompt", json={"template": INVALID_PROMPT})
response = client.post("api/v1/validate/prompt", json={"template": INVALID_PROMPT})
assert response.status_code == 200
assert response.json() == {"input_variables": []}
@ -116,7 +116,7 @@ def test_invalid_prompt(client: TestClient):
],
)
def test_various_prompts(client, prompt, expected_input_variables):
response = client.post("/validate/prompt", json={"template": prompt})
response = client.post("api/v1/validate/prompt", json={"template": prompt})
assert response.status_code == 200
assert response.json() == {
"input_variables": expected_input_variables,

83
tests/test_graph.py
View file

@ -1,18 +1,22 @@
from typing import Type, Union
from langflow.graph.edge.base import Edge
from langflow.graph.vertex.base import Vertex
import pytest
from langchain.chains.base import Chain
from langchain.llms.fake import FakeListLLM
from langflow.graph import Edge, Graph, Node
from langflow.graph.nodes import (
AgentNode,
ChainNode,
LLMNode,
PromptNode,
ToolNode,
WrapperNode,
from langflow.graph import Graph
from langflow.graph.vertex.types import (
AgentVertex,
ChainVertex,
FileToolVertex,
LLMVertex,
PromptVertex,
ToolkitVertex,
ToolVertex,
WrapperVertex,
)
from langflow.interface.run import get_result_and_thought
from langflow.processing.process import get_result_and_thought
from langflow.utils.payload import get_root_node
# Test cases for the graph module
@ -21,7 +25,7 @@ from langflow.utils.payload import get_root_node
# BASIC_EXAMPLE_PATH, COMPLEX_EXAMPLE_PATH, OPENAPI_EXAMPLE_PATH
def get_node_by_type(graph, node_type: Type[Node]) -> Union[Node, None]:
def get_node_by_type(graph, node_type: Type[Vertex]) -> Union[Vertex, None]:
"""Get a node by type"""
return next((node for node in graph.nodes if isinstance(node, node_type)), None)
@ -31,7 +35,7 @@ def test_graph_structure(basic_graph):
assert len(basic_graph.nodes) > 0
assert len(basic_graph.edges) > 0
for node in basic_graph.nodes:
assert isinstance(node, Node)
assert isinstance(node, Vertex)
for edge in basic_graph.edges:
assert isinstance(edge, Edge)
assert edge.source in basic_graph.nodes
@ -154,14 +158,16 @@ def test_get_node_neighbors_complex(complex_graph):
tool_neighbors = complex_graph.get_nodes_with_target(tool)
assert tool_neighbors is not None
# Check if there is a PythonFunction in the tool's neighbors
assert any("PythonFunction" in neighbor.data["type"] for neighbor in tool_neighbors)
assert any(
"PythonFunctionTool" in neighbor.data["type"] for neighbor in tool_neighbors
)
def test_get_node(basic_graph):
"""Test getting a single node"""
node_id = basic_graph.nodes[0].id
node = basic_graph.get_node(node_id)
assert isinstance(node, Node)
assert isinstance(node, Vertex)
assert node.id == node_id
@ -170,7 +176,7 @@ def test_build_nodes(basic_graph):
assert len(basic_graph.nodes) == len(basic_graph._nodes)
for node in basic_graph.nodes:
assert isinstance(node, Node)
assert isinstance(node, Vertex)
def test_build_edges(basic_graph):
@ -178,8 +184,8 @@ def test_build_edges(basic_graph):
assert len(basic_graph.edges) == len(basic_graph._edges)
for edge in basic_graph.edges:
assert isinstance(edge, Edge)
assert isinstance(edge.source, Node)
assert isinstance(edge.target, Node)
assert isinstance(edge.source, Vertex)
assert isinstance(edge.target, Vertex)
def test_get_root_node(basic_graph, complex_graph):
@ -187,13 +193,13 @@ def test_get_root_node(basic_graph, complex_graph):
assert isinstance(basic_graph, Graph)
root = get_root_node(basic_graph)
assert root is not None
assert isinstance(root, Node)
assert isinstance(root, Vertex)
assert root.data["type"] == "TimeTravelGuideChain"
# For complex example, the root node is a ZeroShotAgent too
assert isinstance(complex_graph, Graph)
root = get_root_node(complex_graph)
assert root is not None
assert isinstance(root, Node)
assert isinstance(root, Vertex)
assert root.data["type"] == "ZeroShotAgent"
@ -235,11 +241,10 @@ def test_build_params(basic_graph):
assert "memory" in root.params
def test_build(basic_graph, complex_graph, openapi_graph):
def test_build(basic_graph, complex_graph):
"""Test Node's build method"""
assert_agent_was_built(basic_graph)
assert_agent_was_built(complex_graph)
# assert_agent_was_built(openapi_graph)
def assert_agent_was_built(graph):
@ -253,14 +258,14 @@ def assert_agent_was_built(graph):
def test_agent_node_build(complex_graph):
agent_node = get_node_by_type(complex_graph, AgentNode)
agent_node = get_node_by_type(complex_graph, AgentVertex)
assert agent_node is not None
built_object = agent_node.build()
assert built_object is not None
def test_tool_node_build(complex_graph):
tool_node = get_node_by_type(complex_graph, ToolNode)
tool_node = get_node_by_type(complex_graph, ToolVertex)
assert tool_node is not None
built_object = tool_node.build()
assert built_object is not None
@ -268,7 +273,7 @@ def test_tool_node_build(complex_graph):
def test_chain_node_build(complex_graph):
chain_node = get_node_by_type(complex_graph, ChainNode)
chain_node = get_node_by_type(complex_graph, ChainVertex)
assert chain_node is not None
built_object = chain_node.build()
assert built_object is not None
@ -276,7 +281,7 @@ def test_chain_node_build(complex_graph):
def test_prompt_node_build(complex_graph):
prompt_node = get_node_by_type(complex_graph, PromptNode)
prompt_node = get_node_by_type(complex_graph, PromptVertex)
assert prompt_node is not None
built_object = prompt_node.build()
assert built_object is not None
@ -284,31 +289,31 @@ def test_prompt_node_build(complex_graph):
def test_llm_node_build(basic_graph):
llm_node = get_node_by_type(basic_graph, LLMNode)
llm_node = get_node_by_type(basic_graph, LLMVertex)
assert llm_node is not None
built_object = llm_node.build()
assert built_object is not None
# Add any further assertions specific to the LLMNode's build() method
# def test_toolkit_node_build(openapi_graph):
# toolkit_node = get_node_by_type(openapi_graph, ToolkitNode)
# assert toolkit_node is not None
# built_object = toolkit_node.build()
# assert built_object is not None
# Add any further assertions specific to the ToolkitNode's build() method
def test_toolkit_node_build(openapi_graph):
toolkit_node = get_node_by_type(openapi_graph, ToolkitVertex)
assert toolkit_node is not None
built_object = toolkit_node.build()
assert built_object is not None
# Add any further assertions specific to the ToolkitNode's build() method
# def test_file_tool_node_build(openapi_graph):
# file_tool_node = get_node_by_type(openapi_graph, FileToolNode)
# assert file_tool_node is not None
# built_object = file_tool_node.build()
# assert built_object is not None
# Add any further assertions specific to the FileToolNode's build() method
def test_file_tool_node_build(openapi_graph):
file_tool_node = get_node_by_type(openapi_graph, FileToolVertex)
assert file_tool_node is not None
built_object = file_tool_node.build()
assert built_object is not None
# Add any further assertions specific to the FileToolNode's build() method
def test_wrapper_node_build(openapi_graph):
wrapper_node = get_node_by_type(openapi_graph, WrapperNode)
wrapper_node = get_node_by_type(openapi_graph, WrapperVertex)
assert wrapper_node is not None
built_object = wrapper_node.build()
assert built_object is not None
@ -323,7 +328,7 @@ def test_get_result_and_thought(basic_graph):
message = "Hello"
# Find the node that is an LLMNode and change the
# _built_object to a FakeListLLM
llm_node = get_node_by_type(basic_graph, LLMNode)
llm_node = get_node_by_type(basic_graph, LLMVertex)
assert llm_node is not None
llm_node._built_object = FakeListLLM(responses=responses)
llm_node._built = True

95
tests/test_llms_template.py
View file

@ -3,7 +3,7 @@ from langflow.settings import settings
def test_llms_settings(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
llms = json_response["llms"]
@ -11,7 +11,7 @@ def test_llms_settings(client: TestClient):
# def test_hugging_face_hub(client: TestClient):
# response = client.get("/all")
# response = client.get("api/v1/all")
# assert response.status_code == 200
# json_response = response.json()
# language_models = json_response["llms"]
@ -103,7 +103,7 @@ def test_llms_settings(client: TestClient):
def test_openai(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
language_models = json_response["llms"]
@ -333,7 +333,7 @@ def test_openai(client: TestClient):
def test_chat_open_ai(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
language_models = json_response["llms"]
@ -371,7 +371,16 @@ def test_chat_open_ai(client: TestClient):
"multiline": False,
"value": "gpt-3.5-turbo",
"password": False,
"options": ["gpt-3.5-turbo", "gpt-4", "gpt-4-32k"],
"options": [
"gpt-3.5-turbo-0613",
"gpt-3.5-turbo",
"gpt-3.5-turbo-16k-0613",
"gpt-3.5-turbo-16k",
"gpt-4-0613",
"gpt-4-32k-0613",
"gpt-4",
"gpt-4-32k",
],
"name": "model_name",
"type": "str",
"list": True,
@ -478,7 +487,83 @@ def test_chat_open_ai(client: TestClient):
== "Wrapper around OpenAI Chat large language models." # noqa E501
)
assert set(model["base_classes"]) == {
"Serializable",
"BaseChatModel",
"ChatOpenAI",
"BaseLanguageModel",
}
# Commenting this out for now, as it requires to activate the nodes
# def test_azure_open_ai(client: TestClient):
# response = client.get("/all")
# assert response.status_code == 200
# json_response = response.json()
# language_models = json_response["llms"]
# model = language_models["AzureOpenAI"]
# template = model["template"]
# assert template["model_name"]["show"] is False
# assert template["deployment_name"] == {
# "required": False,
# "placeholder": "",
# "show": True,
# "multiline": False,
# "value": "",
# "password": False,
# "name": "deployment_name",
# "advanced": False,
# "type": "str",
# "list": False,
# }
# def test_azure_chat_open_ai(client: TestClient):
# response = client.get("/all")
# assert response.status_code == 200
# json_response = response.json()
# language_models = json_response["llms"]
# model = language_models["AzureChatOpenAI"]
# template = model["template"]
# assert template["model_name"]["show"] is False
# assert template["deployment_name"] == {
# "required": False,
# "placeholder": "",
# "show": True,
# "multiline": False,
# "value": "",
# "password": False,
# "name": "deployment_name",
# "advanced": False,
# "type": "str",
# "list": False,
# }
# assert template["openai_api_type"] == {
# "required": False,
# "placeholder": "",
# "show": False,
# "multiline": False,
# "value": "azure",
# "password": False,
# "name": "openai_api_type",
# "display_name": "OpenAI API Type",
# "advanced": False,
# "type": "str",
# "list": False,
# }
# assert template["openai_api_version"] == {
# "required": False,
# "placeholder": "",
# "show": True,
# "multiline": False,
# "value": "2023-03-15-preview",
# "password": False,
# "name": "openai_api_version",
# "display_name": "OpenAI API Version",
# "advanced": False,
# "type": "str",
# "list": False,
# }

11
tests/test_loading.py
View file

@ -2,7 +2,7 @@ import json
import pytest
from langchain.chains.base import Chain
from langflow import load_flow_from_json
from langflow.processing.process import load_flow_from_json
from langflow.graph import Graph
from langflow.utils.payload import get_root_node
@ -14,6 +14,15 @@ def test_load_flow_from_json():
assert isinstance(loaded, Chain)
def test_load_flow_from_json_with_tweaks():
"""Test loading a flow from a json file and applying tweaks"""
tweaks = {"dndnode_82": {"model_name": "test model"}}
loaded = load_flow_from_json(pytest.BASIC_EXAMPLE_PATH, tweaks=tweaks)
assert loaded is not None
assert isinstance(loaded, Chain)
assert loaded.llm.model_name == "test model"
def test_get_root_node():
with open(pytest.BASIC_EXAMPLE_PATH, "r") as f:
flow_graph = json.load(f)

196
tests/test_process.py Normal file
View file

@ -0,0 +1,196 @@
from langflow.processing.process import process_tweaks
def test_no_tweaks():
graph_data = {
"data": {
"nodes": [
{
"id": "node1",
"data": {
"node": {
"template": {
"param1": {"value": 1},
"param2": {"value": 2},
}
}
},
},
{
"id": "node2",
"data": {
"node": {
"template": {
"param1": {"value": 3},
"param2": {"value": 4},
}
}
},
},
]
}
}
tweaks = {}
result = process_tweaks(graph_data, tweaks)
assert result == graph_data
def test_single_tweak():
graph_data = {
"data": {
"nodes": [
{
"id": "node1",
"data": {
"node": {
"template": {
"param1": {"value": 1},
"param2": {"value": 2},
}
}
},
},
{
"id": "node2",
"data": {
"node": {
"template": {
"param1": {"value": 3},
"param2": {"value": 4},
}
}
},
},
]
}
}
tweaks = {"node1": {"param1": 5}}
expected_result = {
"data": {
"nodes": [
{
"id": "node1",
"data": {
"node": {
"template": {
"param1": {"value": 5},
"param2": {"value": 2},
}
}
},
},
{
"id": "node2",
"data": {
"node": {
"template": {
"param1": {"value": 3},
"param2": {"value": 4},
}
}
},
},
]
}
}
result = process_tweaks(graph_data, tweaks)
assert result == expected_result
def test_multiple_tweaks():
graph_data = {
"data": {
"nodes": [
{
"id": "node1",
"data": {
"node": {
"template": {
"param1": {"value": 1},
"param2": {"value": 2},
}
}
},
},
{
"id": "node2",
"data": {
"node": {
"template": {
"param1": {"value": 3},
"param2": {"value": 4},
}
}
},
},
]
}
}
tweaks = {
"node1": {"param1": 5, "param2": 6},
"node2": {"param1": 7},
}
expected_result = {
"data": {
"nodes": [
{
"id": "node1",
"data": {
"node": {
"template": {
"param1": {"value": 5},
"param2": {"value": 6},
}
}
},
},
{
"id": "node2",
"data": {
"node": {
"template": {
"param1": {"value": 7},
"param2": {"value": 4},
}
}
},
},
]
}
}
result = process_tweaks(graph_data, tweaks)
assert result == expected_result
def test_tweak_not_in_template():
graph_data = {
"data": {
"nodes": [
{
"id": "node1",
"data": {
"node": {
"template": {
"param1": {"value": 1},
"param2": {"value": 2},
}
}
},
},
{
"id": "node2",
"data": {
"node": {
"template": {
"param1": {"value": 3},
"param2": {"value": 4},
}
}
},
},
]
}
}
tweaks = {"node1": {"param3": 5}}
result = process_tweaks(graph_data, tweaks)
assert result == graph_data

8
tests/test_prompts_template.py
View file

@ -3,7 +3,7 @@ from langflow.settings import settings
def test_prompts_settings(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
prompts = json_response["prompts"]
@ -11,7 +11,7 @@ def test_prompts_settings(client: TestClient):
def test_prompt_template(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
prompts = json_response["prompts"]
@ -89,7 +89,7 @@ def test_prompt_template(client: TestClient):
def test_few_shot_prompt_template(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
prompts = json_response["prompts"]
@ -168,7 +168,7 @@ def test_few_shot_prompt_template(client: TestClient):
def test_zero_shot_prompt(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
prompts = json_response["prompts"]

2
tests/test_vectorstore_template.py
View file

@ -5,7 +5,7 @@ from langflow.settings import settings
# check that all agents are in settings.agents
# are in json_response["agents"]
def test_vectorstores_settings(client: TestClient):
response = client.get("/all")
response = client.get("api/v1/all")
assert response.status_code == 200
json_response = response.json()
vectorstores = json_response["vectorstores"]

80
tests/test_websocket.py
View file

@ -1,47 +1,49 @@
import json
from unittest.mock import patch
from fastapi import WebSocketDisconnect
from fastapi.testclient import TestClient
# from langflow.chat.manager import ChatManager
import pytest
def test_websocket_connection(client: TestClient):
with client.websocket_connect("/chat/test_client") as websocket:
assert websocket.scope["client"] == ["testclient", 50000]
assert websocket.scope["path"] == "/chat/test_client"
def test_init_build(client):
response = client.post(
"api/v1/build/init", json={"id": "test", "data": {"key": "value"}}
)
assert response.status_code == 201
assert response.json() == {"flowId": "test"}
def test_chat_history(client: TestClient):
# Mock the process_graph function to return a specific value
with patch("langflow.api.chat_manager.process_graph") as mock_process_graph:
mock_process_graph.return_value = ("Hello, I'm a mock response!", "")
# def test_stream_build(client):
# client.post(
# "api/v1/build/init", json={"id": "stream_test", "data": {"key": "value"}}
# )
with client.websocket_connect("/chat/test_client") as websocket:
# First message should be the history
history = websocket.receive_json()
assert history == [] # Empty history
# Send a message
payload = {"message": "Hello"}
websocket.send_json(json.dumps(payload))
# # Test the stream
# response = client.get("api/v1/build/stream/stream_test")
# assert response.status_code == 200
# assert response.headers["content-type"] == "text/event-stream; charset=utf-8"
# Receive the response from the server
response = websocket.receive_json()
assert response == {
"is_bot": True,
"message": None,
"type": "start",
"intermediate_steps": "",
"files": [],
}
# Send another message
payload = {"message": "How are you?"}
websocket.send_json(json.dumps(payload))
# Receive the response from the server
response = websocket.receive_json()
assert response == {
"is_bot": True,
"message": "Hello, I'm a mock response!",
"type": "end",
"intermediate_steps": "",
"files": [],
}
def test_websocket_endpoint(client):
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect(
"api/v1/chat/non_existing_client_id"
) as websocket:
websocket.send_json({"type": "test"})
data = websocket.receive_json()
assert "Please, build the flow before sending messages" in data["message"]
def test_websocket_endpoint_after_build(client, basic_graph_data):
# Assuming your websocket_endpoint uses chat_manager which caches data from stream_build
client.post("api/v1/build/init", json=basic_graph_data)
client.get("api/v1/build/stream/websocket_test")
# There should be more to test here, but it depends on the inner workings of your websocket handler
# and how your chat_manager and other classes behave. The following is just an example structure.
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("api/v1/chat/websocket_test") as websocket:
websocket.send_json({"type": "test"})
# Perform assertions here, based on what you expect the websocket to return
# data = websocket.receive_json()
# assert ...