50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
"""LangflowClient HTTP mapping tests (Langflow mocked via respx, no LFX)."""
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from app.flows.langflow_client import (
|
|
LangflowClient,
|
|
LangflowNotFound,
|
|
LangflowUnavailable,
|
|
)
|
|
|
|
BASE = "https://langflow.test"
|
|
|
|
|
|
def _client():
|
|
return LangflowClient(BASE, api_key="test-key")
|
|
|
|
|
|
@respx.mock
|
|
async def test_export_flow_sends_api_key_and_returns_doc():
|
|
doc = {"id": "abc", "name": "n", "data": {"nodes": [], "edges": []}}
|
|
route = respx.get(f"{BASE}/api/v1/flows/abc").mock(return_value=httpx.Response(200, json=doc))
|
|
result = await _client().export_flow("abc")
|
|
assert result == doc
|
|
assert route.called
|
|
assert route.calls[0].request.headers["x-api-key"] == "test-key"
|
|
|
|
|
|
@respx.mock
|
|
async def test_export_flow_not_found():
|
|
respx.get(f"{BASE}/api/v1/flows/missing").mock(
|
|
return_value=httpx.Response(404, json={"detail": "nope"})
|
|
)
|
|
with pytest.raises(LangflowNotFound):
|
|
await _client().export_flow("missing")
|
|
|
|
|
|
@respx.mock
|
|
async def test_list_flows_unwraps_items():
|
|
respx.get(f"{BASE}/api/v1/flows/").mock(
|
|
return_value=httpx.Response(200, json={"items": [{"id": "1"}]})
|
|
)
|
|
assert await _client().list_flows() == [{"id": "1"}]
|
|
|
|
|
|
async def test_unreachable_raises_unavailable():
|
|
client = LangflowClient("https://langflow.dead", api_key="k")
|
|
with pytest.raises(LangflowUnavailable):
|
|
await client.export_flow("whatever")
|