211 lines
7.1 KiB
Python
211 lines
7.1 KiB
Python
import pytest
|
|
|
|
from src import models # Import your SQLAlchemy models
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_message(client, db_session, sample_data):
|
|
test_app, test_user = sample_data
|
|
# Create a test session
|
|
test_session = models.Session(user_id=test_user.public_id)
|
|
db_session.add(test_session)
|
|
await db_session.commit()
|
|
|
|
response = client.post(
|
|
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{test_session.public_id}/messages",
|
|
json={
|
|
"content": "Test message",
|
|
"is_user": True,
|
|
"metadata": {"message_key": "message_value"},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["content"] == "Test message"
|
|
assert data["is_user"] is True
|
|
assert data["metadata"] == {"message_key": "message_value"}
|
|
assert "id" in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_messages(client, db_session, sample_data):
|
|
test_app, test_user = sample_data
|
|
# Create a test session and message
|
|
test_session = models.Session(user_id=test_user.public_id)
|
|
db_session.add(test_session)
|
|
await db_session.commit()
|
|
test_message = models.Message(
|
|
session_id=test_session.public_id, content="Test message", is_user=True
|
|
)
|
|
db_session.add(test_message)
|
|
await db_session.commit()
|
|
|
|
response = client.post(
|
|
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{test_session.public_id}/messages/list",
|
|
json={},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "items" in data
|
|
assert len(data["items"]) > 0
|
|
assert data["items"][0]["content"] == "Test message"
|
|
assert data["items"][0]["is_user"] is True
|
|
assert data["items"][0]["metadata"] == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_filtered_messages(client, db_session, sample_data):
|
|
test_app, test_user = sample_data
|
|
# Create a test session and message
|
|
test_session = models.Session(user_id=test_user.public_id)
|
|
db_session.add(test_session)
|
|
await db_session.commit()
|
|
test_message = models.Message(
|
|
session_id=test_session.public_id,
|
|
content="Test message",
|
|
is_user=True,
|
|
h_metadata={"key": "value"},
|
|
)
|
|
test_message2 = models.Message(
|
|
session_id=test_session.public_id,
|
|
content="Test message",
|
|
is_user=True,
|
|
h_metadata={"key": "value2"},
|
|
)
|
|
db_session.add(test_message)
|
|
db_session.add(test_message2)
|
|
await db_session.commit()
|
|
|
|
response = client.post(
|
|
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{test_session.public_id}/messages/list",
|
|
json={"filter": {"key": "value2"}},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
print(data)
|
|
assert "items" in data
|
|
assert len(data["items"]) > 0
|
|
assert data["items"][0]["content"] == "Test message"
|
|
assert data["items"][0]["is_user"] is True
|
|
assert data["items"][0]["metadata"] == {"key": "value2"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_message(client, db_session, sample_data):
|
|
test_app, test_user = sample_data
|
|
# Create a test session and message
|
|
test_session = models.Session(user_id=test_user.public_id)
|
|
db_session.add(test_session)
|
|
await db_session.commit()
|
|
test_message = models.Message(
|
|
session_id=test_session.public_id, content="Test message", is_user=True
|
|
)
|
|
db_session.add(test_message)
|
|
await db_session.commit()
|
|
|
|
response = client.put(
|
|
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{test_session.public_id}/messages/{test_message.public_id}",
|
|
json={"metadata": {"new_key": "new_value"}},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["metadata"] == {"new_key": "new_value"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_message_empty_metadata(client, db_session, sample_data):
|
|
test_app, test_user = sample_data
|
|
# Create a test session and message
|
|
test_session = models.Session(user_id=test_user.public_id)
|
|
db_session.add(test_session)
|
|
await db_session.commit()
|
|
test_message = models.Message(
|
|
session_id=test_session.public_id, content="Test message", is_user=True
|
|
)
|
|
db_session.add(test_message)
|
|
await db_session.commit()
|
|
|
|
response = client.put(
|
|
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{test_session.public_id}/messages/{test_message.public_id}",
|
|
json={"metadata": None},
|
|
)
|
|
assert response.status_code == 422
|
|
data = response.json()
|
|
print(data)
|
|
# assert data["detail"] == "Message metadata cannot be empty"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_batch_messages(client, db_session, sample_data):
|
|
test_app, test_user = sample_data
|
|
# Create a test session
|
|
test_session = models.Session(user_id=test_user.public_id)
|
|
db_session.add(test_session)
|
|
await db_session.commit()
|
|
|
|
# Create batch of test messages
|
|
test_messages = {
|
|
"messages": [
|
|
{
|
|
"content": f"Test message {i}",
|
|
"is_user": i % 2 == 0, # Alternating user/non-user messages
|
|
"metadata": {"batch_index": i}
|
|
} for i in range(3)
|
|
]
|
|
}
|
|
|
|
response = client.post(
|
|
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{test_session.public_id}/messages/batch",
|
|
json=test_messages,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Verify the response contains all messages
|
|
assert len(data) == 3
|
|
|
|
# Verify messages are in the correct order and have correct content
|
|
for i, message in enumerate(data):
|
|
assert message["content"] == f"Test message {i}"
|
|
assert message["is_user"] == (i % 2 == 0)
|
|
assert message["metadata"] == {"batch_index": i}
|
|
assert "id" in message
|
|
assert message["session_id"] == test_session.public_id
|
|
|
|
# Verify messages were actually saved to the database
|
|
response = client.post(
|
|
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{test_session.public_id}/messages/list",
|
|
json={},
|
|
)
|
|
assert response.status_code == 200
|
|
saved_messages = response.json()["items"]
|
|
assert len(saved_messages) == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_batch_messages_limit(client, db_session, sample_data):
|
|
test_app, test_user = sample_data
|
|
test_session = models.Session(user_id=test_user.public_id)
|
|
db_session.add(test_session)
|
|
await db_session.commit()
|
|
|
|
# Create batch with more than 100 messages
|
|
test_messages = {
|
|
"messages": [
|
|
{
|
|
"content": f"Test message {i}",
|
|
"is_user": i % 2 == 0,
|
|
"metadata": {"batch_index": i}
|
|
} for i in range(101) # 101 messages
|
|
]
|
|
}
|
|
|
|
response = client.post(
|
|
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{test_session.public_id}/messages/batch",
|
|
json=test_messages,
|
|
)
|
|
|
|
assert response.status_code == 422 # Validation error
|
|
data = response.json()
|
|
assert "messages" in data["detail"][0]["loc"] # Error should mention messages field
|