egg/PROJECT-CONTORL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement workload heatmap module

Browse Source 
- Backend (FastAPI):
  - Workload heatmap API with load level calculation
  - User workload detail endpoint with task breakdown
  - Redis caching for workload calculations (1hr TTL)
  - Department isolation and access control
  - WorkloadSnapshot model for historical data
  - Alembic migration for workload_snapshots table

- API Endpoints:
  - GET /api/workload/heatmap - Team workload overview
  - GET /api/workload/user/{id} - User workload detail
  - GET /api/workload/me - Current user workload

- Load Levels:
  - normal: <80%, warning: 80-99%, overloaded: >=100%

- Tests:
  - 26 unit/API tests
  - 15 E2E automated tests
  - 77 total tests passing

- OpenSpec:
  - add-resource-workload change archived
  - resource-management spec updated

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
beabigegg
2025-12-29 01:13:21 +08:00
parent daca7798e3
commit 61fe01cb6b
17 changed files with 2517 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

537
backend/tests/test_workload.py Normal file
View File

@@ -0,0 +1,537 @@
"""Tests for workload API and service."""
import pytest
from datetime import date, datetime, timedelta
from decimal import Decimal
from app.models import User, Department, Space, Project, Task
from app.models.task_status import TaskStatus
from app.services.workload_service import (
get_week_bounds,
get_current_week_start,
determine_load_level,
calculate_load_percentage,
calculate_user_workload,
get_workload_heatmap,
get_user_workload_detail,
)
from app.schemas.workload import LoadLevel
class TestWeekBounds:
"""Tests for week boundary calculations."""
def test_get_week_bounds_monday(self):
"""Monday should return same day as week start."""
monday = date(2024, 1, 1) # This is a Monday
week_start, week_end = get_week_bounds(monday)
assert week_start == monday
assert week_end == date(2024, 1, 7)
def test_get_week_bounds_wednesday(self):
"""Wednesday should return previous Monday as week start."""
wednesday = date(2024, 1, 3)
week_start, week_end = get_week_bounds(wednesday)
assert week_start == date(2024, 1, 1)
assert week_end == date(2024, 1, 7)
def test_get_week_bounds_sunday(self):
"""Sunday should return previous Monday as week start."""
sunday = date(2024, 1, 7)
week_start, week_end = get_week_bounds(sunday)
assert week_start == date(2024, 1, 1)
assert week_end == date(2024, 1, 7)
def test_get_current_week_start(self):
"""Current week start should be a Monday."""
week_start = get_current_week_start()
# Monday = 0
assert week_start.weekday() == 0
class TestLoadLevel:
"""Tests for load level determination."""
def test_load_level_normal(self):
"""Load below 80% should be normal."""
assert determine_load_level(Decimal("0")) == LoadLevel.NORMAL
assert determine_load_level(Decimal("50")) == LoadLevel.NORMAL
assert determine_load_level(Decimal("79.99")) == LoadLevel.NORMAL
def test_load_level_warning(self):
"""Load 80-99% should be warning."""
assert determine_load_level(Decimal("80")) == LoadLevel.WARNING
assert determine_load_level(Decimal("90")) == LoadLevel.WARNING
assert determine_load_level(Decimal("99.99")) == LoadLevel.WARNING
def test_load_level_overloaded(self):
"""Load 100%+ should be overloaded."""
assert determine_load_level(Decimal("100")) == LoadLevel.OVERLOADED
assert determine_load_level(Decimal("150")) == LoadLevel.OVERLOADED
def test_load_level_unavailable(self):
"""None percentage should be unavailable."""
assert determine_load_level(None) == LoadLevel.UNAVAILABLE
class TestLoadPercentage:
"""Tests for load percentage calculation."""
def test_normal_calculation(self):
"""Normal calculation should work."""
result = calculate_load_percentage(Decimal("32"), Decimal("40"))
assert result == Decimal("80.00")
def test_zero_capacity(self):
"""Zero capacity should return None."""
result = calculate_load_percentage(Decimal("32"), Decimal("0"))
assert result is None
def test_zero_allocated(self):
"""Zero allocated should return 0."""
result = calculate_load_percentage(Decimal("0"), Decimal("40"))
assert result == Decimal("0.00")
class TestWorkloadService:
"""Tests for workload service with database."""
def setup_test_data(self, db):
"""Set up test data for workload tests."""
# Create department
dept = Department(
id="dept-001",
name="R&D",
)
db.add(dept)
# Create engineer user
engineer = User(
id="user-engineer-001",
email="engineer@test.com",
name="Test Engineer",
department_id="dept-001",
role_id="00000000-0000-0000-0000-000000000003",
capacity=40,
is_active=True,
is_system_admin=False,
)
db.add(engineer)
# Create space
space = Space(
id="space-001",
name="Test Space",
owner_id="00000000-0000-0000-0000-000000000001",
is_active=True,
)
db.add(space)
# Create project
project = Project(
id="project-001",
space_id="space-001",
title="Test Project",
owner_id="00000000-0000-0000-0000-000000000001",
department_id="dept-001",
security_level="department",
)
db.add(project)
# Create task status (not done)
status_todo = TaskStatus(
id="status-todo",
project_id="project-001",
name="To Do",
is_done=False,
)
db.add(status_todo)
status_done = TaskStatus(
id="status-done",
project_id="project-001",
name="Done",
is_done=True,
)
db.add(status_done)
db.commit()
return {
"department": dept,
"engineer": engineer,
"space": space,
"project": project,
"status_todo": status_todo,
"status_done": status_done,
}
def create_task(self, db, data, task_id, estimate, due_date, status_id=None, done=False):
"""Helper to create a task."""
task = Task(
id=task_id,
project_id=data["project"].id,
title=f"Task {task_id}",
assignee_id=data["engineer"].id,
status_id=status_id or (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",
)
db.add(task)
db.commit()
return task
def test_calculate_user_workload_empty(self, db):
"""User with no tasks should have 0 allocated hours."""
data = self.setup_test_data(db)
week_start = date(2024, 1, 1)
summary = calculate_user_workload(db, data["engineer"], week_start)
assert summary.user_id == data["engineer"].id
assert summary.allocated_hours == Decimal("0")
assert summary.capacity_hours == Decimal("40")
assert summary.load_percentage == Decimal("0.00")
assert summary.load_level == LoadLevel.NORMAL
assert summary.task_count == 0
def test_calculate_user_workload_with_tasks(self, db):
"""User with tasks should have correct allocated hours."""
data = self.setup_test_data(db)
# Create tasks due in the week of 2024-01-01
week_start = date(2024, 1, 1)
due = datetime(2024, 1, 3, 12, 0, 0) # Wednesday
self.create_task(db, data, "task-1", Decimal("8"), due)
self.create_task(db, data, "task-2", Decimal("16"), due)
summary = calculate_user_workload(db, data["engineer"], week_start)
assert summary.allocated_hours == Decimal("24")
assert summary.load_percentage == Decimal("60.00")
assert summary.load_level == LoadLevel.NORMAL
assert summary.task_count == 2
def test_calculate_user_workload_overloaded(self, db):
"""User with too many tasks should be overloaded."""
data = self.setup_test_data(db)
week_start = date(2024, 1, 1)
due = datetime(2024, 1, 3, 12, 0, 0)
# 48 hours > 40 capacity = overloaded
self.create_task(db, data, "task-1", Decimal("24"), due)
self.create_task(db, data, "task-2", Decimal("24"), due)
summary = calculate_user_workload(db, data["engineer"], week_start)
assert summary.allocated_hours == Decimal("48")
assert summary.load_percentage == Decimal("120.00")
assert summary.load_level == LoadLevel.OVERLOADED
def test_completed_tasks_excluded(self, db):
"""Completed tasks should not count toward workload."""
data = self.setup_test_data(db)
week_start = date(2024, 1, 1)
due = datetime(2024, 1, 3, 12, 0, 0)
self.create_task(db, data, "task-1", Decimal("8"), due, done=False)
self.create_task(db, data, "task-2", Decimal("16"), due, done=True) # Done
summary = calculate_user_workload(db, data["engineer"], week_start)
assert summary.allocated_hours == Decimal("8") # Only uncompleted task
assert summary.task_count == 1
def test_tasks_outside_week_excluded(self, db):
"""Tasks due outside the week should not count."""
data = self.setup_test_data(db)
week_start = date(2024, 1, 1)
# Task due in this week
self.create_task(db, data, "task-1", Decimal("8"), datetime(2024, 1, 3, 12, 0, 0))
# Task due next week
self.create_task(db, data, "task-2", Decimal("16"), datetime(2024, 1, 10, 12, 0, 0))
summary = calculate_user_workload(db, data["engineer"], week_start)
assert summary.allocated_hours == Decimal("8") # Only this week's task
assert summary.task_count == 1
def test_get_workload_heatmap(self, db):
"""Heatmap should return all matching users."""
data = self.setup_test_data(db)
week_start = date(2024, 1, 1)
due = datetime(2024, 1, 3, 12, 0, 0)
self.create_task(db, data, "task-1", Decimal("32"), due)
# Get heatmap for the department
summaries = get_workload_heatmap(
db=db,
week_start=week_start,
department_id="dept-001",
)
# Should include engineer (not admin, admin has no department)
assert len(summaries) == 1
assert summaries[0].user_id == data["engineer"].id
assert summaries[0].load_level == LoadLevel.WARNING # 80%
def test_get_user_workload_detail(self, db):
"""Detail should include task list."""
data = self.setup_test_data(db)
week_start = date(2024, 1, 1)
due = datetime(2024, 1, 3, 12, 0, 0)
self.create_task(db, data, "task-1", Decimal("8"), due)
self.create_task(db, data, "task-2", Decimal("16"), due)
detail = get_user_workload_detail(db, data["engineer"].id, week_start)
assert detail is not None
assert detail.user_id == data["engineer"].id
assert len(detail.tasks) == 2
assert detail.allocated_hours == Decimal("24")
class TestWorkloadAPI:
"""Tests for workload API endpoints."""
def setup_test_data(self, db):
"""Set up test data for API tests."""
# Create department
dept = Department(
id="dept-001",
name="R&D",
)
db.add(dept)
# Create engineer user
engineer = User(
id="user-engineer-001",
email="engineer@test.com",
name="Test Engineer",
department_id="dept-001",
role_id="00000000-0000-0000-0000-000000000003",
capacity=40,
is_active=True,
is_system_admin=False,
)
db.add(engineer)
# Create space
space = Space(
id="space-001",
name="Test Space",
owner_id="00000000-0000-0000-0000-000000000001",
is_active=True,
)
db.add(space)
# Create project
project = Project(
id="project-001",
space_id="space-001",
title="Test Project",
owner_id="00000000-0000-0000-0000-000000000001",
department_id="dept-001",
security_level="department",
)
db.add(project)
# Create task status
status_todo = TaskStatus(
id="status-todo",
project_id="project-001",
name="To Do",
is_done=False,
)
db.add(status_todo)
# Create a task due this week
task = Task(
id="task-001",
project_id="project-001",
title="Test Task",
assignee_id="user-engineer-001",
status_id="status-todo",
original_estimate=Decimal("32"),
due_date=datetime.now() + timedelta(days=1),
created_by="00000000-0000-0000-0000-000000000001",
)
db.add(task)
db.commit()
return {
"department": dept,
"engineer": engineer,
}
def test_heatmap_as_admin(self, client, db, admin_token):
"""Admin should see all users in heatmap."""
data = self.setup_test_data(db)
response = client.get(
"/api/workload/heatmap",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == 200
result = response.json()
assert "week_start" in result
assert "week_end" in result
assert "users" in result
# Admin sees all users including the engineer
assert len(result["users"]) >= 1
def test_heatmap_with_department_filter(self, client, db, admin_token):
"""Admin can filter by department."""
data = self.setup_test_data(db)
response = client.get(
"/api/workload/heatmap?department_id=dept-001",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == 200
result = response.json()
# Should only include users in dept-001
for user in result["users"]:
assert user["department_id"] == "dept-001"
def test_my_workload(self, client, db, admin_token):
"""User can get their own workload."""
response = client.get(
"/api/workload/me",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == 200
result = response.json()
assert result["user_id"] == "00000000-0000-0000-0000-000000000001"
assert "tasks" in result
def test_user_workload_detail(self, client, db, admin_token):
"""Admin can get any user's workload detail."""
data = self.setup_test_data(db)
response = client.get(
f"/api/workload/user/{data['engineer'].id}",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert response.status_code == 200
result = response.json()
assert result["user_id"] == data["engineer"].id
assert len(result["tasks"]) == 1
assert result["allocated_hours"] == "32.00" # Decimal comes as string with precision
def test_unauthorized_access(self, client, db):
"""Unauthenticated requests should fail."""
response = client.get("/api/workload/heatmap")
assert response.status_code == 403 # No auth header
class TestWorkloadAccessControl:
"""Tests for workload access control."""
def setup_test_data(self, db, mock_redis):
"""Set up test data with two departments."""
from app.core.security import create_access_token, create_token_payload
# Create departments
dept_rd = Department(id="dept-rd", name="R&D")
dept_ops = Department(id="dept-ops", name="Operations")
db.add(dept_rd)
db.add(dept_ops)
# Create engineer in R&D
engineer_rd = User(
id="user-rd-001",
email="rd@test.com",
name="R&D Engineer",
department_id="dept-rd",
role_id="00000000-0000-0000-0000-000000000003",
capacity=40,
is_active=True,
is_system_admin=False,
)
db.add(engineer_rd)
# Create engineer in Operations
engineer_ops = User(
id="user-ops-001",
email="ops@test.com",
name="Ops Engineer",
department_id="dept-ops",
role_id="00000000-0000-0000-0000-000000000003",
capacity=40,
is_active=True,
is_system_admin=False,
)
db.add(engineer_ops)
db.commit()
# Create token for R&D engineer
token_data = create_token_payload(
user_id="user-rd-001",
email="rd@test.com",
role="engineer",
department_id="dept-rd",
is_system_admin=False,
)
rd_token = create_access_token(token_data)
mock_redis.setex("session:user-rd-001", 900, rd_token)
return {
"dept_rd": dept_rd,
"dept_ops": dept_ops,
"engineer_rd": engineer_rd,
"engineer_ops": engineer_ops,
"rd_token": rd_token,
}
def test_regular_user_sees_only_self(self, client, db, mock_redis):
"""Regular user should only see their own workload."""
data = self.setup_test_data(db, mock_redis)
response = client.get(
"/api/workload/heatmap",
headers={"Authorization": f"Bearer {data['rd_token']}"},
)
assert response.status_code == 200
result = response.json()
# Should only see themselves
assert len(result["users"]) == 1
assert result["users"][0]["user_id"] == "user-rd-001"
def test_regular_user_cannot_access_other_department(self, client, db, mock_redis):
"""Regular user should not access other department's workload."""
data = self.setup_test_data(db, mock_redis)
response = client.get(
"/api/workload/heatmap?department_id=dept-ops",
headers={"Authorization": f"Bearer {data['rd_token']}"},
)
assert response.status_code == 403
def test_regular_user_cannot_access_other_user_detail(self, client, db, mock_redis):
"""Regular user should not access other user's detail."""
data = self.setup_test_data(db, mock_redis)
response = client.get(
f"/api/workload/user/{data['engineer_ops'].id}",
headers={"Authorization": f"Bearer {data['rd_token']}"},
)
assert response.status_code == 403