Coverage for modules\planner\services.py: 26%

1from django.db import transaction 

2from core.exceptions import NotFoundException, BadRequestException 

3from .models import Project, Board, BoardColumn, Task, TaskComment, TimeLog 

4from modules.authentication.models import User, Membership 

5from modules.team.models import TeamMember, Department 

6 

7class PlannerService: 

8 @staticmethod 

9 def visible_projects(tenant_id, user): 

10 projects = Project.objects.filter(tenant_id=tenant_id) 

11 membership = Membership.objects.filter(organization_id=tenant_id, user=user).first() 

12 if user.is_superuser or (membership and membership.role in ('OWNER', 'ADMIN')): 

13 return projects 

14 department_id = TeamMember.objects.filter(tenant_id=tenant_id, user=user, status='ACTIVE').values_list('department_id', flat=True).first() 

15 return projects.filter(department_id=department_id) if department_id else projects.none() 

16 

17 @staticmethod 

18 def can_access_project(tenant_id, user, project_id): 

19 return PlannerService.visible_projects(tenant_id, user).filter(id=project_id).first() 

20 

21 @staticmethod 

22 @transaction.atomic 

23 def create_project(tenant_id, user, data): 

24 key = data.key.upper().strip() 

25 if Project.objects.filter(tenant_id=tenant_id, key=key).exists(): 

26 raise BadRequestException(f"Project key '{key}' is already in use.") 

27 

28 lead = User.objects.filter(id=data.lead_id).first() if data.lead_id else None 

29 department_id = data.department_id or TeamMember.objects.filter(tenant_id=tenant_id, user=user, status='ACTIVE').values_list('department_id', flat=True).first() 

30 is_global = user.is_superuser or Membership.objects.filter(organization_id=tenant_id, user=user, role__in=('OWNER', 'ADMIN')).exists() 

31 if not department_id and not is_global: 

32 raise BadRequestException('Assign a department before creating a project.') 

33 if not is_global and not TeamMember.objects.filter(tenant_id=tenant_id, user=user, department_id=department_id, status='ACTIVE').exists(): 

34 raise BadRequestException('You can only create projects for your department.') 

35 department = Department.objects.filter(id=department_id, tenant_id=tenant_id).first() if department_id else None 

36 

37 project = Project.objects.create( 

38 tenant_id=tenant_id, 

39 name=data.name, 

40 key=key, 

41 description=data.description, 

42 color=data.color or '#6366F1', 

43 lead=lead, 

44 department=department, 

45 goal=data.goal, 

46 roadmap_start=data.roadmap_start, 

47 roadmap_end=data.roadmap_end, 

48 ) 

49 

50 # Create default board with columns 

51 board = Board.objects.create( 

52 tenant_id=tenant_id, 

53 project=project, 

54 name="Main Kanban Board" 

55 ) 

56 

57 default_columns = [ 

58 ("Backlog", "BACKLOG", 0), 

59 ("To Do", "TODO", 1), 

60 ("In Progress", "IN_PROGRESS", 2), 

61 ("In Review", "IN_REVIEW", 3), 

62 ("Done", "DONE", 4), 

63 ] 

64 for name, cat, order in default_columns: 

65 BoardColumn.objects.create( 

66 tenant_id=tenant_id, 

67 board=board, 

68 name=name, 

69 status_category=cat, 

70 order=order 

71 ) 

72 

73 return project 

74 

75 @staticmethod 

76 @transaction.atomic 

77 def create_task(tenant_id, reporter, data): 

78 project = PlannerService.can_access_project(tenant_id, reporter, data.project_id) 

79 if not project: 

80 raise NotFoundException("Project not found.") 

81 

82 # Sequential task number per project 

83 last_task = Task.objects.filter(project=project).order_by('-task_number').first() 

84 next_number = (last_task.task_number + 1) if last_task else 1 

85 

86 board = Board.objects.filter(project=project, tenant_id=tenant_id).first() 

87 column = None 

88 if data.column_id: 

89 column = BoardColumn.objects.filter(id=data.column_id, tenant_id=tenant_id).first() 

90 elif board: 

91 column = board.columns.filter(status_category='TODO').first() or board.columns.first() 

92 

93 status = column.status_category if column else (data.status or 'TODO') 

94 assignee = User.objects.filter(id=data.assignee_id).first() if data.assignee_id else None 

95 

96 task = Task.objects.create( 

97 tenant_id=tenant_id, 

98 project=project, 

99 board=board, 

100 column=column, 

101 title=data.title, 

102 description=data.description, 

103 task_number=next_number, 

104 priority=data.priority or 'MEDIUM', 

105 status=status, 

106 assignee=assignee, 

107 reporter=reporter, 

108 due_date=data.due_date, 

109 estimated_hours=data.estimated_hours or 0, 

110 tags=data.tags or [], 

111 ) 

112 return task 

113 

114 @staticmethod 

115 def move_task(tenant_id, user, task_id, column_id, order, status=None): 

116 task = Task.objects.filter(id=task_id, tenant_id=tenant_id, project__in=PlannerService.visible_projects(tenant_id, user)).first() 

117 if not task: 

118 raise NotFoundException("Task not found.") 

119 

120 column = BoardColumn.objects.filter(id=column_id, tenant_id=tenant_id).first() 

121 if not column: 

122 raise NotFoundException("Target column not found.") 

123 

124 task.column = column 

125 task.order = order 

126 task.status = status or column.status_category 

127 task.save(update_fields=['column', 'order', 'status']) 

128 return task