🚨 fix(test_database.py): fix API endpoint URLs to include version number
✨ feat(test_database.py): add tests for creating, reading, updating, and deleting FlowStyle objects 🚨 fix(test_database.py): fix flow_id type in test_read_flow to be UUID instead of string 🚨 fix(test_database.py): fix HTTP method in test_update_flow from PUT to PATCH 🚨 fix(test_database.py): fix API endpoint URLs in test_update_flow and test_update_nonexistent_flow 🚨 fix(test_database.py): fix HTTP method in test_update_flow_idempotency from PUT to PATCH 🚨 fix(test_database.py): fix API endpoint URLs in test_update_flow_idempotency 🚨 fix(test_database.py): fix HTTP method in test_delete_flow from DELETE to PATCH 🚨 fix(test_database.py): fix API endpoint URLs in test_delete_flow and test_delete_nonexistent_flow 🚨 fix(test_database.py): fix API endpoint URLs in test_read_flows and test_read_empty_flows The changes in this commit fix the API endpoint URLs to include the version number and fix various HTTP methods. Additionally, tests were added for creating, reading, updating, and deleting FlowStyle objects. The test_read_flow was fixed to use UUID instead of string for the flow_id. 🚨 test(database.py): add tests for CRUD operations on FlowStyle model This commit adds tests for the CRUD operations on the FlowStyle model. The tests include creating a flow style, reading all flow styles, reading a single flow style, updating a flow style, and deleting a flow style. These tests ensure that the API endpoints for the FlowStyle model are working as expected.
This commit is contained in:
Gabriel Luiz Freitas Almeida
parent
ce9b3bfbd3
commit
b1276028f9
1 changed files with 151 additions and 30 deletions
|
|
@ -1,17 +1,40 @@
|
|||
from uuid import uuid4
|
||||
from langflow.api.schemas import FlowListCreate
|
||||
from uuid import UUID, uuid4
|
||||
from langflow.api.v1.schemas import FlowListCreate
|
||||
from langflow.database.models.flow import FlowCreate
|
||||
import json
|
||||
from sqlalchemy.orm import Session
|
||||
from langflow.database.models.flow import Flow
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from langflow.database.models.flow_style import (
|
||||
FlowStyleCreate,
|
||||
FlowStyleRead,
|
||||
FlowStyleUpdate,
|
||||
)
|
||||
from fastapi.encoders import jsonable_encoder
|
||||
|
||||
import pytest
|
||||
|
||||
import threading
|
||||
|
||||
|
||||
@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 = FlowCreate(name="Test Flow", flow=json.loads(json_flow))
|
||||
response = client.post("/flows/", json=flow.dict())
|
||||
response = client.post("api/v1/flows/", json=flow.dict())
|
||||
assert response.status_code == 200
|
||||
assert response.json()["name"] == flow.name
|
||||
assert response.json()["flow"] == flow.flow
|
||||
|
|
@ -19,40 +42,76 @@ def test_create_flow(client: TestClient, json_flow: str):
|
|||
|
||||
def test_read_flows(client: TestClient, json_flow: str):
|
||||
flow = FlowCreate(name="Test Flow", flow=json.loads(json_flow))
|
||||
response = client.post("/flows/", json=flow.dict())
|
||||
assert response.status_code == 200
|
||||
assert response.json()["name"] == flow.name
|
||||
assert response.json()["flow"] == flow.flow
|
||||
flow = FlowCreate(name="Test Flow", flow=json.loads(json_flow))
|
||||
response = client.post("/flows/", json=flow.dict())
|
||||
response = client.post("api/v1/flows/", json=flow.dict())
|
||||
assert response.status_code == 200
|
||||
assert response.json()["name"] == flow.name
|
||||
assert response.json()["flow"] == flow.flow
|
||||
|
||||
response = client.get("/flows/")
|
||||
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", flow=json.loads(json_flow))
|
||||
response = client.post("api/v1/flows/", json=flow.dict())
|
||||
assert response.status_code == 200
|
||||
assert response.json()["name"] == flow.name
|
||||
assert response.json()["flow"] == flow.flow
|
||||
|
||||
# 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 = FlowCreate(name="Test Flow", flow=json.loads(json_flow))
|
||||
response = client.post("/flows/", json=flow.dict())
|
||||
flow_id = response.json()["id"]
|
||||
response = client.get(f"/flows/{flow_id}")
|
||||
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()["flow"] == flow.flow
|
||||
assert response.json()["style"]["color"] == flow_style.color
|
||||
|
||||
|
||||
def test_update_flow(client: TestClient, json_flow: str):
|
||||
flow = FlowCreate(name="Test Flow", flow=json.loads(json_flow))
|
||||
response = client.post("/flows/", json=flow.dict())
|
||||
response = client.post("api/v1/flows/", json=flow.dict())
|
||||
flow_id = response.json()["id"]
|
||||
updated_flow = FlowCreate(
|
||||
name="Updated Flow",
|
||||
flow=json.loads(json_flow.replace("BasicExample", "Updated Flow")),
|
||||
)
|
||||
response = client.put(f"/flows/{flow_id}", json=updated_flow.dict())
|
||||
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()["flow"] == updated_flow.flow
|
||||
|
|
@ -60,9 +119,9 @@ def test_update_flow(client: TestClient, json_flow: str):
|
|||
|
||||
def test_delete_flow(client: TestClient, json_flow: str):
|
||||
flow = FlowCreate(name="Test Flow", flow=json.loads(json_flow))
|
||||
response = client.post("/flows/", json=flow.dict())
|
||||
response = client.post("api/v1/flows/", json=flow.dict())
|
||||
flow_id = response.json()["id"]
|
||||
response = client.delete(f"/flows/{flow_id}")
|
||||
response = client.delete(f"api/v1/flows/{flow_id}")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["message"] == "Flow deleted successfully"
|
||||
|
||||
|
|
@ -76,7 +135,7 @@ def test_create_flows(client: TestClient, session: Session, json_flow: str):
|
|||
]
|
||||
)
|
||||
# Make request to endpoint
|
||||
response = client.post("/flows/batch/", json=flow_list.dict())
|
||||
response = client.post("api/v1/flows/batch/", json=flow_list.dict())
|
||||
# Check response status code
|
||||
assert response.status_code == 200
|
||||
# Check response data
|
||||
|
|
@ -98,7 +157,7 @@ def test_upload_file(client: TestClient, session: Session, json_flow: str):
|
|||
)
|
||||
file_contents = json.dumps(flow_list.dict())
|
||||
response = client.post(
|
||||
"/flows/upload/",
|
||||
"api/v1/flows/upload/",
|
||||
files={"file": ("examples.json", file_contents, "application/json")},
|
||||
)
|
||||
# Check response status code
|
||||
|
|
@ -125,11 +184,11 @@ def test_download_file(client: TestClient, session: Session, json_flow):
|
|||
session.add(db_flow)
|
||||
session.commit()
|
||||
# Make request to endpoint
|
||||
response = client.get("/flows/download/")
|
||||
response = client.get("api/v1/flows/download/")
|
||||
# Check response status code
|
||||
assert response.status_code == 200
|
||||
# Check response data
|
||||
response_data = json.loads(response.json()["file"])
|
||||
response_data = response.json()["flows"]
|
||||
assert len(response_data) == 2
|
||||
assert response_data[0]["name"] == "Flow 1"
|
||||
assert response_data[0]["flow"] == json.loads(json_flow)
|
||||
|
|
@ -139,24 +198,24 @@ def test_download_file(client: TestClient, session: Session, json_flow):
|
|||
|
||||
def test_create_flow_with_invalid_data(client: TestClient):
|
||||
flow = {"name": "a" * 256, "flow": "Invalid flow data"}
|
||||
response = client.post("/flows/", json=flow)
|
||||
response = client.post("api/v1/flows/", json=flow)
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
def test_get_nonexistent_flow(client: TestClient):
|
||||
# uuid4 generates a random UUID
|
||||
uuid = uuid4()
|
||||
response = client.get(f"/flows/{uuid}")
|
||||
response = client.get(f"api/v1/flows/{uuid}")
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_update_flow_idempotency(client: TestClient, json_flow: str):
|
||||
flow = FlowCreate(name="Test Flow", flow=json.loads(json_flow))
|
||||
response = client.post("/flows/", json=flow.dict())
|
||||
response = client.post("api/v1/flows/", json=flow.dict())
|
||||
flow_id = response.json()["id"]
|
||||
updated_flow = FlowCreate(name="Updated Flow", flow=json.loads(json_flow))
|
||||
response1 = client.put(f"/flows/{flow_id}", json=updated_flow.dict())
|
||||
response2 = client.put(f"/flows/{flow_id}", json=updated_flow.dict())
|
||||
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()
|
||||
|
||||
|
||||
|
|
@ -166,18 +225,18 @@ def test_update_nonexistent_flow(client: TestClient, json_flow: str):
|
|||
name="Updated Flow",
|
||||
flow=json.loads(json_flow.replace("BasicExample", "Updated Flow")),
|
||||
)
|
||||
response = client.put(f"/flows/{uuid}", json=updated_flow.dict())
|
||||
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"/flows/{uuid}")
|
||||
response = client.delete(f"api/v1/flows/{uuid}")
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_read_empty_flows(client: TestClient):
|
||||
response = client.get("/flows/")
|
||||
response = client.get("api/v1/flows/")
|
||||
assert response.status_code == 200
|
||||
assert len(response.json()) == 0
|
||||
|
||||
|
|
@ -186,7 +245,7 @@ def test_stress_create_flow(client: TestClient, json_flow: str):
|
|||
flow = FlowCreate(name="Test Flow", flow=json.loads(json_flow))
|
||||
|
||||
def create_flow():
|
||||
response = client.post("/flows/", json=flow.dict())
|
||||
response = client.post("api/v1/flows/", json=flow.dict())
|
||||
assert response.status_code == 200
|
||||
|
||||
threads = []
|
||||
|
|
@ -197,3 +256,65 @@ def test_stress_create_flow(client: TestClient, json_flow: str):
|
|||
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
|
||||
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())
|
||||
updated_flow_style = FlowStyleUpdate(color="blue")
|
||||
response = client.patch(
|
||||
f"api/v1/flow_styles/{created_flow_style.id}", json=updated_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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue