test: add backend flow test (#995)

<!-- .github/pull_request_template.md -->

## Description
Simple Cognee endpoint testing

## DCO Affirmation
I affirm that all code in every commit of this pull request conforms to
the terms of the Topoteretes Developer Certificate of Origin.

---------

Co-authored-by: Boris <boris@topoteretes.com>
This commit is contained in:
Igor Ilic 2025-06-18 21:14:07 +02:00 committed by GitHub
parent e7644f4b3a
commit 72f97bbccc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194

View file

@ -4,6 +4,7 @@ import time
import os
import signal
import requests
from pathlib import Path
import sys
@ -27,7 +28,7 @@ class TestCogneeServerStart(unittest.TestCase):
preexec_fn=os.setsid,
)
# Give the server some time to start
time.sleep(30)
time.sleep(35)
# Check if server started with errors
if cls.server_process.poll() is not None:
@ -54,6 +55,64 @@ class TestCogneeServerStart(unittest.TestCase):
self.assertIn("message", root_response.json())
self.assertEqual(root_response.json()["message"], "Hello, World, I am alive!")
# Login request
url = "http://127.0.0.1:8000/api/v1/auth/login"
form_data = {
"username": "default_user@example.com",
"password": "default_password",
}
login_response = requests.post(url, data=form_data, timeout=10)
login_response.raise_for_status() # raises on HTTP 4xx/5xx
# Define Bearer token to use for authorization
auth_var = (
"Bearer " + login_response.json()["access_token"]
) # e.g. "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6..."
# Add request
url = "http://127.0.0.1:8000/api/v1/add"
file_path = Path(os.path.join(Path(__file__).parent, "test_data/example.png"))
headers = {"Authorization": auth_var}
form_data = {"datasetName": "test"}
file = {
"data": (
file_path.name,
open(file_path, "rb"),
)
}
add_response = requests.post(url, headers=headers, data=form_data, files=file, timeout=15)
add_response.raise_for_status() # raise if HTTP 4xx/5xx
# Cognify request
url = "http://127.0.0.1:8000/api/v1/cognify"
headers = {
"Authorization": auth_var,
"Content-Type": "application/json",
}
payload = {"datasets": ["test"]}
cognify_response = requests.post(url, headers=headers, json=payload, timeout=15)
cognify_response.raise_for_status() # raises on HTTP 4xx/5xx
# TODO: Add test to verify cognify pipeline is complete before testing search
# Search request
url = "http://127.0.0.1:8000/api/v1/search"
headers = {
"Authorization": auth_var,
"Content-Type": "application/json",
}
payload = {"searchType": "GRAPH_COMPLETION", "query": "What's in the document?"}
search_response = requests.post(url, headers=headers, json=payload, timeout=15)
search_response.raise_for_status() # raises on HTTP 4xx/5xx
if __name__ == "__main__":
unittest.main()