+
+ Statistics Dashboard
+
QR Code Dashboard
diff --git a/app/views/admin.py b/app/views/admin.py
index 0b1c23b..4282401 100644
--- a/app/views/admin.py
+++ b/app/views/admin.py
@@ -5,10 +5,15 @@ Admin interface for managing database records.
from fastapi import APIRouter, Depends, Request, Form, HTTPException, Query
from fastapi.responses import HTMLResponse, RedirectResponse
from sqlalchemy.orm import Session
-from sqlalchemy import inspect
+from sqlalchemy import inspect, func, desc, text
import json
from typing import Dict, Any, List, Type, Optional
import inspect as py_inspect
+from datetime import datetime, timedelta
+import time
+import os
+import psutil
+from dateutil.relativedelta import relativedelta
from ..db import SessionLocal, Base
from ..models import (
@@ -42,6 +47,168 @@ def get_db():
finally:
db.close()
+# Get user statistics for the dashboard
+def get_user_statistics(db: Session) -> Dict[str, Any]:
+ """Get user statistics for the admin dashboard."""
+ stats = {}
+
+ # Total users
+ stats["total_users"] = db.query(User).count()
+
+ # Active users (not disabled)
+ stats["active_users"] = db.query(User).filter(User.is_active == True).count()
+
+ # Verified users
+ stats["verified_users"] = db.query(User).filter(User.is_verified == True).count()
+
+ # Admin users
+ stats["admin_users"] = db.query(User).filter(User.is_admin == True).count()
+
+ # New registrations in the last 30 days
+ thirty_days_ago = datetime.utcnow() - timedelta(days=30)
+ stats["new_registrations_30d"] = db.query(User).filter(
+ User.created_at >= thirty_days_ago
+ ).count()
+
+ # Monthly user registration data for the chart (last 6 months)
+ monthly_data = []
+ month_labels = []
+
+ # Get the current month and year
+ current_date = datetime.now()
+
+ # Loop through the last 6 months
+ for i in range(5, -1, -1):
+ # Calculate month and year for this data point
+ month_date = current_date - relativedelta(months=i)
+ start_of_month = datetime(month_date.year, month_date.month, 1)
+
+ # For the current month, only count until today
+ if i == 0:
+ end_of_month = current_date
+ else:
+ # Calculate the end of the month
+ if month_date.month == 12:
+ end_of_month = datetime(month_date.year + 1, 1, 1) - timedelta(days=1)
+ else:
+ end_of_month = datetime(month_date.year, month_date.month + 1, 1) - timedelta(days=1)
+
+ # Format month as abbreviated month name
+ month_name = month_date.strftime('%b')
+ month_labels.append(month_name)
+
+ # Count users registered in this month
+ monthly_count = db.query(User).filter(
+ User.created_at >= start_of_month,
+ User.created_at <= end_of_month
+ ).count()
+
+ monthly_data.append(monthly_count)
+
+ # Add the data to the stats
+ stats["monthly_registrations"] = monthly_data
+ stats["month_labels"] = month_labels
+
+ return stats
+
+# Get team statistics for the dashboard
+def get_team_statistics(db: Session) -> Dict[str, Any]:
+ """Get team statistics for the admin dashboard."""
+ stats = {}
+
+ # Total teams
+ stats["total_teams"] = db.query(Team).count()
+
+ # Active teams
+ stats["active_teams"] = db.query(Team).filter(Team.is_active == True).count()
+
+ # Public teams
+ stats["public_teams"] = db.query(Team).filter(Team.is_public == True).count()
+
+ # Team size distribution - teams grouped by member count
+ # Modified query to correctly count team members and group by team
+ team_sizes = db.query(
+ TeamMembership.team_id,
+ func.count(TeamMembership.user_id).label('member_count')
+ ).group_by(TeamMembership.team_id).subquery()
+
+ # Now we can query the distribution from the subquery
+ team_distribution = db.query(
+ team_sizes.c.member_count,
+ func.count().label('count')
+ ).group_by(team_sizes.c.member_count).all()
+
+ # Convert to a list of dictionaries for easier handling in the template
+ team_dist_list = [{"member_count": size[0], "count": size[1]} for size in team_distribution]
+
+ stats["team_distribution"] = team_dist_list
+
+ return stats
+
+# Get event statistics for the dashboard
+def get_event_statistics(db: Session) -> Dict[str, Any]:
+ """Get event statistics for the admin dashboard."""
+ stats = {}
+
+ # Total events
+ stats["total_events"] = db.query(Event).count()
+
+ # Past events
+ today = datetime.now().date()
+ stats["past_events"] = db.query(Event).filter(Event.event_date < today).count()
+
+ # Upcoming events
+ stats["upcoming_events_count"] = db.query(Event).filter(Event.event_date >= today).count()
+
+ # List of upcoming events
+ upcoming_events = db.query(Event).filter(
+ Event.event_date >= today
+ ).order_by(Event.event_date).limit(5).all()
+
+ stats["upcoming_events"] = upcoming_events
+
+ # Events with highest attendance
+ attendance_rates = db.query(
+ Event.id.label('event_id'),
+ Event.name.label('event_name'),
+ func.count(EventAttendee.id).label('attendee_count')
+ ).join(EventAttendee).group_by(Event.id, Event.name).order_by(
+ desc('attendee_count')
+ ).limit(5).all()
+
+ stats["attendance_rates"] = attendance_rates
+
+ return stats
+
+# Get system health information
+def get_system_health(db: Session) -> Dict[str, Any]:
+ """Get system health information for the admin dashboard."""
+ health_info = {}
+
+ # Database status
+ try:
+ result = db.execute(text("SELECT 'online' as status")).fetchall()
+ health_info["database_status"] = "online" if result else "offline"
+ except Exception as e:
+ health_info["database_status"] = "error"
+ health_info["database_error"] = str(e)
+
+ # System uptime
+ try:
+ uptime_seconds = time.time() - psutil.boot_time()
+ days, remainder = divmod(uptime_seconds, 86400)
+ hours, remainder = divmod(remainder, 3600)
+ minutes, seconds = divmod(remainder, 60)
+ health_info["uptime"] = f"{int(days)} days, {int(hours)} hours, {int(minutes)} minutes"
+ except Exception:
+ health_info["uptime"] = "Unknown"
+
+ # Recent errors (would be fetched from a logging system in production)
+ # For this example, we'll return a placeholder
+ health_info["recent_errors"] = []
+
+ return health_info
+
def get_model_info(model_class: Type[Base]) -> Dict[str, Dict[str, Any]]:
"""Get column information for a model."""
mapper = inspect(model_class)
@@ -78,6 +245,28 @@ def get_relationships(model_class: Type[Base]) -> Dict[str, str]:
@require_admin(redirect_url="/auth/login?next=/admin/")
async def admin_home(request: Request, db: Session = Depends(get_db)):
"""Admin dashboard home."""
+ # Get statistics
+ user_stats = get_user_statistics(db)
+ team_stats = get_team_statistics(db)
+ event_stats = get_event_statistics(db)
+ system_health = get_system_health(db)
+
+ return templates.TemplateResponse(
+ "admin/dashboard.html",
+ {
+ "request": request,
+ "user": request.user,
+ "user_stats": user_stats,
+ "team_stats": team_stats,
+ "event_stats": event_stats,
+ "system_health": system_health
+ }
+ )
+
+@router.get("/models", response_class=HTMLResponse)
+@require_admin(redirect_url="/auth/login?next=/admin/models")
+async def admin_models(request: Request, db: Session = Depends(get_db)):
+ """Admin models overview."""
# Since we're using Starlette's authentication, the user is now available in request.user
model_list = [(key, name) for key, (_, name) in MODELS.items()]
return templates.TemplateResponse(
diff --git a/requirements.txt b/requirements.txt
index eda161a..3835f65 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -30,6 +30,8 @@ python-dotenv>=1.0.0
email-validator>=2.0.0
pydantic>=2.3.0
qrcode>=7.4.2
+psutil>=5.9.0 # System monitoring and statistics
+python-dateutil>=2.8.2 # Date manipulation utilities
# Email support
fastapi-mail>=1.4.2
diff --git a/tests/unit/views/test_admin_dashboard.py b/tests/unit/views/test_admin_dashboard.py
new file mode 100644
index 0000000..1396ad5
--- /dev/null
+++ b/tests/unit/views/test_admin_dashboard.py
@@ -0,0 +1,170 @@
+#!/usr/bin/env python3
+import pytest
+from unittest import mock
+from sqlalchemy.orm import Session
+from fastapi.testclient import TestClient
+from datetime import datetime, timedelta
+
+from app.main import app
+from app.models import User, Team, Event, QRCode
+from app.views.admin import get_user_statistics, get_team_statistics
+from app.views.admin import get_event_statistics, get_system_health
+
+# Fixture for mocking the database session
+@pytest.fixture
+def mock_db():
+ """Create a mock database session for testing."""
+ mock_session = mock.MagicMock(spec=Session)
+ return mock_session
+
+# Test cases for user statistics
+def test_get_user_statistics(mock_db):
+ """Test getting user statistics."""
+ # Setup mock query results
+ mock_db.query().count.side_effect = [100, 80, 20]
+ mock_db.query().filter().count.return_value = 10
+
+ # Get last 30 days
+ thirty_days_ago = datetime.now() - timedelta(days=30)
+ mock_db.query().filter().filter().count.return_value = 15
+
+ # Get the statistics
+ stats = get_user_statistics(mock_db)
+
+ # Assert the expected results
+ assert stats["total_users"] == 100
+ assert stats["active_users"] == 80
+ assert stats["verified_users"] == 20
+ assert stats["admin_users"] == 10
+ assert stats["new_registrations_30d"] == 15
+
+# Test cases for team statistics
+def test_get_team_statistics(mock_db):
+ """Test getting team statistics."""
+ # Setup mock query results
+ mock_db.query().count.side_effect = [50, 45]
+ mock_db.query().filter().count.return_value = 5
+
+ # Team distribution mock
+ mock_team_distribution = [
+ {"member_count": 0, "count": 5},
+ {"member_count": 1, "count": 10},
+ {"member_count": 2, "count": 15},
+ {"member_count": 3, "count": 10},
+ {"member_count": 4, "count": 7},
+ {"member_count": 5, "count": 3}
+ ]
+ mock_db.query().group_by().all.return_value = mock_team_distribution
+
+ # Get the statistics
+ stats = get_team_statistics(mock_db)
+
+ # Assert the expected results
+ assert stats["total_teams"] == 50
+ assert stats["active_teams"] == 45
+ assert stats["public_teams"] == 5
+ assert stats["team_distribution"] == mock_team_distribution
+
+# Test cases for event statistics
+def test_get_event_statistics(mock_db):
+ """Test getting event statistics."""
+ # Setup mock query results
+ mock_db.query().count.side_effect = [30, 25, 5]
+
+ # Setup mock for upcoming events
+ today = datetime.now().date()
+ upcoming_events = [
+ mock.MagicMock(name="Event 1", event_date=today + timedelta(days=1), location="Location 1"),
+ mock.MagicMock(name="Event 2", event_date=today + timedelta(days=3), location="Location 2"),
+ mock.MagicMock(name="Event 3", event_date=today + timedelta(days=7), location="Location 3")
+ ]
+ mock_db.query().filter().order_by().limit().all.return_value = upcoming_events
+
+ # Setup mock for attendance rates
+ mock_attendance_rates = [
+ {"event_id": 1, "event_name": "Event A", "attendee_count": 25},
+ {"event_id": 2, "event_name": "Event B", "attendee_count": 18},
+ {"event_id": 3, "event_name": "Event C", "attendee_count": 30}
+ ]
+ mock_db.query().join().group_by().order_by().limit().all.return_value = mock_attendance_rates
+
+ # Get the statistics
+ stats = get_event_statistics(mock_db)
+
+ # Assert the expected results
+ assert stats["total_events"] == 30
+ assert stats["past_events"] == 25
+ assert stats["upcoming_events_count"] == 5
+ assert len(stats["upcoming_events"]) == 3
+ assert stats["attendance_rates"] == mock_attendance_rates
+
+# Test cases for system health
+def test_get_system_health(mock_db):
+ """Test getting system health information."""
+ # Mock database status
+ mock_db.execute().fetchall.return_value = [{"status": "online"}]
+
+ # Get the health information
+ health_info = get_system_health(mock_db)
+
+ # Assert expected results
+ assert health_info["database_status"] == "online"
+ assert "uptime" in health_info
+ assert "recent_errors" in health_info
+
+# Integration test for admin dashboard endpoint
+@mock.patch("app.views.admin.get_db")
+def test_admin_dashboard_endpoint(mock_get_db, mock_db):
+ """Test the admin dashboard endpoint."""
+ # Setup mock DB to be returned from get_db
+ mock_get_db.return_value = mock_db
+
+ # Mock user stats
+ mock_user_stats = {
+ "total_users": 100,
+ "active_users": 80,
+ "verified_users": 20,
+ "admin_users": 10,
+ "new_registrations_30d": 15
+ }
+
+ # Mock team stats
+ mock_team_stats = {
+ "total_teams": 50,
+ "active_teams": 45,
+ "public_teams": 5,
+ "team_distribution": []
+ }
+
+ # Mock event stats
+ mock_event_stats = {
+ "total_events": 30,
+ "past_events": 25,
+ "upcoming_events_count": 5,
+ "upcoming_events": [],
+ "attendance_rates": []
+ }
+
+ # Mock system health
+ mock_system_health = {
+ "database_status": "online",
+ "uptime": "3 days, 2 hours",
+ "recent_errors": []
+ }
+
+ # Setup mock return values for our statistics functions
+ with mock.patch("app.views.admin.get_user_statistics", return_value=mock_user_stats), \
+ mock.patch("app.views.admin.get_team_statistics", return_value=mock_team_stats), \
+ mock.patch("app.views.admin.get_event_statistics", return_value=mock_event_stats), \
+ mock.patch("app.views.admin.get_system_health", return_value=mock_system_health), \
+ mock.patch("app.views.admin.require_admin", return_value=lambda f: f):
+
+ client = TestClient(app)
+ response = client.get("/admin/dashboard")
+
+ # Assert the response
+ assert response.status_code == 200
+ assert "user_stats" in response.context
+ assert "team_stats" in response.context
+ assert "event_stats" in response.context
+ assert "system_health" in response.context
\ No newline at end of file