Samir Boulahtit
4adc35a674
Update all test files to import from canonical module locations: - Integration tests: orders, inventory, messages, invoices - Unit tests: services and models - Fixtures: customer, vendor, message fixtures Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
390 lines
13 KiB
Python
390 lines
13 KiB
Python
# tests/integration/api/v1/admin/test_messages.py
|
|
"""
|
|
Integration tests for admin messaging endpoints.
|
|
|
|
Tests the /api/v1/admin/messages/* endpoints.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from app.modules.messaging.models import ConversationType, ParticipantType
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.admin
|
|
class TestAdminMessagesListAPI:
|
|
"""Tests for admin message list endpoints."""
|
|
|
|
def test_list_conversations_empty(self, client, admin_headers):
|
|
"""Test listing conversations when none exist."""
|
|
response = client.get("/api/v1/admin/messages", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "conversations" in data
|
|
assert "total" in data
|
|
assert "total_unread" in data
|
|
assert data["conversations"] == []
|
|
assert data["total"] == 0
|
|
|
|
def test_list_conversations_requires_auth(self, client):
|
|
"""Test that listing requires authentication."""
|
|
response = client.get("/api/v1/admin/messages")
|
|
assert response.status_code == 401
|
|
|
|
def test_list_conversations_requires_admin(self, client, auth_headers):
|
|
"""Test that listing requires admin role."""
|
|
response = client.get("/api/v1/admin/messages", headers=auth_headers)
|
|
assert response.status_code == 403
|
|
|
|
def test_list_conversations_with_data(
|
|
self, client, admin_headers, test_conversation_admin_vendor
|
|
):
|
|
"""Test listing conversations with existing data."""
|
|
response = client.get("/api/v1/admin/messages", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] >= 1
|
|
assert len(data["conversations"]) >= 1
|
|
|
|
# Check conversation structure
|
|
conv = data["conversations"][0]
|
|
assert "id" in conv
|
|
assert "conversation_type" in conv
|
|
assert "subject" in conv
|
|
assert "is_closed" in conv
|
|
|
|
def test_list_conversations_filter_by_type(
|
|
self, client, admin_headers, test_conversation_admin_vendor
|
|
):
|
|
"""Test filtering conversations by type."""
|
|
response = client.get(
|
|
"/api/v1/admin/messages",
|
|
params={"conversation_type": "admin_vendor"},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
for conv in data["conversations"]:
|
|
assert conv["conversation_type"] == "admin_vendor"
|
|
|
|
def test_list_conversations_filter_closed(
|
|
self, client, admin_headers, closed_conversation
|
|
):
|
|
"""Test filtering closed conversations."""
|
|
response = client.get(
|
|
"/api/v1/admin/messages",
|
|
params={"is_closed": True},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
for conv in data["conversations"]:
|
|
assert conv["is_closed"] is True
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.admin
|
|
class TestAdminMessagesUnreadCountAPI:
|
|
"""Tests for unread count endpoint."""
|
|
|
|
def test_get_unread_count(self, client, admin_headers):
|
|
"""Test getting unread count."""
|
|
response = client.get("/api/v1/admin/messages/unread-count", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "total_unread" in data
|
|
assert isinstance(data["total_unread"], int)
|
|
|
|
def test_get_unread_count_with_unread(
|
|
self, client, admin_headers, test_message
|
|
):
|
|
"""Test unread count with unread messages."""
|
|
response = client.get("/api/v1/admin/messages/unread-count", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
# The test_message is sent by admin, so no unread count for admin
|
|
assert data["total_unread"] >= 0
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.admin
|
|
class TestAdminMessagesRecipientsAPI:
|
|
"""Tests for recipients endpoint."""
|
|
|
|
def test_get_vendor_recipients(
|
|
self, client, admin_headers, test_vendor_with_vendor_user
|
|
):
|
|
"""Test getting vendor recipients."""
|
|
response = client.get(
|
|
"/api/v1/admin/messages/recipients",
|
|
params={"recipient_type": "vendor"},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "recipients" in data
|
|
assert "total" in data
|
|
|
|
def test_get_customer_recipients(self, client, admin_headers, test_customer):
|
|
"""Test getting customer recipients."""
|
|
response = client.get(
|
|
"/api/v1/admin/messages/recipients",
|
|
params={"recipient_type": "customer"},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "recipients" in data
|
|
assert "total" in data
|
|
|
|
def test_get_recipients_requires_type(self, client, admin_headers):
|
|
"""Test that recipient_type is required."""
|
|
response = client.get("/api/v1/admin/messages/recipients", headers=admin_headers)
|
|
assert response.status_code == 422 # Validation error
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.admin
|
|
class TestAdminMessagesCreateAPI:
|
|
"""Tests for conversation creation."""
|
|
|
|
def test_create_conversation_admin_vendor(
|
|
self, client, admin_headers, test_vendor_user, test_vendor_with_vendor_user
|
|
):
|
|
"""Test creating admin-vendor conversation."""
|
|
response = client.post(
|
|
"/api/v1/admin/messages",
|
|
json={
|
|
"conversation_type": "admin_vendor",
|
|
"subject": "Test Conversation",
|
|
"recipient_type": "vendor",
|
|
"recipient_id": test_vendor_user.id,
|
|
"vendor_id": test_vendor_with_vendor_user.id,
|
|
"initial_message": "Hello vendor!",
|
|
},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["subject"] == "Test Conversation"
|
|
assert data["conversation_type"] == "admin_vendor"
|
|
assert len(data["messages"]) == 1
|
|
assert data["messages"][0]["content"] == "Hello vendor!"
|
|
|
|
def test_create_conversation_admin_customer(
|
|
self, client, admin_headers, test_customer, test_vendor
|
|
):
|
|
"""Test creating admin-customer conversation."""
|
|
response = client.post(
|
|
"/api/v1/admin/messages",
|
|
json={
|
|
"conversation_type": "admin_customer",
|
|
"subject": "Customer Support",
|
|
"recipient_type": "customer",
|
|
"recipient_id": test_customer.id,
|
|
"vendor_id": test_vendor.id,
|
|
"initial_message": "How can I help you?",
|
|
},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["conversation_type"] == "admin_customer"
|
|
|
|
def test_create_conversation_wrong_recipient_type(
|
|
self, client, admin_headers, test_vendor_user, test_vendor
|
|
):
|
|
"""Test error when recipient type doesn't match conversation type."""
|
|
response = client.post(
|
|
"/api/v1/admin/messages",
|
|
json={
|
|
"conversation_type": "admin_vendor",
|
|
"subject": "Test",
|
|
"recipient_type": "customer", # Wrong type
|
|
"recipient_id": 1,
|
|
"vendor_id": test_vendor.id,
|
|
},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
def test_create_conversation_invalid_type(
|
|
self, client, admin_headers, test_vendor_user, test_vendor
|
|
):
|
|
"""Test error when admin tries to create vendor_customer conversation."""
|
|
response = client.post(
|
|
"/api/v1/admin/messages",
|
|
json={
|
|
"conversation_type": "vendor_customer",
|
|
"subject": "Test",
|
|
"recipient_type": "customer",
|
|
"recipient_id": 1,
|
|
"vendor_id": test_vendor.id,
|
|
},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.admin
|
|
class TestAdminMessagesDetailAPI:
|
|
"""Tests for conversation detail."""
|
|
|
|
def test_get_conversation_detail(
|
|
self, client, admin_headers, test_conversation_admin_vendor
|
|
):
|
|
"""Test getting conversation detail."""
|
|
response = client.get(
|
|
f"/api/v1/admin/messages/{test_conversation_admin_vendor.id}",
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == test_conversation_admin_vendor.id
|
|
assert "participants" in data
|
|
assert "messages" in data
|
|
|
|
def test_get_conversation_not_found(self, client, admin_headers):
|
|
"""Test getting nonexistent conversation."""
|
|
response = client.get("/api/v1/admin/messages/99999", headers=admin_headers)
|
|
assert response.status_code == 404
|
|
|
|
def test_get_conversation_marks_read(
|
|
self, client, admin_headers, test_conversation_admin_vendor
|
|
):
|
|
"""Test that getting detail marks as read by default."""
|
|
response = client.get(
|
|
f"/api/v1/admin/messages/{test_conversation_admin_vendor.id}",
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["unread_count"] == 0
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.admin
|
|
class TestAdminMessagesSendAPI:
|
|
"""Tests for sending messages."""
|
|
|
|
def test_send_message(self, client, admin_headers, test_conversation_admin_vendor):
|
|
"""Test sending a message."""
|
|
response = client.post(
|
|
f"/api/v1/admin/messages/{test_conversation_admin_vendor.id}/messages",
|
|
data={"content": "Test message content"},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["content"] == "Test message content"
|
|
assert data["sender_type"] == "admin"
|
|
|
|
def test_send_message_to_closed(self, client, admin_headers, closed_conversation):
|
|
"""Test cannot send to closed conversation."""
|
|
response = client.post(
|
|
f"/api/v1/admin/messages/{closed_conversation.id}/messages",
|
|
data={"content": "Test message"},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
def test_send_message_not_found(self, client, admin_headers):
|
|
"""Test sending to nonexistent conversation."""
|
|
response = client.post(
|
|
"/api/v1/admin/messages/99999/messages",
|
|
data={"content": "Test message"},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.admin
|
|
class TestAdminMessagesActionsAPI:
|
|
"""Tests for conversation actions."""
|
|
|
|
def test_close_conversation(
|
|
self, client, admin_headers, test_conversation_admin_vendor
|
|
):
|
|
"""Test closing a conversation."""
|
|
response = client.post(
|
|
f"/api/v1/admin/messages/{test_conversation_admin_vendor.id}/close",
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert "closed" in data["message"].lower()
|
|
|
|
def test_close_conversation_not_found(self, client, admin_headers):
|
|
"""Test closing nonexistent conversation."""
|
|
response = client.post(
|
|
"/api/v1/admin/messages/99999/close",
|
|
headers=admin_headers,
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
def test_reopen_conversation(self, client, admin_headers, closed_conversation):
|
|
"""Test reopening a closed conversation."""
|
|
response = client.post(
|
|
f"/api/v1/admin/messages/{closed_conversation.id}/reopen",
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert "reopen" in data["message"].lower()
|
|
|
|
def test_mark_read(self, client, admin_headers, test_conversation_admin_vendor):
|
|
"""Test marking conversation as read."""
|
|
response = client.put(
|
|
f"/api/v1/admin/messages/{test_conversation_admin_vendor.id}/read",
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert data["unread_count"] == 0
|
|
|
|
def test_update_preferences(
|
|
self, client, admin_headers, test_conversation_admin_vendor
|
|
):
|
|
"""Test updating notification preferences."""
|
|
response = client.put(
|
|
f"/api/v1/admin/messages/{test_conversation_admin_vendor.id}/preferences",
|
|
json={"email_notifications": False, "muted": True},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|