File size: 5,803 Bytes
365de9c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 |
from app.backend.models.users import User # noqa: F401
from app.backend.models.chats import Chat # noqa: F401
from app.backend.models.messages import Message # noqa: F401
import os
from uuid import uuid4
import httpx
import pytest
from app.backend.models.messages import get_messages_by_chat_id
from app.settings import BASE_DIR
BASE_URL = os.environ.get('HF1_URL')
# --- Fixtures for Setup ---
@pytest.fixture
def artificial_user():
"""Fixture to create and log in an artificial user, returning user data."""
email = f"Test{uuid4()}@test.com"
password = "Goida123!"
payload = {"email": email, "password": password}
with httpx.Client(verify=False) as client:
# Create user
response = client.post(url=BASE_URL + "/new_user", json=payload, timeout=30.0)
if response.status_code != 200:
raise RuntimeError(f"Failed to create artificial user: {response.status_code} - {response.text}")
id = response.json().get("id", -1)
# Log in to get access token
headers = {"Content-Type": "application/json", "Accept": "application/json"}
response = client.post(url=BASE_URL + "/login", json=payload, headers=headers, timeout=30.0)
if response.status_code != 200:
raise RuntimeError(f"Login failed: {response.status_code} - {response.text}")
# Extract access token from set-cookie header
set_cookie_header = response.headers.get("set-cookie")
access_token = None
if set_cookie_header and "access_token=" in set_cookie_header:
access_token = set_cookie_header.split("access_token=")[1].split(";")[0]
if not access_token:
raise RuntimeError("No access token received from login")
return {"id": id, "email": email, "password": password, "access_token": access_token}
@pytest.fixture
def chat_data(artificial_user):
"""Fixture to create a chat for the artificial user, returning chat data."""
cookie = {"access_token": artificial_user["access_token"]}
with httpx.Client(verify=False, cookies=cookie) as client:
# Create chat
response = client.post(url=BASE_URL + "/new_chat", timeout=30.0)
if response.status_code != 303:
raise RuntimeError(f"Error while trying to create chat: {response.status_code} - {response.text}")
redirect_to = response.headers.get("location")
if not redirect_to or "login" in redirect_to:
raise RuntimeError(f"Authentication failed, redirected to: {redirect_to}")
# Follow redirect
response = client.get(url=BASE_URL + redirect_to)
if response.status_code != 200:
raise RuntimeError(f"Error while accessing chat: {response.status_code} - {response.text}")
try:
chat_id = int(redirect_to.split("/")[-1].split("id=")[-1])
except ValueError as e:
raise RuntimeError(f"Failed to parse chat_id from URL: {redirect_to} - {e}")
return {"chat_id": chat_id, "cookie": cookie}
# --- Test Functions ---
def test_create_artificial_user(artificial_user):
"""Test that an artificial user can be created and logged in."""
assert artificial_user["email"] is not None
assert artificial_user["password"] == "Goida123!"
assert artificial_user["access_token"] is not None
def test_validate_chat_creation(chat_data):
"""Test that a chat can be created successfully."""
assert chat_data["chat_id"] > 0
assert "access_token" in chat_data["cookie"]
def test_validate_message_sending(chat_data):
"""Test that a message can be sent to the chat."""
with httpx.Client(verify=False, cookies=chat_data["cookie"]) as client:
payload = {"prompt": "How is your day?", "chat_id": chat_data["chat_id"]}
response = client.post(
url=BASE_URL + "/message_with_docs",
data=payload,
timeout=180,
)
assert response.status_code == 200, f"Failed to send message: {response.status_code} - {response.text}"
def test_validate_docs_uploading(chat_data):
"""Test that a document can be uploaded with a message."""
file_path = os.path.join(BASE_DIR, "app", "tests", "integration", "testfile.txt")
# Create a test file if it doesn't exist
if not os.path.exists(file_path):
with open(file_path, "w") as f:
f.write("This is a test file for validation.")
with httpx.Client(verify=False, cookies=chat_data["cookie"]) as client:
with open(file_path, "rb") as f:
form_fields = {
"prompt": "How is your day?",
"chat_id": str(chat_data["chat_id"]),
}
files = [("files", ("testfile.txt", f, "text/plain"))]
response = client.post(
url=BASE_URL + "/message_with_docs",
data=form_fields,
files=files,
timeout=180,
)
assert response.status_code == 200, f"Failed to upload docs: {response.status_code} - {response.text}"
def test_validate_message_registration(chat_data):
"""Test that a sent message is registered in the chat."""
with httpx.Client(verify=False, cookies=chat_data["cookie"]) as client:
initial = get_messages_by_chat_id(chat_data["chat_id"]).count()
payload = {"prompt": "How is your day?", "chat_id": chat_data["chat_id"]}
response = client.post(
url=BASE_URL + "/message_with_docs",
data=payload,
timeout=180,
)
assert response.status_code == 200, f"Failed to send message: {response.status_code} - {response.text}"
after_sending = get_messages_by_chat_id(chat_data["chat_id"]).count()
assert after_sending - initial == 1, "Message was not registered"
|