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

test: add unit tests for routes (#4249)

Browse source
This commit is contained in:
Ítalo Johnny 2024-10-28 17:06:05 -03:00 • committed by GitHub
commit fadb20115d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
23 changed files with 623 additions and 272 deletions
Download patch file Download diff file Expand all files Collapse all files

151
src/backend/tests/unit/test_endpoints.py
View file

@ -108,157 +108,6 @@ PROMPT_REQUEST = {
}
# def test_process_flow_invalid_api_key(client, flow, monkeypatch):
# # Mock de process_graph_cached
# from langflow.api.v1 import endpoints
# from langflow.services.database.models.api_key import crud
# settings_service = get_settings_service()
# settings_service.auth_settings.AUTO_LOGIN = False
# async def mock_process_graph_cached(*args, **kwargs):
# return Result(result={}, session_id="session_id_mock")
# def mock_update_total_uses(*args, **kwargs):
# return created_api_key
# monkeypatch.setattr(endpoints, "process_graph_cached", mock_process_graph_cached)
# monkeypatch.setattr(crud, "update_total_uses", mock_update_total_uses)
# headers = {"x-api-key": "invalid_api_key"}
# post_data = {
# "inputs": {"key": "value"},
# "tweaks": None,
# "clear_cache": False,
# "session_id": None,
# }
# response = await client.post(f"api/v1/process/{flow.id}", headers=headers, json=post_data)
# assert response.status_code == 403
# assert response.json() == {"detail": "Invalid or missing API key"}
# def test_process_flow_invalid_id(client, monkeypatch, created_api_key):
# async def mock_process_graph_cached(*args, **kwargs):
# return Result(result={}, session_id="session_id_mock")
# from langflow.api.v1 import endpoints
# monkeypatch.setattr(endpoints, "process_graph_cached", mock_process_graph_cached)
# api_key = created_api_key.api_key
# headers = {"x-api-key": api_key}
# post_data = {
# "inputs": {"key": "value"},
# "tweaks": None,
# "clear_cache": False,
# "session_id": None,
# }
# invalid_id = uuid.uuid4()
# response = await client.post(f"api/v1/process/{invalid_id}", headers=headers, json=post_data)
# assert response.status_code == 404
# assert f"Flow {invalid_id} not found" in response.json()["detail"]
# def test_process_flow_without_autologin(client, flow, monkeypatch, created_api_key):
# # Mock de process_graph_cached
# from langflow.api.v1 import endpoints
# from langflow.services.database.models.api_key import crud
# settings_service = get_settings_service()
# settings_service.auth_settings.AUTO_LOGIN = False
# async def mock_process_graph_cached(*args, **kwargs):
# return Result(result={}, session_id="session_id_mock")
# def mock_process_graph_cached_task(*args, **kwargs):
# return Result(result={}, session_id="session_id_mock")
# # The task function is ran like this:
# # if not self.use_celery:
# # return None, await task_func(*args, **kwargs)
# # if not hasattr(task_func, "apply"):
# # raise ValueError(f"Task function {task_func} does not have an apply method")
# # task = task_func.apply(args=args, kwargs=kwargs)
# # result = task.get()
# # return task.id, result
# # So we need to mock the task function to return a task object
# # and then mock the task object to return a result
# # maybe a named tuple would be better here
# task = namedtuple("task", ["id", "get"])
# mock_process_graph_cached_task.apply = lambda *args, **kwargs: task(
# id="task_id_mock", get=lambda: Result(result={}, session_id="session_id_mock")
# )
# def mock_update_total_uses(*args, **kwargs):
# return created_api_key
# monkeypatch.setattr(endpoints, "process_graph_cached", mock_process_graph_cached)
# monkeypatch.setattr(crud, "update_total_uses", mock_update_total_uses)
# monkeypatch.setattr(endpoints, "process_graph_cached_task", mock_process_graph_cached_task)
# api_key = created_api_key.api_key
# headers = {"x-api-key": api_key}
# # Dummy POST data
# post_data = {
# "inputs": {"input": "value"},
# "tweaks": None,
# "clear_cache": False,
# "session_id": None,
# }
# # Make the request to the FastAPI TestClient
# response = await client.post(f"api/v1/process/{flow.id}", headers=headers, json=post_data)
# # Check the response
# assert response.status_code == 200, response.json()
# assert response.json()["result"] == {}, response.json()
# assert response.json()["session_id"] == "session_id_mock", response.json()
# def test_process_flow_fails_autologin_off(client, flow, monkeypatch):
# # Mock de process_graph_cached
# from langflow.api.v1 import endpoints
# from langflow.services.database.models.api_key import crud
# settings_service = get_settings_service()
# settings_service.auth_settings.AUTO_LOGIN = False
# async def mock_process_graph_cached(*args, **kwargs):
# return Result(result={}, session_id="session_id_mock")
# async def mock_update_total_uses(*args, **kwargs):
# return created_api_key
# monkeypatch.setattr(endpoints, "process_graph_cached", mock_process_graph_cached)
# monkeypatch.setattr(crud, "update_total_uses", mock_update_total_uses)
# headers = {"x-api-key": "api_key"}
# # Dummy POST data
# post_data = {
# "inputs": {"key": "value"},
# "tweaks": None,
# "clear_cache": False,
# "session_id": None,
# }
# # Make the request to the FastAPI TestClient
# response = await client.post(f"api/v1/process/{flow.id}", headers=headers, json=post_data)
# # Check the response
# assert response.status_code == 403, response.json()
# assert response.json() == {"detail": "Invalid or missing API key"}
async def test_get_all(client: AsyncClient, logged_in_headers):
response = await client.get("api/v1/all", headers=logged_in_headers)
assert response.status_code == 200