"""Tests for project health API and service.""" import pytest from datetime import datetime, timedelta from decimal import Decimal from app.models import User, Department, Space, Project, Task, Blocker from app.models.task_status import TaskStatus from app.models.project_health import ProjectHealth from app.services.health_service import ( calculate_health_metrics, get_or_create_project_health, update_project_health, get_project_health, get_all_projects_health, HealthService, _determine_risk_level, _determine_schedule_status, _determine_resource_status, BLOCKER_PENALTY_PER_ITEM, BLOCKER_PENALTY_MAX, OVERDUE_PENALTY_PER_ITEM, OVERDUE_PENALTY_MAX, ) from app.schemas.project_health import RiskLevel, ScheduleStatus, ResourceStatus class TestRiskLevelDetermination: """Tests for risk level determination logic.""" def test_low_risk(self): """Health score >= 80 should be low risk.""" assert _determine_risk_level(100) == "low" assert _determine_risk_level(80) == "low" def test_medium_risk(self): """Health score 60-79 should be medium risk.""" assert _determine_risk_level(79) == "medium" assert _determine_risk_level(60) == "medium" def test_high_risk(self): """Health score 40-59 should be high risk.""" assert _determine_risk_level(59) == "high" assert _determine_risk_level(40) == "high" def test_critical_risk(self): """Health score < 40 should be critical risk.""" assert _determine_risk_level(39) == "critical" assert _determine_risk_level(0) == "critical" class TestScheduleStatusDetermination: """Tests for schedule status determination logic.""" def test_on_track(self): """No overdue tasks means on track.""" assert _determine_schedule_status(0) == "on_track" def test_at_risk(self): """1-2 overdue tasks means at risk.""" assert _determine_schedule_status(1) == "at_risk" assert _determine_schedule_status(2) == "at_risk" def test_delayed(self): """More than 2 overdue tasks means delayed.""" assert _determine_schedule_status(3) == "delayed" assert _determine_schedule_status(10) == "delayed" class TestResourceStatusDetermination: """Tests for resource status determination logic.""" def test_adequate(self): """No blockers means adequate resources.""" assert _determine_resource_status(0) == "adequate" def test_constrained(self): """1-2 blockers means constrained resources.""" assert _determine_resource_status(1) == "constrained" assert _determine_resource_status(2) == "constrained" def test_overloaded(self): """More than 2 blockers means overloaded.""" assert _determine_resource_status(3) == "overloaded" assert _determine_resource_status(10) == "overloaded" class TestHealthMetricsCalculation: """Tests for health metrics calculation with database.""" def setup_test_data(self, db): """Set up test data for health tests.""" # Create department dept = Department( id="dept-health-001", name="Health Test Department", ) db.add(dept) # Create space space = Space( id="space-health-001", name="Health Test Space", owner_id="00000000-0000-0000-0000-000000000001", is_active=True, ) db.add(space) # Create project project = Project( id="project-health-001", space_id="space-health-001", title="Health Test Project", owner_id="00000000-0000-0000-0000-000000000001", department_id="dept-health-001", security_level="department", status="active", ) db.add(project) # Create task statuses status_todo = TaskStatus( id="status-health-todo", project_id="project-health-001", name="To Do", is_done=False, ) db.add(status_todo) status_done = TaskStatus( id="status-health-done", project_id="project-health-001", name="Done", is_done=True, ) db.add(status_done) db.commit() return { "department": dept, "space": space, "project": project, "status_todo": status_todo, "status_done": status_done, } def create_task(self, db, data, task_id, done=False, overdue=False, has_blocker=False): """Helper to create a task with optional characteristics.""" due_date = datetime.utcnow() if overdue: due_date = datetime.utcnow() - timedelta(days=3) else: due_date = datetime.utcnow() + timedelta(days=3) task = Task( id=task_id, project_id=data["project"].id, title=f"Task {task_id}", status_id=data["status_done"].id if done else data["status_todo"].id, due_date=due_date, created_by="00000000-0000-0000-0000-000000000001", is_deleted=False, ) db.add(task) db.commit() if has_blocker: blocker = Blocker( id=f"blocker-{task_id}", task_id=task_id, reported_by="00000000-0000-0000-0000-000000000001", reason="Test blocker", resolved_at=None, ) db.add(blocker) db.commit() return task def test_calculate_metrics_no_tasks(self, db): """Project with no tasks should have 100 health score.""" data = self.setup_test_data(db) metrics = calculate_health_metrics(db, data["project"]) assert metrics["health_score"] == 100 assert metrics["risk_level"] == "low" assert metrics["schedule_status"] == "on_track" assert metrics["resource_status"] == "adequate" assert metrics["task_count"] == 0 assert metrics["completed_task_count"] == 0 assert metrics["blocker_count"] == 0 assert metrics["overdue_task_count"] == 0 def test_calculate_metrics_all_completed(self, db): """Project with all completed tasks should have high health score.""" data = self.setup_test_data(db) self.create_task(db, data, "task-c1", done=True) self.create_task(db, data, "task-c2", done=True) self.create_task(db, data, "task-c3", done=True) metrics = calculate_health_metrics(db, data["project"]) assert metrics["health_score"] == 100 assert metrics["task_count"] == 3 assert metrics["completed_task_count"] == 3 assert metrics["overdue_task_count"] == 0 def test_calculate_metrics_with_blockers(self, db): """Blockers should reduce health score.""" data = self.setup_test_data(db) # Create 3 tasks with blockers self.create_task(db, data, "task-b1", has_blocker=True) self.create_task(db, data, "task-b2", has_blocker=True) self.create_task(db, data, "task-b3", has_blocker=True) metrics = calculate_health_metrics(db, data["project"]) # 3 blockers * 10 points = 30 penalty, also low completion penalty expected_blocker_penalty = min(3 * BLOCKER_PENALTY_PER_ITEM, BLOCKER_PENALTY_MAX) assert metrics["blocker_count"] == 3 assert metrics["resource_status"] == "overloaded" assert metrics["health_score"] < 100 def test_calculate_metrics_with_overdue_tasks(self, db): """Overdue tasks should reduce health score.""" data = self.setup_test_data(db) # Create 3 overdue tasks self.create_task(db, data, "task-o1", overdue=True) self.create_task(db, data, "task-o2", overdue=True) self.create_task(db, data, "task-o3", overdue=True) metrics = calculate_health_metrics(db, data["project"]) assert metrics["overdue_task_count"] == 3 assert metrics["schedule_status"] == "delayed" assert metrics["health_score"] < 100 def test_calculate_metrics_overdue_completed_not_counted(self, db): """Completed overdue tasks should not count as overdue.""" data = self.setup_test_data(db) # Create task that is overdue but completed task = Task( id="task-oc1", project_id=data["project"].id, title="Overdue Completed Task", status_id=data["status_done"].id, due_date=datetime.utcnow() - timedelta(days=5), created_by="00000000-0000-0000-0000-000000000001", is_deleted=False, ) db.add(task) db.commit() metrics = calculate_health_metrics(db, data["project"]) assert metrics["overdue_task_count"] == 0 assert metrics["completed_task_count"] == 1 def test_calculate_metrics_deleted_tasks_excluded(self, db): """Soft-deleted tasks should be excluded from calculations.""" data = self.setup_test_data(db) # Create a normal task self.create_task(db, data, "task-normal") # Create a deleted task deleted_task = Task( id="task-deleted", project_id=data["project"].id, title="Deleted Task", 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() metrics = calculate_health_metrics(db, data["project"]) assert metrics["task_count"] == 1 # Only non-deleted task assert metrics["overdue_task_count"] == 0 # Deleted task not counted def test_calculate_metrics_combined_penalties(self, db): """Multiple issues should stack penalties correctly.""" data = self.setup_test_data(db) # Create mixed tasks: 2 overdue with blockers self.create_task(db, data, "task-mix1", overdue=True, has_blocker=True) self.create_task(db, data, "task-mix2", overdue=True, has_blocker=True) metrics = calculate_health_metrics(db, data["project"]) assert metrics["blocker_count"] == 2 assert metrics["overdue_task_count"] == 2 # Should have penalties from both # 2 blockers = 20 penalty, 2 overdue = 10 penalty, plus completion penalty assert metrics["health_score"] < 80 class TestHealthServiceClass: """Tests for HealthService class.""" def setup_test_data(self, db): """Set up test data for health service tests.""" # Create department dept = Department( id="dept-svc-001", name="Service Test Department", ) db.add(dept) # Create space space = Space( id="space-svc-001", name="Service Test Space", owner_id="00000000-0000-0000-0000-000000000001", is_active=True, ) db.add(space) # Create project project = Project( id="project-svc-001", space_id="space-svc-001", title="Service Test Project", owner_id="00000000-0000-0000-0000-000000000001", department_id="dept-svc-001", security_level="department", status="active", ) db.add(project) # Create inactive project inactive_project = Project( id="project-svc-inactive", space_id="space-svc-001", title="Inactive Project", owner_id="00000000-0000-0000-0000-000000000001", department_id="dept-svc-001", security_level="department", status="archived", ) db.add(inactive_project) db.commit() return { "department": dept, "space": space, "project": project, "inactive_project": inactive_project, } def test_get_or_create_health_creates_new(self, db): """Should create new ProjectHealth if none exists.""" data = self.setup_test_data(db) health = get_or_create_project_health(db, data["project"]) db.commit() assert health is not None assert health.project_id == data["project"].id assert health.health_score == 100 # Default def test_get_or_create_health_returns_existing(self, db): """Should return existing ProjectHealth if one exists.""" data = self.setup_test_data(db) # Create initial health record health1 = get_or_create_project_health(db, data["project"]) health1.health_score = 75 db.commit() # Should return same record health2 = get_or_create_project_health(db, data["project"]) assert health2.id == health1.id assert health2.health_score == 75 def test_get_project_health(self, db): """Should return health details for a project.""" data = self.setup_test_data(db) result = get_project_health(db, data["project"].id) assert result is not None assert result.project_id == data["project"].id assert result.project_title == "Service Test Project" assert result.health_score == 100 def test_get_project_health_not_found(self, db): """Should return None for non-existent project.""" data = self.setup_test_data(db) result = get_project_health(db, "non-existent-id") assert result is None def test_get_all_projects_health_active_only(self, db): """Dashboard should only include active projects by default.""" data = self.setup_test_data(db) result = get_all_projects_health(db, status_filter="active") project_ids = [p.project_id for p in result.projects] assert data["project"].id in project_ids assert data["inactive_project"].id not in project_ids def test_get_all_projects_health_summary(self, db): """Dashboard should include correct summary statistics.""" data = self.setup_test_data(db) result = get_all_projects_health(db, status_filter="active") assert result.summary.total_projects >= 1 assert result.summary.average_health_score <= 100 def test_health_service_class_interface(self, db): """HealthService class should provide same functionality.""" data = self.setup_test_data(db) service = HealthService(db) # Test get_project_health health = service.get_project_health(data["project"].id) assert health is not None assert health.project_id == data["project"].id # Test get_dashboard dashboard = service.get_dashboard() assert dashboard.summary.total_projects >= 1 # Test calculate_metrics metrics = service.calculate_metrics(data["project"]) assert "health_score" in metrics assert "risk_level" in metrics class TestHealthAPI: """Tests for health API endpoints.""" def setup_test_data(self, db): """Set up test data for API tests.""" # Create department dept = Department( id="dept-api-001", name="API Test Department", ) db.add(dept) # Create space space = Space( id="space-api-001", name="API Test Space", owner_id="00000000-0000-0000-0000-000000000001", is_active=True, ) db.add(space) # Create projects project1 = Project( id="project-api-001", space_id="space-api-001", title="API Test Project 1", owner_id="00000000-0000-0000-0000-000000000001", department_id="dept-api-001", security_level="department", status="active", ) db.add(project1) project2 = Project( id="project-api-002", space_id="space-api-001", title="API Test Project 2", owner_id="00000000-0000-0000-0000-000000000001", department_id="dept-api-001", security_level="department", status="active", ) db.add(project2) # Create task statuses status_todo = TaskStatus( id="status-api-todo", project_id="project-api-001", name="To Do", is_done=False, ) db.add(status_todo) # Create a task with blocker for project1 task = Task( id="task-api-001", project_id="project-api-001", title="API Test Task", status_id="status-api-todo", due_date=datetime.utcnow() - timedelta(days=2), # Overdue created_by="00000000-0000-0000-0000-000000000001", is_deleted=False, ) db.add(task) blocker = Blocker( id="blocker-api-001", task_id="task-api-001", reported_by="00000000-0000-0000-0000-000000000001", reason="Test blocker", resolved_at=None, ) db.add(blocker) db.commit() return { "department": dept, "space": space, "project1": project1, "project2": project2, "task": task, "blocker": blocker, } def test_get_dashboard(self, client, db, admin_token): """Admin should be able to get health dashboard.""" data = self.setup_test_data(db) response = client.get( "/api/projects/health/dashboard", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 result = response.json() assert "projects" in result assert "summary" in result assert result["summary"]["total_projects"] >= 2 def test_get_dashboard_summary_fields(self, client, db, admin_token): """Dashboard summary should include all expected fields.""" data = self.setup_test_data(db) response = client.get( "/api/projects/health/dashboard", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 summary = response.json()["summary"] assert "total_projects" in summary assert "healthy_count" in summary assert "at_risk_count" in summary assert "critical_count" in summary assert "average_health_score" in summary assert "projects_with_blockers" in summary assert "projects_delayed" in summary def test_get_project_health(self, client, db, admin_token): """Admin should be able to get single project health.""" data = self.setup_test_data(db) response = client.get( f"/api/projects/health/{data['project1'].id}", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 result = response.json() assert result["project_id"] == data["project1"].id assert result["project_title"] == "API Test Project 1" assert "health_score" in result assert "risk_level" in result assert "schedule_status" in result assert "resource_status" in result def test_get_project_health_not_found(self, client, db, admin_token): """Should return 404 for non-existent project.""" self.setup_test_data(db) response = client.get( "/api/projects/health/non-existent-id", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 404 assert response.json()["detail"] == "Project not found" def test_get_project_health_with_issues(self, client, db, admin_token): """Project with issues should have correct metrics.""" data = self.setup_test_data(db) response = client.get( f"/api/projects/health/{data['project1'].id}", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 result = response.json() # Project1 has 1 overdue task with 1 blocker assert result["blocker_count"] == 1 assert result["overdue_task_count"] == 1 assert result["health_score"] < 100 # Should be penalized def test_unauthorized_access(self, client, db): """Unauthenticated requests should fail.""" response = client.get("/api/projects/health/dashboard") assert response.status_code == 403 def test_dashboard_with_status_filter(self, client, db, admin_token): """Dashboard should respect status filter.""" data = self.setup_test_data(db) # Create an archived project archived = Project( id="project-archived", space_id="space-api-001", title="Archived Project", owner_id="00000000-0000-0000-0000-000000000001", department_id="dept-api-001", security_level="department", status="archived", ) db.add(archived) db.commit() # Default filter should exclude archived response = client.get( "/api/projects/health/dashboard", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 project_ids = [p["project_id"] for p in response.json()["projects"]] assert "project-archived" not in project_ids def test_project_health_response_structure(self, client, db, admin_token): """Response should match ProjectHealthWithDetails schema.""" data = self.setup_test_data(db) response = client.get( f"/api/projects/health/{data['project1'].id}", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 result = response.json() # Required fields from schema required_fields = [ "id", "project_id", "health_score", "risk_level", "schedule_status", "resource_status", "last_updated", "project_title", "project_status", "task_count", "completed_task_count", "blocker_count", "overdue_task_count" ] for field in required_fields: assert field in result, f"Missing field: {field}" # Check enum values assert result["risk_level"] in ["low", "medium", "high", "critical"] assert result["schedule_status"] in ["on_track", "at_risk", "delayed"] assert result["resource_status"] in ["adequate", "constrained", "overloaded"]