🔧 fix(test_endpoints.py): fix typo in function name 'created_api_key' to 'create_api_key' for consistency
✨ feat(test_endpoints.py): add test case for handling invalid API key in 'test_process_flow_invalid_api_key' function ✨ feat(test_endpoints.py): add test case for handling invalid flow ID in 'test_process_flow_invalid_id' function ✨ feat(test_endpoints.py): add test case for handling flow not found in 'test_process_flow_invalid_id' function ✨ feat(test_endpoints.py): add test case for testing process flow without autologin in 'test_process_flow_without_autologin' function
This commit is contained in:
parent
a700ab1f71
commit
ea4cd294c1
1 changed files with 54 additions and 0 deletions
|
|
@ -1,3 +1,4 @@
|
|||
import uuid
|
||||
from langflow.services.auth.utils import get_password_hash
|
||||
from langflow.services.database.models.api_key.api_key import ApiKey
|
||||
from langflow.services.utils import get_settings_manager
|
||||
|
|
@ -102,6 +103,59 @@ def created_api_key(session, active_user):
|
|||
return api_key
|
||||
|
||||
|
||||
def test_process_flow_invalid_api_key(client, flow, monkeypatch):
|
||||
# Mock de process_graph_cached
|
||||
def mock_process_graph_cached(*args, **kwargs):
|
||||
return {}, "session_id_mock"
|
||||
|
||||
settings_manager = get_settings_manager()
|
||||
settings_manager.auth_settings.AUTO_LOGIN = False
|
||||
from langflow.api.v1 import endpoints
|
||||
|
||||
monkeypatch.setattr(endpoints, "process_graph_cached", mock_process_graph_cached)
|
||||
|
||||
headers = {"api-key": "invalid_api_key"}
|
||||
|
||||
post_data = {
|
||||
"inputs": {"key": "value"},
|
||||
"tweaks": None,
|
||||
"clear_cache": False,
|
||||
"session_id": None,
|
||||
}
|
||||
|
||||
response = 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):
|
||||
def mock_process_graph_cached(*args, **kwargs):
|
||||
return {}, "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 = {"api-key": api_key}
|
||||
|
||||
post_data = {
|
||||
"inputs": {"key": "value"},
|
||||
"tweaks": None,
|
||||
"clear_cache": False,
|
||||
"session_id": None,
|
||||
}
|
||||
|
||||
invalid_id = uuid.uuid4()
|
||||
response = 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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue