egg/PROJECT-CONTORL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
35c90fe76b3d383ec65f7ff6d15beb4376345ad5
PROJECT-CONTORL/backend/tests/test_dashboard.py
beabigegg 35c90fe76b feat: implement 5 QA-driven security and quality proposals 
Implemented proposals from comprehensive QA review:

1. extend-csrf-protection
   - Add POST to CSRF protected methods in frontend
   - Global CSRF middleware for all state-changing operations
   - Update tests with CSRF token fixtures

2. tighten-cors-websocket-security
   - Replace wildcard CORS with explicit method/header lists
   - Disable query parameter auth in production (code 4002)
   - Add per-user WebSocket connection limit (max 5, code 4005)

3. shorten-jwt-expiry
   - Reduce JWT expiry from 7 days to 60 minutes
   - Add refresh token support with 7-day expiry
   - Implement token rotation on refresh
   - Frontend auto-refresh when token near expiry (<5 min)

4. fix-frontend-quality
   - Add React.lazy() code splitting for all pages
   - Fix useCallback dependency arrays (Dashboard, Comments)
   - Add localStorage data validation in AuthContext
   - Complete i18n for AttachmentUpload component

5. enhance-backend-validation
   - Add SecurityAuditMiddleware for access denied logging
   - Add ErrorSanitizerMiddleware for production error messages
   - Protect /health/detailed with admin authentication
   - Add input length validation (comment 5000, desc 10000)

All 521 backend tests passing. Frontend builds successfully.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-12 23:19:05 +08:00

653 lines
20 KiB
Python
Raw Blame History

"""Tests for dashboard API and service functions."""
import pytest
from datetime import datetime, timedelta
from decimal import Decimal
from app.models import User, Department, Space, Project, Task
from app.models.task_status import TaskStatus
from app.api.dashboard.router import (
get_task_statistics,
get_workload_summary,
get_health_summary,
)
from app.services.workload_service import get_week_bounds
from app.schemas.workload import LoadLevel
class TestTaskStatistics:
"""Tests for task statistics calculation."""
def setup_test_data(self, db):
"""Set up test data for task statistics tests."""
# Create department
dept = Department(
id="dept-dash-001",
name="Dashboard Test Department",
)
db.add(dept)
# Create test user
user = User(
id="user-dash-001",
email="dashboard@test.com",
name="Dashboard Test User",
department_id="dept-dash-001",
role_id="00000000-0000-0000-0000-000000000003",
capacity=40,
is_active=True,
is_system_admin=False,
)
db.add(user)
# Create space
space = Space(
id="space-dash-001",
name="Dashboard Test Space",
owner_id="00000000-0000-0000-0000-000000000001",
is_active=True,
)
db.add(space)
# Create project
project = Project(
id="project-dash-001",
space_id="space-dash-001",
title="Dashboard Test Project",
owner_id="00000000-0000-0000-0000-000000000001",
department_id="dept-dash-001",
security_level="department",
status="active",
)
db.add(project)
# Create task statuses
status_todo = TaskStatus(
id="status-dash-todo",
project_id="project-dash-001",
name="To Do",
is_done=False,
)
db.add(status_todo)
status_done = TaskStatus(
id="status-dash-done",
project_id="project-dash-001",
name="Done",
is_done=True,
)
db.add(status_done)
db.commit()
return {
"department": dept,
"user": user,
"space": space,
"project": project,
"status_todo": status_todo,
"status_done": status_done,
}
def create_task(
self,
db,
data,
task_id,
done=False,
overdue=False,
due_this_week=False,
estimate=None,
):
"""Helper to create a task with optional characteristics."""
now = datetime.utcnow()
week_start, week_end = get_week_bounds(now.date())
if overdue:
due_date = datetime.combine(week_start, datetime.min.time()) - timedelta(days=1)
elif due_this_week:
# Due in the middle of current week
due_date = datetime.combine(week_start, datetime.min.time()) + timedelta(days=2)
else:
# Due next week
due_date = datetime.combine(week_end, datetime.min.time()) + timedelta(days=2)
task = Task(
id=task_id,
project_id=data["project"].id,
title=f"Task {task_id}",
assignee_id=data["user"].id,
status_id=data["status_done"].id if done else data["status_todo"].id,
original_estimate=estimate,
due_date=due_date,
created_by="00000000-0000-0000-0000-000000000001",
is_deleted=False,
)
db.add(task)
db.commit()
return task
def test_empty_statistics(self, db):
"""User with no tasks should have zero counts."""
data = self.setup_test_data(db)
stats = get_task_statistics(db, data["user"].id)
assert stats.assigned_count == 0
assert stats.due_this_week == 0
assert stats.overdue_count == 0
assert stats.completion_rate == 0.0
def test_assigned_count(self, db):
"""Should count non-completed tasks assigned to user."""
data = self.setup_test_data(db)
# Create 3 tasks: 2 active, 1 completed
self.create_task(db, data, "task-1", done=False)
self.create_task(db, data, "task-2", done=False)
self.create_task(db, data, "task-3", done=True)
stats = get_task_statistics(db, data["user"].id)
assert stats.assigned_count == 2 # Only non-completed
def test_due_this_week_count(self, db):
"""Should count tasks due this week."""
data = self.setup_test_data(db)
# Create tasks with different due dates
self.create_task(db, data, "task-1", due_this_week=True)
self.create_task(db, data, "task-2", due_this_week=True)
self.create_task(db, data, "task-3", due_this_week=False) # Next week
stats = get_task_statistics(db, data["user"].id)
assert stats.due_this_week == 2
def test_overdue_count(self, db):
"""Should count overdue tasks."""
data = self.setup_test_data(db)
# Create overdue and non-overdue tasks
self.create_task(db, data, "task-1", overdue=True)
self.create_task(db, data, "task-2", overdue=True)
self.create_task(db, data, "task-3", overdue=False)
stats = get_task_statistics(db, data["user"].id)
assert stats.overdue_count == 2
def test_overdue_completed_not_counted(self, db):
"""Completed overdue tasks should not be counted as overdue."""
data = self.setup_test_data(db)
# Create overdue task that is completed
self.create_task(db, data, "task-1", overdue=True, done=True)
self.create_task(db, data, "task-2", overdue=True, done=False)
stats = get_task_statistics(db, data["user"].id)
assert stats.overdue_count == 1
def test_completion_rate(self, db):
"""Should calculate correct completion rate."""
data = self.setup_test_data(db)
# Create 4 tasks: 1 completed, 3 active = 25%
self.create_task(db, data, "task-1", done=True)
self.create_task(db, data, "task-2", done=False)
self.create_task(db, data, "task-3", done=False)
self.create_task(db, data, "task-4", done=False)
stats = get_task_statistics(db, data["user"].id)
assert stats.completion_rate == 25.0
def test_deleted_tasks_excluded(self, db):
"""Soft-deleted tasks should not be counted."""
data = self.setup_test_data(db)
# Create normal task
self.create_task(db, data, "task-1")
# Create deleted task
deleted_task = Task(
id="task-deleted",
project_id=data["project"].id,
title="Deleted Task",
assignee_id=data["user"].id,
status_id=data["status_todo"].id,
due_date=datetime.utcnow() - timedelta(days=5), # Overdue
created_by="00000000-0000-0000-0000-000000000001",
is_deleted=True,
deleted_at=datetime.utcnow(),
)
db.add(deleted_task)
db.commit()
stats = get_task_statistics(db, data["user"].id)
assert stats.assigned_count == 1
assert stats.overdue_count == 0 # Deleted task not counted
class TestWorkloadSummary:
"""Tests for workload summary calculation."""
def setup_test_data(self, db):
"""Set up test data for workload summary tests."""
# Create department
dept = Department(
id="dept-wl-001",
name="Workload Test Department",
)
db.add(dept)
# Create test user with 40h capacity
user = User(
id="user-wl-001",
email="workload@test.com",
name="Workload Test User",
department_id="dept-wl-001",
role_id="00000000-0000-0000-0000-000000000003",
capacity=40,
is_active=True,
is_system_admin=False,
)
db.add(user)
# Create space
space = Space(
id="space-wl-001",
name="Workload Test Space",
owner_id="00000000-0000-0000-0000-000000000001",
is_active=True,
)
db.add(space)
# Create project
project = Project(
id="project-wl-001",
space_id="space-wl-001",
title="Workload Test Project",
owner_id="00000000-0000-0000-0000-000000000001",
department_id="dept-wl-001",
security_level="department",
status="active",
)
db.add(project)
# Create task status
status_todo = TaskStatus(
id="status-wl-todo",
project_id="project-wl-001",
name="To Do",
is_done=False,
)
db.add(status_todo)
status_done = TaskStatus(
id="status-wl-done",
project_id="project-wl-001",
name="Done",
is_done=True,
)
db.add(status_done)
db.commit()
return {
"department": dept,
"user": user,
"space": space,
"project": project,
"status_todo": status_todo,
"status_done": status_done,
}
def test_empty_workload(self, db):
"""User with no tasks should have zero allocated hours."""
data = self.setup_test_data(db)
workload = get_workload_summary(db, data["user"])
assert workload.allocated_hours == Decimal("0")
assert workload.capacity_hours == Decimal("40")
assert workload.load_percentage == Decimal("0.00")
assert workload.load_level == LoadLevel.NORMAL
def test_zero_capacity(self, db):
"""User with zero capacity should show unavailable load level."""
data = self.setup_test_data(db)
data["user"].capacity = 0
db.commit()
workload = get_workload_summary(db, data["user"])
assert workload.capacity_hours == Decimal("0")
assert workload.load_percentage is None
assert workload.load_level == LoadLevel.UNAVAILABLE
def test_workload_with_tasks(self, db):
"""Should calculate correct allocated hours."""
data = self.setup_test_data(db)
# Create tasks due this week with estimates
now = datetime.utcnow()
week_start, _ = get_week_bounds(now.date())
due_date = datetime.combine(week_start, datetime.min.time()) + timedelta(days=2)
task1 = Task(
id="task-wl-1",
project_id=data["project"].id,
title="Task 1",
assignee_id=data["user"].id,
status_id=data["status_todo"].id,
original_estimate=Decimal("16"),
due_date=due_date,
created_by="00000000-0000-0000-0000-000000000001",
is_deleted=False,
)
db.add(task1)
task2 = Task(
id="task-wl-2",
project_id=data["project"].id,
title="Task 2",
assignee_id=data["user"].id,
status_id=data["status_todo"].id,
original_estimate=Decimal("16"),
due_date=due_date,
created_by="00000000-0000-0000-0000-000000000001",
is_deleted=False,
)
db.add(task2)
db.commit()
workload = get_workload_summary(db, data["user"])
assert workload.allocated_hours == Decimal("32")
assert workload.load_percentage == Decimal("80.00")
assert workload.load_level == LoadLevel.WARNING
def test_workload_overloaded(self, db):
"""User with more than capacity should be overloaded."""
data = self.setup_test_data(db)
now = datetime.utcnow()
week_start, _ = get_week_bounds(now.date())
due_date = datetime.combine(week_start, datetime.min.time()) + timedelta(days=2)
# Create task with 48h estimate (> 40h capacity)
task = Task(
id="task-wl-over",
project_id=data["project"].id,
title="Big Task",
assignee_id=data["user"].id,
status_id=data["status_todo"].id,
original_estimate=Decimal("48"),
due_date=due_date,
created_by="00000000-0000-0000-0000-000000000001",
is_deleted=False,
)
db.add(task)
db.commit()
workload = get_workload_summary(db, data["user"])
assert workload.allocated_hours == Decimal("48")
assert workload.load_percentage == Decimal("120.00")
assert workload.load_level == LoadLevel.OVERLOADED
def test_completed_tasks_excluded(self, db):
"""Completed tasks should not count toward workload."""
data = self.setup_test_data(db)
now = datetime.utcnow()
due_date = now + timedelta(days=2)
# Create completed task
task = Task(
id="task-wl-done",
project_id=data["project"].id,
title="Done Task",
assignee_id=data["user"].id,
status_id=data["status_done"].id,
original_estimate=Decimal("24"),
due_date=due_date,
created_by="00000000-0000-0000-0000-000000000001",
is_deleted=False,
)
db.add(task)
db.commit()
workload = get_workload_summary(db, data["user"])
assert workload.allocated_hours == Decimal("0")
class TestHealthSummary:
"""Tests for health summary aggregation."""
def setup_test_data(self, db):
"""Set up test data for health summary tests."""
# Create department
dept = Department(
id="dept-hs-001",
name="Health Summary Test Department",
)
db.add(dept)
# Create space
space = Space(
id="space-hs-001",
name="Health Summary Test Space",
owner_id="00000000-0000-0000-0000-000000000001",
is_active=True,
)
db.add(space)
# Create active project
project = Project(
id="project-hs-001",
space_id="space-hs-001",
title="Health Test Project",
owner_id="00000000-0000-0000-0000-000000000001",
department_id="dept-hs-001",
security_level="department",
status="active",
)
db.add(project)
db.commit()
return {
"department": dept,
"space": space,
"project": project,
}
def test_health_summary_structure(self, db):
"""Health summary should have correct structure."""
data = self.setup_test_data(db)
summary = get_health_summary(db)
assert summary.total_projects >= 1
assert summary.healthy_count >= 0
assert summary.at_risk_count >= 0
assert summary.critical_count >= 0
assert summary.average_health_score >= 0
assert summary.average_health_score <= 100
class TestDashboardAPI:
"""Tests for dashboard API endpoint."""
def setup_test_data(self, db):
"""Set up test data for dashboard API tests."""
# Create department
dept = Department(
id="dept-api-dash-001",
name="API Dashboard Test Department",
)
db.add(dept)
# Create space
space = Space(
id="space-api-dash-001",
name="API Dashboard Test Space",
owner_id="00000000-0000-0000-0000-000000000001",
is_active=True,
)
db.add(space)
# Create project
project = Project(
id="project-api-dash-001",
space_id="space-api-dash-001",
title="API Dashboard Test Project",
owner_id="00000000-0000-0000-0000-000000000001",
department_id="dept-api-dash-001",
security_level="department",
status="active",
)
db.add(project)
# Create task status
status_todo = TaskStatus(
id="status-api-dash-todo",
project_id="project-api-dash-001",
name="To Do",
is_done=False,
)
db.add(status_todo)
# Create a task for the admin user
now = datetime.utcnow()
week_start, _ = get_week_bounds(now.date())
task = Task(
id="task-api-dash-001",
project_id="project-api-dash-001",
title="Admin Task",
assignee_id="00000000-0000-0000-0000-000000000001",
status_id="status-api-dash-todo",
original_estimate=Decimal("8"),
due_date=datetime.combine(week_start, datetime.min.time()) + timedelta(days=2),
created_by="00000000-0000-0000-0000-000000000001",
is_deleted=False,
)
db.add(task)
db.commit()
return {
"department": dept,
"space": space,
"project": project,
"task": task,
}
def test_get_dashboard(self, client, db, admin_token):
"""Should return complete dashboard data."""
data = self.setup_test_data(db)
response = client.get(
"/api/dashboard",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == 200
result = response.json()
# Check structure
assert "task_stats" in result
assert "workload" in result
assert "health_summary" in result
def test_dashboard_task_stats_fields(self, client, db, admin_token):
"""Dashboard task_stats should include all expected fields."""
data = self.setup_test_data(db)
response = client.get(
"/api/dashboard",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == 200
task_stats = response.json()["task_stats"]
assert "assigned_count" in task_stats
assert "due_this_week" in task_stats
assert "overdue_count" in task_stats
assert "completion_rate" in task_stats
def test_dashboard_workload_fields(self, client, db, admin_token):
"""Dashboard workload should include all expected fields."""
data = self.setup_test_data(db)
response = client.get(
"/api/dashboard",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == 200
workload = response.json()["workload"]
assert "allocated_hours" in workload
assert "capacity_hours" in workload
assert "load_percentage" in workload
assert "load_level" in workload
def test_dashboard_health_summary_fields(self, client, db, admin_token):
"""Dashboard health_summary should include all expected fields."""
data = self.setup_test_data(db)
response = client.get(
"/api/dashboard",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == 200
health_summary = response.json()["health_summary"]
assert "total_projects" in health_summary
assert "healthy_count" in health_summary
assert "at_risk_count" in health_summary
assert "critical_count" in health_summary
assert "average_health_score" in health_summary
def test_dashboard_unauthorized(self, client, db):
"""Unauthenticated requests should fail."""
response = client.get("/api/dashboard")
assert response.status_code == 401 # 401 for unauthenticated, 403 for unauthorized
def test_dashboard_with_user_tasks(self, client, db, admin_token):
"""Dashboard should reflect user's tasks correctly."""
data = self.setup_test_data(db)
response = client.get(
"/api/dashboard",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == 200
result = response.json()
# Admin has 1 task assigned (created in setup)
assert result["task_stats"]["assigned_count"] >= 1
assert result["task_stats"]["due_this_week"] >= 1
def test_dashboard_workload_load_level_values(self, client, db, admin_token):
"""Workload load_level should be a valid enum value."""
data = self.setup_test_data(db)
response = client.get(
"/api/dashboard",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == 200
load_level = response.json()["workload"]["load_level"]
assert load_level in ["normal", "warning", "overloaded", "unavailable"]
Reference in New Issue View Git Blame Copy Permalink