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

Refactor test_endpoints.py file

Browse source
This commit is contained in:
Gabriel Luiz Freitas Almeida 2023-11-28 14:35:50 -03:00
commit 71473fcec6
1 changed files with 9 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

19
tests/test_endpoints.py
View file

@ -1,18 +1,17 @@
from collections import namedtuple import time
import uuid import uuid
from langflow.processing.process import Result from collections import namedtuple
from langflow.services.auth.utils import get_password_hash
from langflow.services.database.models.api_key.model import ApiKey
from langflow.services.deps import get_settings_service
from langflow.services.database.utils import session_getter
from langflow.services.deps import get_db_service
import pytest import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from langflow.interface.tools.constants import CUSTOM_TOOLS from langflow.interface.tools.constants import CUSTOM_TOOLS
from langflow.processing.process import Result
from langflow.services.auth.utils import get_password_hash
from langflow.services.database.models.api_key.model import ApiKey
from langflow.services.database.utils import session_getter
from langflow.services.deps import get_db_service, get_settings_service
from langflow.template.frontend_node.chains import TimeTravelGuideChainNode from langflow.template.frontend_node.chains import TimeTravelGuideChainNode
import time
def run_post(client, flow_id, headers, post_data): def run_post(client, flow_id, headers, post_data):
response = client.post( response = client.post(
@ -25,7 +24,7 @@ def run_post(client, flow_id, headers, post_data):
# Helper function to poll task status # Helper function to poll task status
def poll_task_status(client, headers, href, max_attempts=20, sleep_time=1): def poll_task_status(client, headers, href, max_attempts=20, sleep_time=3):
for _ in range(max_attempts): for _ in range(max_attempts):
task_status_response = client.get( task_status_response = client.get(
href, href,