7d96807fb1
DateTime Timezone Handling (Critical Bug Fix): - Updated all models to use DateTime(timezone=True) for PostgreSQL - Changed datetime defaults to lambda: datetime.now(timezone.utc) - Fixed mixing of timezone-aware and timezone-naive datetime objects - Resolved Internal Server Errors in session completion endpoint - Affected models: User, Team, Tree, Session, Attachment Production Logging System: - Created logging_config.py with structured logging setup - Added log rotation (10MB files, 10 backups) for production - Implemented RequestLoggingMiddleware with correlation IDs - Added ErrorLoggingMiddleware for comprehensive error tracking - Integrated logging into main.py application startup - Supports dev/prod modes with appropriate log levels Integration Tests - Session Workflow: - Created test_sessions.py with 12 comprehensive tests - Session lifecycle: create, update, complete - Session export in multiple formats (markdown, text, HTML) - Error handling and authorization checks - Added pytest.ini with coverage configuration - Added requirements-dev.txt with pytest dependencies Following 2026 FastAPI best practices for timezone handling and structured logging. Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
318 lines
9.7 KiB
Python
318 lines
9.7 KiB
Python
"""Integration tests for session endpoints."""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
|
|
class TestSessions:
|
|
"""Test suite for troubleshooting session endpoints."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_session(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test starting a new troubleshooting session."""
|
|
session_data = {
|
|
"tree_id": test_tree["id"],
|
|
"ticket_number": "TICKET-123",
|
|
"client_name": "Test Client"
|
|
}
|
|
|
|
response = await client.post(
|
|
"/api/v1/sessions",
|
|
json=session_data,
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["tree_id"] == test_tree["id"]
|
|
assert data["ticket_number"] == session_data["ticket_number"]
|
|
assert data["client_name"] == session_data["client_name"]
|
|
assert data["path_taken"] == []
|
|
assert data["decisions"] == []
|
|
assert data["completed_at"] is None
|
|
assert "id" in data
|
|
assert "started_at" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test retrieving a specific session."""
|
|
# Create a session first
|
|
create_response = await client.post(
|
|
"/api/v1/sessions",
|
|
json={"tree_id": test_tree["id"]},
|
|
headers=auth_headers
|
|
)
|
|
session_id = create_response.json()["id"]
|
|
|
|
# Get the session
|
|
response = await client.get(
|
|
f"/api/v1/sessions/{session_id}",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == session_id
|
|
assert data["tree_id"] == test_tree["id"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_sessions(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test listing user's sessions."""
|
|
# Create a session
|
|
await client.post(
|
|
"/api/v1/sessions",
|
|
json={"tree_id": test_tree["id"]},
|
|
headers=auth_headers
|
|
)
|
|
|
|
# List sessions
|
|
response = await client.get("/api/v1/sessions", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) >= 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_session_path(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test updating session with path taken."""
|
|
# Create session
|
|
create_response = await client.post(
|
|
"/api/v1/sessions",
|
|
json={"tree_id": test_tree["id"]},
|
|
headers=auth_headers
|
|
)
|
|
session_id = create_response.json()["id"]
|
|
|
|
# Update path
|
|
update_data = {
|
|
"path_taken": ["root", "solution1"]
|
|
}
|
|
|
|
response = await client.put(
|
|
f"/api/v1/sessions/{session_id}",
|
|
json=update_data,
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["path_taken"] == update_data["path_taken"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_session_ticket(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test updating session metadata."""
|
|
# Create session
|
|
create_response = await client.post(
|
|
"/api/v1/sessions",
|
|
json={"tree_id": test_tree["id"]},
|
|
headers=auth_headers
|
|
)
|
|
session_id = create_response.json()["id"]
|
|
|
|
# Update metadata
|
|
update_data = {
|
|
"ticket_number": "UPDATED-456",
|
|
"client_name": "Updated Client"
|
|
}
|
|
|
|
response = await client.put(
|
|
f"/api/v1/sessions/{session_id}",
|
|
json=update_data,
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["ticket_number"] == update_data["ticket_number"]
|
|
assert data["client_name"] == update_data["client_name"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_complete_session(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test marking a session as complete."""
|
|
# Create session
|
|
create_response = await client.post(
|
|
"/api/v1/sessions",
|
|
json={"tree_id": test_tree["id"]},
|
|
headers=auth_headers
|
|
)
|
|
session_id = create_response.json()["id"]
|
|
|
|
# Complete session
|
|
response = await client.post(
|
|
f"/api/v1/sessions/{session_id}/complete",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["completed_at"] is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_complete_already_completed_session(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test that completing an already completed session fails."""
|
|
# Create and complete session
|
|
create_response = await client.post(
|
|
"/api/v1/sessions",
|
|
json={"tree_id": test_tree["id"]},
|
|
headers=auth_headers
|
|
)
|
|
session_id = create_response.json()["id"]
|
|
|
|
await client.post(
|
|
f"/api/v1/sessions/{session_id}/complete",
|
|
headers=auth_headers
|
|
)
|
|
|
|
# Try to complete again
|
|
response = await client.post(
|
|
f"/api/v1/sessions/{session_id}/complete",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_session_markdown(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test exporting session to markdown format."""
|
|
# Create session with ticket number
|
|
create_response = await client.post(
|
|
"/api/v1/sessions",
|
|
json={"tree_id": test_tree["id"], "ticket_number": "EXP-001"},
|
|
headers=auth_headers
|
|
)
|
|
session_id = create_response.json()["id"]
|
|
|
|
# Export as markdown
|
|
export_data = {
|
|
"format": "markdown",
|
|
"include_timestamps": True,
|
|
"include_tree_info": True
|
|
}
|
|
|
|
response = await client.post(
|
|
f"/api/v1/sessions/{session_id}/export",
|
|
json=export_data,
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
content = response.text
|
|
assert "EXP-001" in content # Should contain ticket number
|
|
assert "#" in content # Markdown headers
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_session_text(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test exporting session to text format."""
|
|
# Create session
|
|
create_response = await client.post(
|
|
"/api/v1/sessions",
|
|
json={"tree_id": test_tree["id"]},
|
|
headers=auth_headers
|
|
)
|
|
session_id = create_response.json()["id"]
|
|
|
|
# Export as text
|
|
export_data = {"format": "text"}
|
|
|
|
response = await client.post(
|
|
f"/api/v1/sessions/{session_id}/export",
|
|
json=export_data,
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "text/plain; charset=utf-8"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_session_html(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test exporting session to HTML format."""
|
|
# Create session
|
|
create_response = await client.post(
|
|
"/api/v1/sessions",
|
|
json={"tree_id": test_tree["id"]},
|
|
headers=auth_headers
|
|
)
|
|
session_id = create_response.json()["id"]
|
|
|
|
# Export as HTML
|
|
export_data = {"format": "html"}
|
|
|
|
response = await client.post(
|
|
f"/api/v1/sessions/{session_id}/export",
|
|
json=export_data,
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
content = response.text
|
|
assert "<!DOCTYPE html>" in content
|
|
assert "<html>" in content
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filter_sessions_by_completion(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test filtering sessions by completion status."""
|
|
# Create two sessions, complete one
|
|
create1 = await client.post(
|
|
"/api/v1/sessions",
|
|
json={"tree_id": test_tree["id"]},
|
|
headers=auth_headers
|
|
)
|
|
session1_id = create1.json()["id"]
|
|
|
|
await client.post(
|
|
"/api/v1/sessions",
|
|
json={"tree_id": test_tree["id"]},
|
|
headers=auth_headers
|
|
)
|
|
|
|
# Complete first session
|
|
await client.post(
|
|
f"/api/v1/sessions/{session1_id}/complete",
|
|
headers=auth_headers
|
|
)
|
|
|
|
# Get completed sessions
|
|
response = await client.get(
|
|
"/api/v1/sessions?completed=true",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) >= 1
|
|
assert all(s["completed_at"] is not None for s in data)
|
|
|
|
# Get incomplete sessions
|
|
response = await client.get(
|
|
"/api/v1/sessions?completed=false",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) >= 1
|
|
assert all(s["completed_at"] is None for s in data)
|