egg/OCR
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add admin dashboard, audit logs, token expiry check and test suite

Browse Source 
Frontend Features:
- Add ProtectedRoute component with token expiry validation
- Create AdminDashboardPage with system statistics and user management
- Create AuditLogsPage with filtering and pagination
- Add admin-only navigation (Shield icon) for ymirliu@panjit.com.tw
- Add admin API methods to apiV2 service
- Add admin type definitions (SystemStats, AuditLog, etc.)

Token Management:
- Auto-redirect to login on token expiry
- Check authentication on route change
- Show loading state during auth check
- Admin privilege verification

Backend Testing:
- Add pytest configuration (pytest.ini)
- Create test fixtures (conftest.py)
- Add unit tests for auth, tasks, and admin endpoints
- Add integration tests for complete workflows
- Test user isolation and admin access control

Documentation:
- Add TESTING.md with comprehensive testing guide
- Include test running instructions
- Document fixtures and best practices

Routes:
- /admin - Admin dashboard (admin only)
- /admin/audit-logs - Audit logs viewer (admin only)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
egg
2025-11-16 18:01:50 +08:00
parent fd98018ddd
commit 8f94191914
13 changed files with 1554 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
backend/tests/test_admin.py Normal file
View File

@@ -0,0 +1,56 @@
"""
Unit tests for admin endpoints
"""
import pytest
class TestAdmin:
"""Test admin endpoints"""
def test_get_system_stats(self, client, admin_token):
"""Test get system statistics"""
response = client.get(
'/api/v2/admin/stats',
headers={'Authorization': f'Bearer {admin_token}'}
)
assert response.status_code == 200
data = response.json()
assert 'total_users' in data
assert 'total_tasks' in data
assert 'task_stats' in data
def test_get_system_stats_non_admin(self, client, auth_token):
"""Test that non-admin cannot access admin endpoints"""
response = client.get(
'/api/v2/admin/stats',
headers={'Authorization': f'Bearer {auth_token}'}
)
assert response.status_code == 403
def test_list_users(self, client, admin_token):
"""Test list all users"""
response = client.get(
'/api/v2/admin/users',
headers={'Authorization': f'Bearer {admin_token}'}
)
assert response.status_code == 200
data = response.json()
assert 'users' in data
assert 'total' in data
def test_get_audit_logs(self, client, admin_token):
"""Test get audit logs"""
response = client.get(
'/api/v2/admin/audit-logs',
headers={'Authorization': f'Bearer {admin_token}'}
)
assert response.status_code == 200
data = response.json()
assert 'logs' in data
assert 'total' in data
assert 'page' in data

75
backend/tests/test_auth.py Normal file
View File

@@ -0,0 +1,75 @@
"""
Unit tests for authentication endpoints
"""
import pytest
from unittest.mock import patch, MagicMock
class TestAuth:
"""Test authentication endpoints"""
def test_login_success(self, client, db):
"""Test successful login"""
# Mock external auth service
with patch('app.routers.auth.external_auth_service.authenticate_user') as mock_auth:
mock_auth.return_value = (True, {
'access_token': 'test-token',
'expires_in': 3600,
'user_info': {
'email': 'test@example.com',
'name': 'Test User'
}
}, None)
response = client.post('/api/v2/auth/login', json={
'username': 'test@example.com',
'password': 'password123'
})
assert response.status_code == 200
data = response.json()
assert 'access_token' in data
assert data['token_type'] == 'bearer'
assert 'user' in data
def test_login_invalid_credentials(self, client):
"""Test login with invalid credentials"""
with patch('app.routers.auth.external_auth_service.authenticate_user') as mock_auth:
mock_auth.return_value = (False, None, 'Invalid credentials')
response = client.post('/api/v2/auth/login', json={
'username': 'test@example.com',
'password': 'wrongpassword'
})
assert response.status_code == 401
assert 'detail' in response.json()
def test_get_me(self, client, auth_token):
"""Test get current user info"""
response = client.get(
'/api/v2/auth/me',
headers={'Authorization': f'Bearer {auth_token}'}
)
assert response.status_code == 200
data = response.json()
assert 'email' in data
assert 'display_name' in data
def test_get_me_unauthorized(self, client):
"""Test get current user without token"""
response = client.get('/api/v2/auth/me')
assert response.status_code == 403
def test_logout(self, client, auth_token):
"""Test logout"""
response = client.post(
'/api/v2/auth/logout',
headers={'Authorization': f'Bearer {auth_token}'}
)
assert response.status_code == 200
data = response.json()
assert data['message'] == 'Logged out successfully'

161
backend/tests/test_integration.py Normal file
View File

@@ -0,0 +1,161 @@
"""
Integration tests for Tool_OCR
Tests the complete flow of authentication, task creation, and file operations
"""
import pytest
from unittest.mock import patch
class TestIntegration:
"""Integration tests for end-to-end workflows"""
def test_complete_auth_and_task_flow(self, client, db):
"""Test complete flow: login -> create task -> get task -> delete task"""
# Step 1: Login
with patch('app.routers.auth.external_auth_service.authenticate_user') as mock_auth:
mock_auth.return_value = (True, {
'access_token': 'test-token',
'expires_in': 3600,
'user_info': {
'email': 'integration@example.com',
'name': 'Integration Test User'
}
}, None)
login_response = client.post('/api/v2/auth/login', json={
'username': 'integration@example.com',
'password': 'password123'
})
assert login_response.status_code == 200
token = login_response.json()['access_token']
headers = {'Authorization': f'Bearer {token}'}
# Step 2: Create task
create_response = client.post(
'/api/v2/tasks/',
headers=headers,
json={
'filename': 'integration_test.pdf',
'file_type': 'application/pdf'
}
)
assert create_response.status_code == 201
task_data = create_response.json()
task_id = task_data['task_id']
# Step 3: Get task
get_response = client.get(
f'/api/v2/tasks/{task_id}',
headers=headers
)
assert get_response.status_code == 200
assert get_response.json()['task_id'] == task_id
# Step 4: List tasks
list_response = client.get(
'/api/v2/tasks/',
headers=headers
)
assert list_response.status_code == 200
assert len(list_response.json()['tasks']) > 0
# Step 5: Get stats
stats_response = client.get(
'/api/v2/tasks/stats',
headers=headers
)
assert stats_response.status_code == 200
stats = stats_response.json()
assert stats['total'] > 0
assert stats['pending'] > 0
# Step 6: Delete task
delete_response = client.delete(
f'/api/v2/tasks/{task_id}',
headers=headers
)
assert delete_response.status_code == 200
# Step 7: Verify deletion
get_after_delete = client.get(
f'/api/v2/tasks/{task_id}',
headers=headers
)
assert get_after_delete.status_code == 404
def test_admin_workflow(self, client, db):
"""Test admin workflow: login as admin -> access admin endpoints"""
# Login as admin
with patch('app.routers.auth.external_auth_service.authenticate_user') as mock_auth:
mock_auth.return_value = (True, {
'access_token': 'admin-token',
'expires_in': 3600,
'user_info': {
'email': 'ymirliu@panjit.com.tw',
'name': 'Admin User'
}
}, None)
login_response = client.post('/api/v2/auth/login', json={
'username': 'ymirliu@panjit.com.tw',
'password': 'adminpass'
})
assert login_response.status_code == 200
token = login_response.json()['access_token']
headers = {'Authorization': f'Bearer {token}'}
# Access admin endpoints
stats_response = client.get('/api/v2/admin/stats', headers=headers)
assert stats_response.status_code == 200
users_response = client.get('/api/v2/admin/users', headers=headers)
assert users_response.status_code == 200
logs_response = client.get('/api/v2/admin/audit-logs', headers=headers)
assert logs_response.status_code == 200
def test_task_lifecycle(self, client, auth_token, test_task, db):
"""Test complete task lifecycle: pending -> processing -> completed"""
headers = {'Authorization': f'Bearer {auth_token}'}
# Check initial status
response = client.get(f'/api/v2/tasks/{test_task.task_id}', headers=headers)
assert response.json()['status'] == 'pending'
# Start task
start_response = client.post(
f'/api/v2/tasks/{test_task.task_id}/start',
headers=headers
)
assert start_response.status_code == 200
assert start_response.json()['status'] == 'processing'
# Update task to completed
update_response = client.patch(
f'/api/v2/tasks/{test_task.task_id}',
headers=headers,
json={
'status': 'completed',
'processing_time_ms': 1500
}
)
assert update_response.status_code == 200
assert update_response.json()['status'] == 'completed'
# Verify final state
final_response = client.get(f'/api/v2/tasks/{test_task.task_id}', headers=headers)
final_data = final_response.json()
assert final_data['status'] == 'completed'
assert final_data['processing_time_ms'] == 1500

105
backend/tests/test_tasks.py Normal file
View File

@@ -0,0 +1,105 @@
"""
Unit tests for task management endpoints
"""
import pytest
from app.models.task import Task
class TestTasks:
"""Test task management endpoints"""
def test_create_task(self, client, auth_token):
"""Test task creation"""
response = client.post(
'/api/v2/tasks/',
headers={'Authorization': f'Bearer {auth_token}'},
json={
'filename': 'test.pdf',
'file_type': 'application/pdf'
}
)
assert response.status_code == 201
data = response.json()
assert 'task_id' in data
assert data['filename'] == 'test.pdf'
assert data['status'] == 'pending'
def test_list_tasks(self, client, auth_token, test_task):
"""Test listing user tasks"""
response = client.get(
'/api/v2/tasks/',
headers={'Authorization': f'Bearer {auth_token}'}
)
assert response.status_code == 200
data = response.json()
assert 'tasks' in data
assert 'total' in data
assert len(data['tasks']) > 0
def test_get_task(self, client, auth_token, test_task):
"""Test get single task"""
response = client.get(
f'/api/v2/tasks/{test_task.task_id}',
headers={'Authorization': f'Bearer {auth_token}'}
)
assert response.status_code == 200
data = response.json()
assert data['task_id'] == test_task.task_id
def test_get_task_stats(self, client, auth_token, test_task):
"""Test get task statistics"""
response = client.get(
'/api/v2/tasks/stats',
headers={'Authorization': f'Bearer {auth_token}'}
)
assert response.status_code == 200
data = response.json()
assert 'total' in data
assert 'pending' in data
assert 'processing' in data
assert 'completed' in data
assert 'failed' in data
def test_delete_task(self, client, auth_token, test_task):
"""Test task deletion"""
response = client.delete(
f'/api/v2/tasks/{test_task.task_id}',
headers={'Authorization': f'Bearer {auth_token}'}
)
assert response.status_code == 200
def test_user_isolation(self, client, db, test_user):
"""Test that users can only access their own tasks"""
# Create another user
from app.models.user import User
other_user = User(email="other@example.com", display_name="Other User")
db.add(other_user)
db.commit()
# Create task for other user
other_task = Task(
user_id=other_user.id,
task_id="other-task-123",
filename="other.pdf",
status="pending"
)
db.add(other_task)
db.commit()
# Create token for test_user
from app.core.security import create_access_token
token = create_access_token({"sub": str(test_user.id)})
# Try to access other user's task
response = client.get(
f'/api/v2/tasks/{other_task.task_id}',
headers={'Authorization': f'Bearer {token}'}
)
assert response.status_code == 404 # Task not found (user isolation)