Add capacity to sprint

By default sprint capacity is set to 40 hours (ideal week) but can be
adjusted during the creating /editing.
Tasks cannot be added to the sprint that is overplanned.

terka/adapters/orm.py

@@ -221,6 +221,7 @@
                 Column("start_date", Date, nullable=False),
                 Column("end_date", Date, nullable=False),
                 Column("status", Enum(sprint.SprintStatus)),
+                Column("capacity", Integer, nullable=False),
                 Column("goal", String(225), nullable=True))
 
 sprint_tasks = Table(

terka/domain/commands.py

@@ -350,6 +350,7 @@ class CreateSprint(Command):
     goal: str | None = None
     start_date: str | None = None
     end_date: str | None = None
+    capacity: int = 40
 
     def __bool__(self) -> bool:
         if self.start_date and self.end_date:

terka/domain/entities/sprint.py

@@ -1,4 +1,5 @@
-from typing import Set
+from __future__ import annotations
+
 from collections import defaultdict
 from enum import Enum
 import logging
@@ -21,10 +22,11 @@ class SprintStatus(Enum):
 class Sprint(Entity):
 
     def __init__(self,
-                 start_date: datetime = None,
-                 end_date: datetime = None,
+                 start_date: datetime | None = None,
+                 end_date: datetime | None = None,
                  status: str = "PLANNED",
-                 goal: str = None,
+                 capacity: int = 40,
+                 goal: str | None = None,
                  **kwargs) -> None:
         if not start_date and not end_date:
             raise ValueError("Please add start and end date of the sprint")
@@ -45,6 +47,7 @@ def __init__(self,
             raise ValueError(f"Sprint end date cannot be less than today")
         self.status = self._validate_status(status)
         self.goal = goal
+        self.capacity = capacity
         self.is_completed = False
 
     def _validate_status(self, status):
@@ -64,6 +67,15 @@ def complete(self, tasks) -> None:
         self.is_completed = True
 
     @property
+    def overplanned(self) -> bool:
+        return self.velocity > self.capacity
+
+    @property
+    def remaining_capatity(self) -> float:
+        if not self.overplanned:
+            return self.capacity - self.velocity
+        return 0
+    @property
     def velocity(self) -> float:
         return round(sum([t.story_points for t in self.tasks]), 1)
 

terka/exceptions.py

@@ -33,6 +33,10 @@ class TerkaSprintActive(TerkaException):
 class TerkaInitError(TerkaException):
     ...
 
+
 class TerkaRefreshException(TerkaException):
     ...
 
+
+class TerkaSprintOutOfCapacity(TerkaException):
+    ...

terka/presentations/text_ui/ui.py

@@ -858,7 +858,7 @@ def compose(self) -> ComposeResult:
 * Period: {self.entity.start_date} - {self.entity.end_date}
 * Open tasks: {len(self.entity.open_tasks)} ({len(self.entity.tasks)})
 * Pct Completed: {round(self.entity.pct_completed, 2) :.0%}
-* Velocity: {self.entity.velocity}
+* Velocity: {self.entity.velocity} ({self.entity.capacity})
 * Time spend: {Formatter.format_time_spent(self.entity.total_time_spent)}
 * Utilization: {round(self.entity.utilization, 2) :.0%}
 ## Collaborator split:

terka/presentations/vim/templates.py

@@ -60,6 +60,7 @@ def new_sprint_template() -> str:
         description: 
         start_date: {next_monday}
         end_date:  {next_sunday}
+        capacity: 40
         """
 
 
@@ -75,6 +76,8 @@ def edit_sprint_template(sprint: entities.sprint.Sprint) -> str:
         status: {sprint.status.name}
         start_date: {sprint.start_date}
         end_date: {sprint.end_date}
+        capacity: {sprint.capacity}
+
         """
 
 

terka/service_layer/handlers.py

@@ -385,19 +385,29 @@ def _add(cmd: commands.AddTask, bus: "messagebus.MessageBus",
                 uow.commit()
                 logging.debug(f"Task added to {entity_name}, context {cmd}")
                 if entity_name == "sprint":
-                    if existing_entity.status == entities.sprint.SprintStatus.COMPLETED:
+                    if (existing_entity.status ==
+                            entities.sprint.SprintStatus.COMPLETED):
                         raise exceptions.TerkaSprintCompleted(
                             f"Sprint {entity_id} is completed")
+                    if existing_entity.overplanned:
+                        raise exceptions.TerkaSprintOutOfCapacity(
+                            f"Sprint {entity_id} is overplanned")
+                    if (entity_task.story_points >
+                            existing_entity.remaining_capacity):
+                        raise exceptions.TerkaSprintOutOfCapacity(
+                            f"Sprint {entity_id} will overplanned "
+                            f"when task with {entity_task.story_points} is added"
+                        )
                     if existing_task := uow.tasks.get_by_id(
                             entities.task.Task, cmd.id):
                         task_params = {}
                         if existing_task.status.name == "BACKLOG":
                             task_params.update({"status": "TODO"})
                         if (not existing_task.due_date
-                                or existing_task.due_date
-                                > existing_entity.end_date
-                                or existing_task.due_date
-                                < existing_entity.start_date):
+                                or existing_task.due_date >
+                                existing_entity.end_date
+                                or existing_task.due_date <
+                                existing_entity.start_date):
                             task_params.update(
                                 {"due_date": existing_entity.end_date})
                         if task_params:

tests/test_handers.py

@@ -64,10 +64,10 @@ def test_deleting_task_creates_status_task_event(self, bus):
     def test_creating_task_with_tag_creates_tag(self, bus):
         cmd = commands.CreateTask(name="test")
         task_id = bus.handle(cmd, context={"tags": "new_tag"})
-        new_task_tag = bus.uow.tasks.get_by_conditions(
-            entities.tag.TaskTag, {"task": task_id})
-        new_base_tag = bus.uow.tasks.get_by_conditions(
-            entities.tag.BaseTag, {"text": "new_tag"})
+        new_task_tag = bus.uow.tasks.get_by_conditions(entities.tag.TaskTag,
+                                                       {"task": task_id})
+        new_base_tag = bus.uow.tasks.get_by_conditions(entities.tag.BaseTag,
+                                                       {"text": "new_tag"})
         assert new_task_tag
         assert new_base_tag
 
@@ -78,7 +78,7 @@ def test_creating_task_with_collaborator_creates_user(self, bus):
         [new_task_collaborator] = bus.uow.tasks.get_by_conditions(
             entities.collaborator.TaskCollaborator, {"task": task_id})
         new_user = bus.uow.tasks.get_by_id(entities.user.User,
-                                                   new_task_collaborator.id)
+                                           new_task_collaborator.id)
         assert new_task_collaborator
         assert new_user
 
@@ -99,16 +99,16 @@ def test_commenting_task_with_empty_comment_is_ignored(self, bus):
     def test_tagging_task_with_empty_tag_is_ignored(self, bus):
         cmd = commands.CreateTask(name="test")
         task_id = bus.handle(cmd, context={"tag": " "})
-        new_task_tag = bus.uow.tasks.get_by_conditions(
-            entities.tag.TaskTag, {"task": task_id})
+        new_task_tag = bus.uow.tasks.get_by_conditions(entities.tag.TaskTag,
+                                                       {"task": task_id})
         assert not new_task_tag
 
     @pytest.mark.parametrize("entity", ["epic", "sprint", "story"])
     def test_tagging_task_with_service_tag_is_ignored(self, bus, entity):
         cmd = commands.CreateTask(name="test")
         task_id = bus.handle(cmd, context={"tag": f"{entity}: 1"})
-        new_task_tag = bus.uow.tasks.get_by_conditions(
-            entities.tag.TaskTag, {"task": task_id})
+        new_task_tag = bus.uow.tasks.get_by_conditions(entities.tag.TaskTag,
+                                                       {"task": task_id})
         assert not new_task_tag
 
 
@@ -128,7 +128,18 @@ def new_sprint(self, bus):
         next_monday = (today + timedelta(days=(7 - today.weekday())))
         next_sunday = (today + timedelta(days=(13 - today.weekday())))
         cmd = commands.CreateSprint(start_date=next_monday,
-                                     end_date=next_sunday)
+                                    end_date=next_sunday)
+        sprint_id = bus.handle(cmd)
+        return sprint_id
+
+    @pytest.fixture
+    def new_sprint_with_limited_capacity(self, bus):
+        today = datetime.now()
+        next_monday = (today + timedelta(days=(7 - today.weekday())))
+        next_sunday = (today + timedelta(days=(13 - today.weekday())))
+        cmd = commands.CreateSprint(start_date=next_monday,
+                                    end_date=next_sunday,
+                                    capacity=1)
         sprint_id = bus.handle(cmd)
         return sprint_id
 
@@ -164,18 +175,27 @@ def test_adding_task_to_sprint(self, bus, new_sprint, sprint_with_tasks):
         for task in sprint_tasks:
             assert task.story_points == 0
 
+    def test_cannot_add_task_to_sprint(self, bus,
+                                       new_sprint_with_limited_capacity,
+                                       tasks):
+        task_1, task_2 = tasks
+        add_task_1 = commands.AddTask(id=task_1,
+                                      sprint=new_sprint_with_limited_capacity,
+                                      story_points=2)
+        with pytest.raises(exceptions.TerkaSprintOutOfCapacity):
+            bus.handle(add_task_1)
+
     def test_starting_sprint_changes_status_due_date(self, bus, new_sprint,
                                                      sprint_with_tasks):
         cmd = commands.StartSprint(new_sprint)
         bus.handle(cmd)
-        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint,
-                                                 new_sprint)
+        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint, new_sprint)
         assert sprint.status == entities.sprint.SprintStatus.ACTIVE
         sprint_tasks = bus.uow.tasks.get_by_conditions(
             entities.sprint.SprintTask, {"sprint": new_sprint})
         for sprint_task in sprint_tasks:
             task = bus.uow.tasks.get_by_id(entities.task.Task,
-                                                   sprint_task.task)
+                                           sprint_task.task)
             assert task.status == entities.task.TaskStatus.TODO
             assert task.due_date == sprint.end_date
 
@@ -192,38 +212,35 @@ def test_cannot_start_completed_sprint(self, bus, new_sprint):
         cmd = commands.CompleteSprint(new_sprint)
         bus.handle(cmd)
         cmd = commands.StartSprint(new_sprint)
-        with pytest.raises(
-                exceptions.TerkaSprintCompleted):
+        with pytest.raises(exceptions.TerkaSprintCompleted):
             bus.handle(cmd)
 
     def test_completing_sprint_changes_status_due_date(self, bus, new_sprint,
                                                        sprint_with_tasks):
         cmd = commands.CompleteSprint(new_sprint)
         bus.handle(cmd)
-        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint,
-                                                 new_sprint)
+        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint, new_sprint)
         assert sprint.status == entities.sprint.SprintStatus.COMPLETED
         sprint_tasks = bus.uow.tasks.get_by_conditions(
             entities.sprint.SprintTask, {"sprint": new_sprint})
         for sprint_task in sprint_tasks:
             task = bus.uow.tasks.get_by_id(entities.task.Task,
-                                                   sprint_task.task)
+                                           sprint_task.task)
             assert task.status == entities.task.TaskStatus.BACKLOG
             assert not task.due_date
 
     def test_completing_sprint_with_in_progress_review_tasks_doesnt_change_their_status(
             self, bus, new_sprint, sprint_with_tasks):
         cmd = commands.StartSprint(new_sprint)
         bus.handle(cmd)
-        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint,
-                                                 new_sprint)
+        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint, new_sprint)
         update_task = commands.UpdateTask(id=1, status="REVIEW")
         bus.handle(update_task)
         sprint_tasks = bus.uow.tasks.get_by_conditions(
             entities.sprint.SprintTask, {"sprint": new_sprint})
         for sprint_task in sprint_tasks:
             task = bus.uow.tasks.get_by_id(entities.task.Task,
-                                                   sprint_task.task)
+                                           sprint_task.task)
             if task.id == 1:
                 assert task.status == entities.task.TaskStatus.REVIEW
 
@@ -234,21 +251,18 @@ def test_cannot_add_ask_to_completed_sprint(self, bus, new_sprint,
         create_task_3 = commands.CreateTask(name="task_3")
         task_3 = bus.handle(create_task_3)
         cmd = commands.AddTask(id=task_3, sprint=new_sprint)
-        with pytest.raises(
-                exceptions.TerkaSprintCompleted):
+        with pytest.raises(exceptions.TerkaSprintCompleted):
             bus.handle(cmd)
 
     def test_deleting_task_from_sprint(self, bus, new_sprint,
                                        sprint_with_tasks):
         cmd = commands.StartSprint(new_sprint)
         bus.handle(cmd)
-        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint,
-                                                 new_sprint)
+        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint, new_sprint)
         delete_task = commands.DeleteTask(id=sprint.tasks[0].task,
-                                           sprint=new_sprint)
+                                          sprint=new_sprint)
         bus.handle(delete_task)
-        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint,
-                                                 new_sprint)
+        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint, new_sprint)
         assert len(sprint.tasks) == 1
 
     def test_adding_story_to_sprint_adds_non_completed_tasks(
@@ -260,11 +274,9 @@ def test_adding_story_to_sprint_adds_non_completed_tasks(
         add_task_2 = commands.AddTask(id=task_2, story=story_id)
         bus.handle(add_task_1)
         bus.handle(add_task_2)
-        add_story_to_sprint = commands.AddStory(id=story_id,
-                                                 sprint=new_sprint)
+        add_story_to_sprint = commands.AddStory(id=story_id, sprint=new_sprint)
         bus.handle(add_story_to_sprint)
-        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint,
-                                                 new_sprint)
+        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint, new_sprint)
         assert len(sprint.tasks) == 2
 
     def test_adding_epic_to_sprint_adds_non_completed_tasks(
@@ -278,8 +290,7 @@ def test_adding_epic_to_sprint_adds_non_completed_tasks(
         bus.handle(add_task_2)
         add_epic_to_sprint = commands.AddEpic(id=epic_id, sprint=new_sprint)
         bus.handle(add_epic_to_sprint)
-        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint,
-                                                 new_sprint)
+        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint, new_sprint)
         assert len(sprint.tasks) == 2
 
     def test_adding_story_to_sprint_adds_only_non_completed_tasks(
@@ -293,11 +304,9 @@ def test_adding_story_to_sprint_adds_only_non_completed_tasks(
         bus.handle(add_task_2)
         complete_task = commands.CompleteTask(task_1)
         bus.handle(complete_task)
-        add_story_to_sprint = commands.AddStory(id=story_id,
-                                                 sprint=new_sprint)
+        add_story_to_sprint = commands.AddStory(id=story_id, sprint=new_sprint)
         bus.handle(add_story_to_sprint)
-        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint,
-                                                 new_sprint)
+        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint, new_sprint)
         assert len(sprint.tasks) == 1
 
     def test_adding_epic_to_sprint_adds_only_non_completed_tasks(
@@ -313,6 +322,5 @@ def test_adding_epic_to_sprint_adds_only_non_completed_tasks(
         bus.handle(complete_task)
         add_epic_to_sprint = commands.AddEpic(id=epic_id, sprint=new_sprint)
         bus.handle(add_epic_to_sprint)
-        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint,
-                                                 new_sprint)
+        sprint = bus.uow.tasks.get_by_id(entities.sprint.Sprint, new_sprint)
         assert len(sprint.tasks) == 1