677 lines
20 KiB
Python
677 lines
20 KiB
Python
"""
|
|
Tickets Integration Tests - ServiceManagerWeb
|
|
|
|
Tests completos del CRUD de tickets y funcionalidad relacionada.
|
|
"""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
import uuid
|
|
|
|
from app.models.user import User
|
|
from app.models.tenant import Tenant
|
|
from app.models.ticket import Ticket, TicketStatus, TicketPriority
|
|
from app.models.system import System
|
|
from app.models.category import Category
|
|
from app.core.file_handler import file_handler
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.db
|
|
class TestTicketCreation:
|
|
"""Tests de creación de tickets."""
|
|
|
|
async def test_create_ticket_success(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant,
|
|
test_category: Category,
|
|
auth_headers_client: dict
|
|
):
|
|
"""Test crear ticket con datos válidos."""
|
|
|
|
response = await client.post(
|
|
"/v1/tickets/",
|
|
headers={
|
|
**auth_headers_client,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
},
|
|
json={
|
|
"title": "Test ticket",
|
|
"description": "This is a test ticket description",
|
|
"priority": "MEDIUM",
|
|
"category_id": str(test_category.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
|
|
assert data["title"] == "Test ticket"
|
|
assert data["description"] == "This is a test ticket description"
|
|
assert data["priority"] == "MEDIUM"
|
|
assert data["status"] == "NEW"
|
|
assert data["category_id"] == str(test_category.id)
|
|
|
|
async def test_create_ticket_with_all_fields(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant,
|
|
test_category: Category,
|
|
test_system: System,
|
|
auth_headers_admin: dict
|
|
):
|
|
"""Test crear ticket con todos los campos opcionales."""
|
|
|
|
response = await client.post(
|
|
"/v1/tickets/",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
},
|
|
json={
|
|
"title": "Complete ticket",
|
|
"description": "Full ticket with all fields",
|
|
"priority": "HIGH",
|
|
"category_id": str(test_category.id),
|
|
"system_id": str(test_system.id),
|
|
"contact_email": "contact@test.com",
|
|
"contact_phone": "+1234567890"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
|
|
assert data["priority"] == "HIGH"
|
|
assert data["system_id"] == str(test_system.id)
|
|
assert data["contact_email"] == "contact@test.com"
|
|
|
|
async def test_create_ticket_missing_required_fields(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant,
|
|
auth_headers_client: dict
|
|
):
|
|
"""Test crear ticket sin campos requeridos."""
|
|
|
|
response = await client.post(
|
|
"/v1/tickets/",
|
|
headers={
|
|
**auth_headers_client,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
},
|
|
json={
|
|
"description": "Missing title"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 422 # Validation error
|
|
|
|
async def test_create_ticket_invalid_priority(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant,
|
|
test_category: Category,
|
|
auth_headers_client: dict
|
|
):
|
|
"""Test crear ticket con prioridad inválida."""
|
|
|
|
response = await client.post(
|
|
"/v1/tickets/",
|
|
headers={
|
|
**auth_headers_client,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
},
|
|
json={
|
|
"title": "Test ticket",
|
|
"description": "Description",
|
|
"priority": "SUPER_URGENT", # Inválido
|
|
"category_id": str(test_category.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.db
|
|
class TestTicketRetrieval:
|
|
"""Tests de consulta de tickets."""
|
|
|
|
async def test_list_tickets_empty(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant,
|
|
auth_headers_admin: dict
|
|
):
|
|
"""Test listar tickets cuando no hay ninguno."""
|
|
|
|
response = await client.get(
|
|
"/v1/tickets/",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
tickets = response.json()
|
|
assert isinstance(tickets, list)
|
|
|
|
async def test_list_tickets_with_data(
|
|
self,
|
|
client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
test_tenant: Tenant,
|
|
test_admin_user: User,
|
|
test_category: Category,
|
|
auth_headers_admin: dict
|
|
):
|
|
"""Test listar tickets cuando existen."""
|
|
|
|
# Crear algunos tickets
|
|
for i in range(3):
|
|
ticket = Ticket(
|
|
tenant_id=test_tenant.id,
|
|
title=f"Test ticket {i+1}",
|
|
description=f"Description {i+1}",
|
|
status=TicketStatus.NEW,
|
|
priority=TicketPriority.MEDIUM,
|
|
created_by=test_admin_user.id,
|
|
category_id=test_category.id
|
|
)
|
|
db_session.add(ticket)
|
|
await db_session.commit()
|
|
|
|
response = await client.get(
|
|
"/v1/tickets/",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
tickets = response.json()
|
|
assert len(tickets) == 3
|
|
|
|
async def test_get_ticket_by_id(
|
|
self,
|
|
client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
test_tenant: Tenant,
|
|
test_admin_user: User,
|
|
test_category: Category,
|
|
auth_headers_admin: dict
|
|
):
|
|
"""Test obtener ticket específico por ID."""
|
|
|
|
ticket = Ticket(
|
|
tenant_id=test_tenant.id,
|
|
title="Specific ticket",
|
|
description="Get this ticket",
|
|
status=TicketStatus.NEW,
|
|
priority=TicketPriority.HIGH,
|
|
created_by=test_admin_user.id,
|
|
category_id=test_category.id
|
|
)
|
|
db_session.add(ticket)
|
|
await db_session.commit()
|
|
await db_session.refresh(ticket)
|
|
|
|
response = await client.get(
|
|
f"/v1/tickets/{ticket.id}",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == str(ticket.id)
|
|
assert data["title"] == "Specific ticket"
|
|
|
|
async def test_get_nonexistent_ticket(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant,
|
|
auth_headers_admin: dict
|
|
):
|
|
"""Test obtener ticket inexistente."""
|
|
|
|
fake_id = str(uuid.uuid4())
|
|
|
|
response = await client.get(
|
|
f"/v1/tickets/{fake_id}",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.db
|
|
class TestTicketUpdate:
|
|
"""Tests de actualización de tickets."""
|
|
|
|
async def test_update_ticket_status(
|
|
self,
|
|
client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
test_tenant: Tenant,
|
|
test_admin_user: User,
|
|
test_category: Category,
|
|
auth_headers_admin: dict
|
|
):
|
|
"""Test actualizar status de ticket."""
|
|
|
|
ticket = Ticket(
|
|
tenant_id=test_tenant.id,
|
|
title="Ticket to update",
|
|
description="Description",
|
|
status=TicketStatus.NEW,
|
|
priority=TicketPriority.MEDIUM,
|
|
created_by=test_admin_user.id,
|
|
category_id=test_category.id
|
|
)
|
|
db_session.add(ticket)
|
|
await db_session.commit()
|
|
await db_session.refresh(ticket)
|
|
|
|
response = await client.patch(
|
|
f"/v1/tickets/{ticket.id}",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
},
|
|
json={
|
|
"status": "IN_PROGRESS"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "IN_PROGRESS"
|
|
|
|
async def test_update_ticket_priority(
|
|
self,
|
|
client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
test_tenant: Tenant,
|
|
test_admin_user: User,
|
|
test_category: Category,
|
|
auth_headers_admin: dict
|
|
):
|
|
"""Test actualizar prioridad de ticket."""
|
|
|
|
ticket = Ticket(
|
|
tenant_id=test_tenant.id,
|
|
title="Ticket priority test",
|
|
description="Description",
|
|
status=TicketStatus.NEW,
|
|
priority=TicketPriority.LOW,
|
|
created_by=test_admin_user.id,
|
|
category_id=test_category.id
|
|
)
|
|
db_session.add(ticket)
|
|
await db_session.commit()
|
|
await db_session.refresh(ticket)
|
|
|
|
response = await client.patch(
|
|
f"/v1/tickets/{ticket.id}",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
},
|
|
json={
|
|
"priority": "URGENT"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["priority"] == "URGENT"
|
|
|
|
async def test_update_ticket_assignment(
|
|
self,
|
|
client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
test_tenant: Tenant,
|
|
test_admin_user: User,
|
|
test_agent_user: User,
|
|
test_category: Category,
|
|
auth_headers_admin: dict
|
|
):
|
|
"""Test asignar ticket a un agente."""
|
|
|
|
ticket = Ticket(
|
|
tenant_id=test_tenant.id,
|
|
title="Ticket to assign",
|
|
description="Description",
|
|
status=TicketStatus.NEW,
|
|
priority=TicketPriority.MEDIUM,
|
|
created_by=test_admin_user.id,
|
|
category_id=test_category.id
|
|
)
|
|
db_session.add(ticket)
|
|
await db_session.commit()
|
|
await db_session.refresh(ticket)
|
|
|
|
response = await client.patch(
|
|
f"/v1/tickets/{ticket.id}",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
},
|
|
json={
|
|
"assigned_to": str(test_agent_user.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["assigned_to"] == str(test_agent_user.id)
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.db
|
|
class TestTicketFilters:
|
|
"""Tests de filtros de tickets."""
|
|
|
|
async def test_filter_by_status(
|
|
self,
|
|
client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
test_tenant: Tenant,
|
|
test_admin_user: User,
|
|
test_category: Category,
|
|
auth_headers_admin: dict
|
|
):
|
|
"""Test filtrar tickets por status."""
|
|
|
|
# Crear tickets con diferentes status
|
|
ticket_new = Ticket(
|
|
tenant_id=test_tenant.id,
|
|
title="New ticket",
|
|
description="Description",
|
|
status=TicketStatus.NEW,
|
|
priority=TicketPriority.MEDIUM,
|
|
created_by=test_admin_user.id,
|
|
category_id=test_category.id
|
|
)
|
|
ticket_progress = Ticket(
|
|
tenant_id=test_tenant.id,
|
|
title="In progress ticket",
|
|
description="Description",
|
|
status=TicketStatus.IN_PROGRESS,
|
|
priority=TicketPriority.MEDIUM,
|
|
created_by=test_admin_user.id,
|
|
category_id=test_category.id
|
|
)
|
|
db_session.add_all([ticket_new, ticket_progress])
|
|
await db_session.commit()
|
|
|
|
# Filtrar por status NEW
|
|
response = await client.get(
|
|
"/v1/tickets/?status=NEW",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
tickets = response.json()
|
|
assert all(t["status"] == "NEW" for t in tickets)
|
|
|
|
async def test_filter_by_priority(
|
|
self,
|
|
client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
test_tenant: Tenant,
|
|
test_admin_user: User,
|
|
test_category: Category,
|
|
auth_headers_admin: dict
|
|
):
|
|
"""Test filtrar tickets por prioridad."""
|
|
|
|
# Crear tickets con diferentes prioridades
|
|
ticket_low = Ticket(
|
|
tenant_id=test_tenant.id,
|
|
title="Low priority",
|
|
description="Description",
|
|
status=TicketStatus.NEW,
|
|
priority=TicketPriority.LOW,
|
|
created_by=test_admin_user.id,
|
|
category_id=test_category.id
|
|
)
|
|
ticket_urgent = Ticket(
|
|
tenant_id=test_tenant.id,
|
|
title="Urgent priority",
|
|
description="Description",
|
|
status=TicketStatus.NEW,
|
|
priority=TicketPriority.URGENT,
|
|
created_by=test_admin_user.id,
|
|
category_id=test_category.id
|
|
)
|
|
db_session.add_all([ticket_low, ticket_urgent])
|
|
await db_session.commit()
|
|
|
|
# Filtrar por URGENT
|
|
response = await client.get(
|
|
"/v1/tickets/?priority=URGENT",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
tickets = response.json()
|
|
assert all(t["priority"] == "URGENT" for t in tickets)
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.db
|
|
class TestTicketPermissions:
|
|
"""Tests de permisos en tickets."""
|
|
|
|
async def test_client_can_create_ticket(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant,
|
|
test_category: Category,
|
|
auth_headers_client: dict
|
|
):
|
|
"""Test que cliente puede crear tickets."""
|
|
|
|
response = await client.post(
|
|
"/v1/tickets/",
|
|
headers={
|
|
**auth_headers_client,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
},
|
|
json={
|
|
"title": "Client ticket",
|
|
"description": "Created by client",
|
|
"priority": "MEDIUM",
|
|
"category_id": str(test_category.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
|
|
async def test_client_can_only_see_own_tickets(
|
|
self,
|
|
client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
test_tenant: Tenant,
|
|
test_client_user: User,
|
|
test_admin_user: User,
|
|
test_category: Category,
|
|
auth_headers_client: dict
|
|
):
|
|
"""Test que cliente solo ve sus propios tickets."""
|
|
|
|
# Ticket del cliente
|
|
ticket_own = Ticket(
|
|
tenant_id=test_tenant.id,
|
|
title="My ticket",
|
|
description="Description",
|
|
status=TicketStatus.NEW,
|
|
priority=TicketPriority.MEDIUM,
|
|
created_by=test_client_user.id,
|
|
category_id=test_category.id
|
|
)
|
|
|
|
# Ticket de otro usuario
|
|
ticket_other = Ticket(
|
|
tenant_id=test_tenant.id,
|
|
title="Other ticket",
|
|
description="Description",
|
|
status=TicketStatus.NEW,
|
|
priority=TicketPriority.MEDIUM,
|
|
created_by=test_admin_user.id,
|
|
category_id=test_category.id
|
|
)
|
|
|
|
db_session.add_all([ticket_own, ticket_other])
|
|
await db_session.commit()
|
|
|
|
# Cliente lista tickets
|
|
response = await client.get(
|
|
"/v1/tickets/",
|
|
headers={
|
|
**auth_headers_client,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
tickets = response.json()
|
|
|
|
# Solo debe ver su propio ticket
|
|
ticket_ids = [t["id"] for t in tickets]
|
|
assert str(ticket_own.id) in ticket_ids
|
|
assert str(ticket_other.id) not in ticket_ids
|
|
|
|
async def test_agent_can_see_all_tenant_tickets(
|
|
self,
|
|
client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
test_tenant: Tenant,
|
|
test_agent_user: User,
|
|
test_admin_user: User,
|
|
test_category: Category,
|
|
auth_headers_agent: dict
|
|
):
|
|
"""Test que agente ve todos los tickets del tenant."""
|
|
|
|
# Crear tickets de diferentes usuarios
|
|
ticket_1 = Ticket(
|
|
tenant_id=test_tenant.id,
|
|
title="Ticket 1",
|
|
description="Description",
|
|
status=TicketStatus.NEW,
|
|
priority=TicketPriority.MEDIUM,
|
|
created_by=test_agent_user.id,
|
|
category_id=test_category.id
|
|
)
|
|
ticket_2 = Ticket(
|
|
tenant_id=test_tenant.id,
|
|
title="Ticket 2",
|
|
description="Description",
|
|
status=TicketStatus.NEW,
|
|
priority=TicketPriority.MEDIUM,
|
|
created_by=test_admin_user.id,
|
|
category_id=test_category.id
|
|
)
|
|
|
|
db_session.add_all([ticket_1, ticket_2])
|
|
await db_session.commit()
|
|
|
|
# Agente lista tickets
|
|
response = await client.get(
|
|
"/v1/tickets/",
|
|
headers={
|
|
**auth_headers_agent,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
tickets = response.json()
|
|
|
|
# Debe ver ambos tickets
|
|
assert len(tickets) >= 2
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.db
|
|
class TestTicketAttachmentPermissions:
|
|
async def test_client_cannot_download_other_users_attachment(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant,
|
|
test_category: Category,
|
|
auth_headers_admin: dict,
|
|
auth_headers_client: dict,
|
|
):
|
|
# Admin crea ticket
|
|
create_resp = await client.post(
|
|
"/v1/tickets/",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id),
|
|
},
|
|
json={
|
|
"title": "Admin ticket",
|
|
"description": "Ticket with attachment",
|
|
"priority": "MEDIUM",
|
|
"category_id": str(test_category.id),
|
|
},
|
|
)
|
|
assert create_resp.status_code == 201
|
|
ticket_id = create_resp.json()["id"]
|
|
|
|
# Admin sube adjunto (PDF válido por magic bytes)
|
|
pdf_bytes = b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n1 0 obj\n<<>>\nendobj\ntrailer\n<<>>\n%%EOF\n"
|
|
upload_resp = await client.post(
|
|
f"/v1/tickets/{ticket_id}/attachments",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id),
|
|
},
|
|
files={
|
|
"file": ("test.pdf", pdf_bytes, "application/pdf"),
|
|
},
|
|
)
|
|
assert upload_resp.status_code == 201
|
|
attachment_data = upload_resp.json()["data"]
|
|
attachment_id = attachment_data["id"]
|
|
|
|
# Cliente intenta descargar adjunto de ticket ajeno -> 404
|
|
download_resp = await client.get(
|
|
f"/v1/tickets/{ticket_id}/attachments/{attachment_id}/download",
|
|
headers={
|
|
**auth_headers_client,
|
|
"X-Tenant-ID": str(test_tenant.id),
|
|
},
|
|
)
|
|
assert download_resp.status_code == 404
|
|
|
|
# Limpieza del archivo subido (mejor esfuerzo)
|
|
try:
|
|
uploaded_path = file_handler.get_file_path(attachment_data["file_path"])
|
|
if uploaded_path.exists():
|
|
uploaded_path.unlink()
|
|
except Exception:
|
|
pass
|