chihlasm/resolutionflow
1
0
Fork 0
Code Issues 23 Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
89d343d49abe2e6aa8ba4e3b11c3af330d922ac2
resolutionflow/backend/tests/test_sessions.py
Michael Chihlas 9f92547309 feat: implement session history search and filtering (Issue #35) 
Implement comprehensive search and filtering for Session History to dramatically
improve findability of past troubleshooting sessions.

Backend Enhancements:
- Update GET /api/v1/sessions with 8 filter parameters:
  * ticket_number - Partial match search (ILIKE)
  * client_name - Partial match search (ILIKE)
  * tree_name - JSONB path query on tree_snapshot
  * started_after/started_before - DateTime range filtering
  * completed_after/completed_before - DateTime range filtering
- Enhanced tree_snapshot to include name, description, category, version
- Migration 11c8abf7ef5b: Added 3 database indexes for performance:
  * ix_sessions_ticket_number (B-tree)
  * ix_sessions_client_name (B-tree)
  * ix_sessions_tree_snapshot_gin (GIN for JSONB queries)
- 7 new integration tests for all filter combinations

Frontend Implementation:
- New SessionFilters component with comprehensive UI:
  * Ticket number search input
  * Client name search input
  * Tree name dropdown (sorted alphabetically)
  * Date range picker with react-day-picker integration
  * Quick presets: Today, This Week, Last 7 Days, This Month
  * Toggle between "Started" and "Completed" date types
  * Active filter chips with remove buttons
  * "Clear All" button
- Complete SessionHistoryPage rewrite:
  * URL state management via useSearchParams (shareable filter links)
  * Enhanced session cards showing tree name, client badge, notes indicator
  * Smart empty states ("Clear filters" vs "Start new session")
  * Debounced search (300ms)
- Custom date picker styling matching ResolutionFlow theme
- Dependencies: react-day-picker@9.13.1, date-fns@4.1.0

Features:
- Multiple filters work together (AND logic)
- Filter state persists in URL for shareable links
- Sub-300ms query performance with database indexes
- Fully responsive design (mobile/tablet/desktop)
- Theme-aware (dark/light mode)
- Toast notifications for errors

Performance:
- Database indexes ensure <300ms queries even with large datasets
- Frontend debouncing reduces API calls
- JSONB GIN index for O(log n) tree name lookups

Bundle Impact:
- JS: +87.83 KB (+12.2%, due to react-day-picker library)
- CSS: +10.53 KB (+25.8%, date picker styles)
- Gzipped: +24.52 KB JS, +1.82 KB CSS

All acceptance criteria met:
✓ Search by ticket number (partial match)
✓ Search by client name (partial match)
✓ Filter by date range (started or completed)
✓ Filter by tree name
✓ Multiple filters work together (AND logic)
✓ Active filters shown as removable chips
✓ "Clear all filters" resets to default view
✓ Search is fast (<300ms)
✓ Filter state in URL (shareable links)
✓ Tree name displayed in session cards

Tests: 34/34 session tests passing (7 new filter tests)

Closes #35

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-07 21:17:25 -05:00

933 lines
30 KiB
Python
Raw Blame History

"""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"
@pytest.mark.asyncio
async def test_export_includes_scratchpad_markdown(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test that markdown export includes scratchpad content."""
# Create session with scratchpad
create_response = await client.post(
"/api/v1/sessions",
json={"tree_id": test_tree["id"], "ticket_number": "SP-001"},
headers=auth_headers
)
session_id = create_response.json()["id"]
# Add scratchpad content
await client.patch(
f"/api/v1/sessions/{session_id}/scratchpad",
json={"scratchpad": "- Server IP: 192.168.1.50\n- Error: 0x80070005"},
headers=auth_headers
)
# Export as markdown
response = await client.post(
f"/api/v1/sessions/{session_id}/export",
json={"format": "markdown", "include_tree_info": True},
headers=auth_headers
)
assert response.status_code == 200
content = response.text
assert "## Evidence / Reference" in content
assert "192.168.1.50" in content
assert "0x80070005" in content
@pytest.mark.asyncio
async def test_export_includes_scratchpad_text(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test that text export includes scratchpad content."""
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.patch(
f"/api/v1/sessions/{session_id}/scratchpad",
json={"scratchpad": "Error code: 12345"},
headers=auth_headers
)
response = await client.post(
f"/api/v1/sessions/{session_id}/export",
json={"format": "text"},
headers=auth_headers
)
assert response.status_code == 200
content = response.text
assert "EVIDENCE / REFERENCE" in content
assert "Error code: 12345" in content
@pytest.mark.asyncio
async def test_export_includes_scratchpad_html(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test that HTML export includes scratchpad content."""
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.patch(
f"/api/v1/sessions/{session_id}/scratchpad",
json={"scratchpad": "DNS server: 10.0.0.5"},
headers=auth_headers
)
response = await client.post(
f"/api/v1/sessions/{session_id}/export",
json={"format": "html"},
headers=auth_headers
)
assert response.status_code == 200
content = response.text
assert "Evidence / Reference" in content
assert "DNS server: 10.0.0.5" in content
@pytest.mark.asyncio
async def test_export_excludes_empty_scratchpad(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test that export omits scratchpad section when empty."""
create_response = await client.post(
"/api/v1/sessions",
json={"tree_id": test_tree["id"]},
headers=auth_headers
)
session_id = create_response.json()["id"]
# Export without setting scratchpad
response = await client.post(
f"/api/v1/sessions/{session_id}/export",
json={"format": "markdown"},
headers=auth_headers
)
assert response.status_code == 200
content = response.text
assert "Evidence / Reference" not in content
@pytest.mark.asyncio
async def test_export_excludes_whitespace_only_scratchpad(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test that export omits scratchpad section when only whitespace."""
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.patch(
f"/api/v1/sessions/{session_id}/scratchpad",
json={"scratchpad": " \n \n "},
headers=auth_headers
)
response = await client.post(
f"/api/v1/sessions/{session_id}/export",
json={"format": "markdown"},
headers=auth_headers
)
assert response.status_code == 200
content = response.text
assert "Evidence / Reference" not in content
@pytest.mark.asyncio
async def test_html_export_escapes_script_tags(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test that HTML export escapes script tags in user content (XSS prevention)."""
session_data = {
"tree_id": test_tree["id"],
"ticket_number": '<script>alert("xss")</script>',
"client_name": '<img onerror="alert(1)" src=x>'
}
create_response = await client.post(
"/api/v1/sessions", json=session_data, headers=auth_headers
)
session_id = create_response.json()["id"]
response = await client.post(
f"/api/v1/sessions/{session_id}/export",
json={"format": "html", "include_tree_info": True},
headers=auth_headers
)
assert response.status_code == 200
content = response.text
assert '<script>' not in content
assert '&lt;script&gt;' in content
assert '&lt;img' in content
@pytest.mark.asyncio
async def test_html_export_escapes_special_chars(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test that HTML export properly escapes special characters."""
session_data = {
"tree_id": test_tree["id"],
"ticket_number": 'TICK-001 <b>"bold"</b> & \'quoted\''
}
create_response = await client.post(
"/api/v1/sessions", json=session_data, headers=auth_headers
)
session_id = create_response.json()["id"]
response = await client.post(
f"/api/v1/sessions/{session_id}/export",
json={"format": "html", "include_tree_info": True},
headers=auth_headers
)
assert response.status_code == 200
content = response.text
assert '&amp;' in content
assert '&lt;b&gt;' in content
@pytest.mark.asyncio
async def test_html_export_escapes_scratchpad(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test that HTML export escapes scratchpad content."""
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.patch(
f"/api/v1/sessions/{session_id}/scratchpad",
json={"scratchpad": '<script>document.cookie</script>'},
headers=auth_headers
)
response = await client.post(
f"/api/v1/sessions/{session_id}/export",
json={"format": "html"},
headers=auth_headers
)
assert response.status_code == 200
content = response.text
assert '<script>' not in content
assert '&lt;script&gt;' in content
@pytest.mark.asyncio
async def test_start_session_on_others_private_tree_forbidden(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test that a user cannot start a session on another user's private tree."""
# Register a second user
await client.post("/api/v1/auth/register", json={
"email": "other@example.com",
"password": "OtherPassword123!",
"name": "Other User"
})
login_resp = await client.post("/api/v1/auth/login/json", json={
"email": "other@example.com",
"password": "OtherPassword123!"
})
other_headers = {"Authorization": f"Bearer {login_resp.json()['access_token']}"}
# test_tree is owned by test_user (not public, not default)
response = await client.post(
"/api/v1/sessions",
json={"tree_id": test_tree["id"]},
headers=other_headers
)
assert response.status_code == 403
@pytest.mark.asyncio
async def test_start_session_super_admin_any_tree(
self, client: AsyncClient, admin_auth_headers: dict, test_tree: dict
):
"""Test that a super admin can start a session on any tree."""
response = await client.post(
"/api/v1/sessions",
json={"tree_id": test_tree["id"]},
headers=admin_auth_headers
)
assert response.status_code == 201
@pytest.mark.asyncio
async def test_filter_sessions_by_ticket_number(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test filtering sessions by ticket number (partial match)."""
# Create sessions with different ticket numbers
await client.post(
"/api/v1/sessions",
json={"tree_id": test_tree["id"], "ticket_number": "INC-12345"},
headers=auth_headers
)
await client.post(
"/api/v1/sessions",
json={"tree_id": test_tree["id"], "ticket_number": "REQ-67890"},
headers=auth_headers
)
# Filter by ticket number
response = await client.get(
"/api/v1/sessions?ticket_number=INC",
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert len(data) == 1
assert data[0]["ticket_number"] == "INC-12345"
@pytest.mark.asyncio
async def test_filter_sessions_by_client_name(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test filtering sessions by client name (partial match, case-insensitive)."""
# Create sessions with different clients
await client.post(
"/api/v1/sessions",
json={"tree_id": test_tree["id"], "client_name": "Acme Corporation"},
headers=auth_headers
)
await client.post(
"/api/v1/sessions",
json={"tree_id": test_tree["id"], "client_name": "TechStart Inc"},
headers=auth_headers
)
# Filter by client name (case-insensitive partial match)
response = await client.get(
"/api/v1/sessions?client_name=acme",
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert len(data) == 1
assert data[0]["client_name"] == "Acme Corporation"
@pytest.mark.asyncio
async def test_filter_sessions_by_tree_name(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test filtering sessions by tree name from snapshot."""
# Create session (tree_snapshot includes tree name)
response = await client.post(
"/api/v1/sessions",
json={"tree_id": test_tree["id"]},
headers=auth_headers
)
assert response.status_code == 201
# Filter by tree name (partial match from snapshot)
tree_name_part = test_tree["name"][:5] # First 5 chars
response = await client.get(
f"/api/v1/sessions?tree_name={tree_name_part}",
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert len(data) >= 1
assert test_tree["name"] in data[0]["tree_snapshot"]["name"]
@pytest.mark.asyncio
async def test_filter_sessions_by_started_date_range(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test filtering sessions by started date range."""
from datetime import datetime, timezone, timedelta
from urllib.parse import quote
# Create a session
response = await client.post(
"/api/v1/sessions",
json={"tree_id": test_tree["id"], "ticket_number": "TEST-001"},
headers=auth_headers
)
assert response.status_code == 201
# Get current time and create date range
now = datetime.now(timezone.utc)
yesterday = now - timedelta(days=1)
tomorrow = now + timedelta(days=1)
# Filter by started date range (should include the session)
# URL encode the datetime strings
started_after = quote(yesterday.isoformat())
started_before = quote(tomorrow.isoformat())
response = await client.get(
f"/api/v1/sessions?started_after={started_after}&started_before={started_before}",
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert len(data) >= 1
assert data[0]["ticket_number"] == "TEST-001"
@pytest.mark.asyncio
async def test_filter_sessions_by_completed_date_range(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test filtering sessions by completed date range."""
from datetime import datetime, timezone, timedelta
from urllib.parse import quote
# Create and complete a session
create_response = await client.post(
"/api/v1/sessions",
json={"tree_id": test_tree["id"], "ticket_number": "TEST-002"},
headers=auth_headers
)
session_id = create_response.json()["id"]
await client.post(
f"/api/v1/sessions/{session_id}/complete",
headers=auth_headers
)
# Get current time and create date range
now = datetime.now(timezone.utc)
yesterday = now - timedelta(days=1)
tomorrow = now + timedelta(days=1)
# Filter by completed date range
completed_after = quote(yesterday.isoformat())
completed_before = quote(tomorrow.isoformat())
response = await client.get(
f"/api/v1/sessions?completed_after={completed_after}&completed_before={completed_before}",
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert len(data) >= 1
assert any(s["ticket_number"] == "TEST-002" for s in data)
@pytest.mark.asyncio
async def test_filter_sessions_combined(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test combining multiple filters (AND logic)."""
# Create sessions with various attributes
await client.post(
"/api/v1/sessions",
json={
"tree_id": test_tree["id"],
"ticket_number": "INC-111",
"client_name": "Client A"
},
headers=auth_headers
)
await client.post(
"/api/v1/sessions",
json={
"tree_id": test_tree["id"],
"ticket_number": "INC-222",
"client_name": "Client B"
},
headers=auth_headers
)
# Filter by both ticket and client (should match only one)
response = await client.get(
"/api/v1/sessions?ticket_number=INC-111&client_name=Client A",
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert len(data) == 1
assert data[0]["ticket_number"] == "INC-111"
assert data[0]["client_name"] == "Client A"
@pytest.mark.asyncio
async def test_filter_sessions_no_results(
self, client: AsyncClient, auth_headers: dict, test_tree: dict
):
"""Test that filtering returns empty list when no matches."""
response = await client.get(
"/api/v1/sessions?ticket_number=NONEXISTENT-999",
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) == 0
Reference in New Issue View Git Blame Copy Permalink