464 lines
14 KiB
Python
464 lines
14 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)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_session_has_scratchpad(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test that new sessions include scratchpad field."""
|
|
session_data = {
|
|
"tree_id": test_tree["id"],
|
|
}
|
|
|
|
response = await client.post(
|
|
"/api/v1/sessions",
|
|
json=session_data,
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert "scratchpad" in data
|
|
assert data["scratchpad"] == ""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_scratchpad_via_put(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test updating scratchpad through the existing PUT endpoint."""
|
|
# 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 scratchpad via PUT
|
|
update_data = {
|
|
"scratchpad": "- Server IP: 192.168.1.50\n- Error: 0x80070005"
|
|
}
|
|
|
|
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["scratchpad"] == update_data["scratchpad"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_scratchpad(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test the dedicated PATCH scratchpad endpoint."""
|
|
# 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"]
|
|
|
|
# Patch scratchpad
|
|
response = await client.patch(
|
|
f"/api/v1/sessions/{session_id}/scratchpad",
|
|
json={"scratchpad": "- IP: 10.0.0.1\n- User: jsmith"},
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["scratchpad"] == "- IP: 10.0.0.1\n- User: jsmith"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_scratchpad_not_found(
|
|
self, client: AsyncClient, auth_headers: dict
|
|
):
|
|
"""Test PATCH scratchpad with invalid session ID."""
|
|
import uuid
|
|
fake_id = str(uuid.uuid4())
|
|
|
|
response = await client.patch(
|
|
f"/api/v1/sessions/{fake_id}/scratchpad",
|
|
json={"scratchpad": "test"},
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_scratchpad_empty_string(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test PATCH scratchpad with empty string (clear scratchpad)."""
|
|
# Create session and set scratchpad
|
|
create_response = await client.post(
|
|
"/api/v1/sessions",
|
|
json={"tree_id": test_tree["id"]},
|
|
headers=auth_headers
|
|
)
|
|
session_id = create_response.json()["id"]
|
|
|
|
# Set scratchpad
|
|
await client.patch(
|
|
f"/api/v1/sessions/{session_id}/scratchpad",
|
|
json={"scratchpad": "some notes"},
|
|
headers=auth_headers
|
|
)
|
|
|
|
# Clear scratchpad
|
|
response = await client.patch(
|
|
f"/api/v1/sessions/{session_id}/scratchpad",
|
|
json={"scratchpad": ""},
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["scratchpad"] == ""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_scratchpad_completed_session(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test that scratchpad can still be updated on completed sessions."""
|
|
# 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
|
|
)
|
|
|
|
# Should still be able to update scratchpad on completed sessions
|
|
response = await client.patch(
|
|
f"/api/v1/sessions/{session_id}/scratchpad",
|
|
json={"scratchpad": "post-resolution notes"},
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["scratchpad"] == "post-resolution notes"
|