80 lines
2.5 KiB
Bash
Executable File
80 lines
2.5 KiB
Bash
Executable File
#!/usr/bin/env python3
|
|
import sys
|
|
import requests
|
|
|
|
BASE_URL = "http://127.0.0.1:8000"
|
|
COL = "test_collection"
|
|
SRC = "unit-test-src"
|
|
|
|
def fail(msg):
|
|
print("✗", msg)
|
|
sys.exit(1)
|
|
|
|
def test_openapi():
|
|
r = requests.get(f"{BASE_URL}/openapi.json")
|
|
if r.status_code != 200:
|
|
fail(f"/openapi.json returned {r.status_code}")
|
|
print("✓ OpenAPI: 200 OK")
|
|
|
|
def test_embed():
|
|
payload = {
|
|
"collection": COL,
|
|
"chunks": [
|
|
{"text": "Das ist ein Testtext für Embed.", "source": SRC}
|
|
]
|
|
}
|
|
r = requests.post(f"{BASE_URL}/embed", json=payload)
|
|
if r.status_code != 200:
|
|
fail(f"/embed returned {r.status_code}: {r.text}")
|
|
data = r.json()
|
|
if data.get("count") != 1:
|
|
fail(f"/embed count != 1: {data}")
|
|
print("✓ Embed: 1 Eintrag gespeichert")
|
|
|
|
def test_search():
|
|
params = {"query": "Testtext", "collection": COL}
|
|
r = requests.get(f"{BASE_URL}/search", params=params)
|
|
if r.status_code != 200:
|
|
fail(f"/search returned {r.status_code}: {r.text}")
|
|
results = r.json()
|
|
if not any("score" in item for item in results):
|
|
fail(f"/search lieferte keine Treffer: {results}")
|
|
print("✓ Search: Treffer gefunden")
|
|
|
|
def test_prompt():
|
|
payload = {"query": "Wie lautet dieser Testtext?", "context_limit": 1, "collection": COL}
|
|
r = requests.post(f"{BASE_URL}/prompt", json=payload)
|
|
if r.status_code != 200:
|
|
fail(f"/prompt returned {r.status_code}: {r.text}")
|
|
data = r.json()
|
|
if "answer" not in data:
|
|
fail(f"/prompt liefert kein 'answer'-Feld: {data}")
|
|
print("✓ Prompt: Antwort erhalten")
|
|
|
|
def test_delete_source():
|
|
params = {"collection": COL, "source": SRC}
|
|
r = requests.delete(f"{BASE_URL}/delete-source", params=params)
|
|
if r.status_code != 200:
|
|
fail(f"/delete-source returned {r.status_code}: {r.text}")
|
|
data = r.json()
|
|
if data.get("count") != 1:
|
|
fail(f"/delete-source count != 1: {data}")
|
|
print("✓ Delete-source: 1 Eintrag gelöscht")
|
|
|
|
def test_delete_collection():
|
|
params = {"collection": COL}
|
|
r = requests.delete(f"{BASE_URL}/delete-collection", params=params)
|
|
if r.status_code != 200:
|
|
fail(f"/delete-collection returned {r.status_code}: {r.text}")
|
|
print("✓ Delete-collection: Collection gelöscht")
|
|
|
|
if __name__ == "__main__":
|
|
print("\nStarte API-Tests...\n")
|
|
test_openapi()
|
|
test_embed()
|
|
test_search()
|
|
test_prompt()
|
|
test_delete_source()
|
|
test_delete_collection()
|
|
print("\n🎉 Alle Tests erfolgreich durchlaufen!")
|