Big refactor on is_closed calculation

remotes/origin/enhancement/email-actions
Jesús Espino 2014-07-22 12:00:16 +02:00
parent 4e043762bf
commit d5df49f59a
4 changed files with 247 additions and 207 deletions

View File

@ -26,6 +26,7 @@ from taiga.base.utils.slug import ref_uniquely
from taiga.projects.notifications import WatchedModelMixin
from taiga.projects.occ import OCCModelMixin
from taiga.projects.userstories.models import UserStory
from taiga.projects.userstories import services as us_service
from taiga.projects.milestones.models import Milestone
from taiga.projects.mixins.blocked import BlockedMixin
@ -83,38 +84,11 @@ class Task(OCCModelMixin, WatchedModelMixin, BlockedMixin, TaggedMixin, models.M
return value
def us_has_open_tasks(us, exclude_task):
qs = us.tasks.all()
if exclude_task.pk:
qs = qs.exclude(pk=exclude_task.pk)
return not all(task.status.is_closed for task in qs)
def milestone_has_open_userstories(milestone):
qs = milestone.user_stories.exclude(is_closed=True)
return qs.exists()
def close_user_story(us):
us.is_closed = True
us.finish_date = timezone.now()
us.save(update_fields=["is_closed", "finish_date"])
def open_user_story(us):
us.is_closed = False
us.finish_date = None
us.save(update_fields=["is_closed", "finish_date"])
@receiver(models.signals.post_delete, sender=Task, dispatch_uid="tasks_us_close_handler_on_delete")
def tasks_us_close_handler_on_delete(sender, instance, **kwargs):
if (instance.user_story_id
and UserStory.objects.filter(id=instance.user_story_id)
and not us_has_open_tasks(us=instance.user_story, exclude_task=instance)):
close_user_story(instance.user_story)
@receiver(models.signals.post_delete, sender=Task, dispatch_uid="tasks_milestone_close_handler_on_delete")
def tasks_milestone_close_handler_on_delete(sender, instance, **kwargs):
if instance.milestone_id and Milestone.objects.filter(id=instance.milestone_id):
@ -123,41 +97,36 @@ def tasks_milestone_close_handler_on_delete(sender, instance, **kwargs):
instance.milestone.save(update_fields=["closed"])
# Define the previous version of the task for use it on the post_save handler
@receiver(models.signals.pre_save, sender=Task, dispatch_uid="tasks_us_close_handler")
def tasks_us_close_handler(sender, instance, **kwargs):
instance.prev = None
if instance.id:
orig_instance = sender.objects.get(id=instance.id)
instance.prev = sender.objects.get(id=instance.id)
if (instance.user_story_id != orig_instance.user_story_id
and orig_instance.user_story_id
and not orig_instance.status.is_closed
and not us_has_open_tasks(us=orig_instance.user_story, exclude_task=orig_instance)):
close_user_story(orig_instance.user_story)
if not instance.user_story_id:
return
if orig_instance.status.is_closed != instance.status.is_closed:
if orig_instance.status.is_closed and not instance.status.is_closed:
open_user_story(instance.user_story)
elif not us_has_open_tasks(us=instance.user_story, exclude_task=instance):
close_user_story(instance.user_story)
if instance.user_story_id != orig_instance.user_story_id and instance.user_story.is_closed:
if instance.status.is_closed:
close_user_story(instance.user_story)
else:
open_user_story(instance.user_story)
else: # ON CREATION
if not instance.user_story_id:
return
if (instance.status.is_closed
and not us_has_open_tasks(us=instance.user_story, exclude_task=instance)):
close_user_story(instance.user_story)
@receiver(models.signals.post_save, sender=Task, dispatch_uid="tasks_us_close_on_create_handler")
def tasks_us_close_on_create_handler(sender, instance, created, **kwargs):
if instance.user_story_id:
if us_service.calculate_userstory_is_closed(instance.user_story):
us_service.close_userstory(instance.user_story)
else:
open_user_story(instance.user_story)
us_service.open_userstory(instance.user_story)
if instance.prev and instance.prev.user_story_id:
if us_service.calculate_userstory_is_closed(instance.prev.user_story):
us_service.close_userstory(instance.prev.user_story)
else:
us_service.open_userstory(instance.prev.user_story)
@receiver(models.signals.post_delete, sender=Task, dispatch_uid="tasks_us_close_handler_on_delete")
def tasks_us_close_handler_on_delete(sender, instance, **kwargs):
if instance.user_story_id:
if us_service.calculate_userstory_is_closed(instance.user_story):
us_service.close_userstory(instance.user_story)
else:
us_service.open_userstory(instance.user_story)
@receiver(models.signals.pre_save, sender=Task, dispatch_uid="tasks_milestone_close_handler")

View File

@ -16,7 +16,6 @@
from django.db import models
from django.contrib.contenttypes import generic
from django.utils import timezone
from django.conf import settings
from django.dispatch import receiver
from django.utils.translation import ugettext_lazy as _
@ -160,11 +159,9 @@ def us_tags_normalization(sender, instance, **kwargs):
@receiver(models.signals.post_save, sender=UserStory,
dispatch_uid="user_story_on_status_change")
def us_close_open_on_status_change(sender, instance, **kwargs):
if instance.tasks.count() == 0:
if instance.is_closed != instance.status.is_closed:
instance.is_closed = instance.status.is_closed
if instance.is_closed:
instance.finish_date = timezone.now()
else:
instance.finish_date = None
instance.save(update_fields=['is_closed', 'finish_date'])
from taiga.projects.userstories import services as service
if service.calculate_userstory_is_closed(instance):
service.close_userstory(instance)
else:
service.open_userstory(instance)

View File

@ -14,6 +14,8 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from django.utils import timezone
from taiga.base.utils import db, text
from . import models
@ -58,3 +60,27 @@ def update_userstories_order_in_bulk(bulk_data):
user_story_ids.append(user_story_id)
new_order_values.append({"order": new_order_value})
db.update_in_bulk_with_ids(user_story_ids, new_order_values, model=models.UserStory)
def calculate_userstory_is_closed(user_story):
if user_story.tasks.count() == 0:
return user_story.status.is_closed
if all([task.status.is_closed for task in user_story.tasks.all()]):
return True
return False
def close_userstory(us):
if not us.is_closed:
us.is_closed = True
us.finish_date = timezone.now()
us.save(update_fields=["is_closed", "finish_date"])
def open_userstory(us):
if us.is_closed:
us.is_closed = False
us.finish_date = None
us.save(update_fields=["is_closed", "finish_date"])

View File

@ -15,11 +15,7 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from unittest.mock import patch, MagicMock, call
from django.core.exceptions import ValidationError
from tests import factories
from tests import factories as f
from taiga.projects.userstories.models import UserStory
@ -28,160 +24,212 @@ import pytest
pytestmark = pytest.mark.django_db
def test_us_without_tasks_close():
closed_status = factories.UserStoryStatusFactory(is_closed=True)
open_status = factories.UserStoryStatusFactory(is_closed=False)
user_story = factories.UserStoryFactory(status=open_status)
assert user_story.is_closed == False
user_story.status = closed_status
user_story.save()
user_story = UserStory.objects.get(pk=user_story.pk)
assert user_story.is_closed == True
@pytest.fixture
def data():
m = type("Models", (object,), {})
m.us_closed_status = f.UserStoryStatusFactory(is_closed=True)
m.us_open_status = f.UserStoryStatusFactory(is_closed=False)
m.task_closed_status = f.TaskStatusFactory(is_closed=True)
m.task_open_status = f.TaskStatusFactory(is_closed=False)
m.user_story1 = f.UserStoryFactory(status=m.us_open_status)
m.user_story2 = f.UserStoryFactory(status=m.us_open_status)
m.task1 = f.TaskFactory(user_story=m.user_story1, status=m.task_open_status)
m.task2 = f.TaskFactory(user_story=m.user_story1, status=m.task_open_status)
m.task3 = f.TaskFactory(user_story=m.user_story1, status=m.task_open_status)
return m
def test_us_without_tasks_open():
closed_status = factories.UserStoryStatusFactory(is_closed=True)
open_status = factories.UserStoryStatusFactory(is_closed=False)
user_story = factories.UserStoryFactory(status=closed_status)
assert user_story.is_closed == True
user_story.status = open_status
user_story.save()
user_story = UserStory.objects.get(pk=user_story.pk)
assert user_story.is_closed == False
def test_us_without_tasks_open_close_us_status(data):
assert data.user_story2.is_closed is False
data.user_story2.status = data.us_closed_status
data.user_story2.save()
data.user_story2 = UserStory.objects.get(pk=data.user_story2.pk)
assert data.user_story2.is_closed is True
data.user_story2.status = data.us_open_status
data.user_story2.save()
data.user_story2 = UserStory.objects.get(pk=data.user_story2.pk)
assert data.user_story2.is_closed is False
def test_us_with_tasks_close():
closed_status = factories.UserStoryStatusFactory(is_closed=True)
open_status = factories.UserStoryStatusFactory(is_closed=False)
closed_task_status = factories.TaskStatusFactory(is_closed=True)
open_task_status = factories.TaskStatusFactory(is_closed=False)
user_story = factories.UserStoryFactory(status=closed_status)
task1 = factories.TaskFactory(user_story=user_story, status=closed_task_status)
task2 = factories.TaskFactory(user_story=user_story, status=closed_task_status)
task3 = factories.TaskFactory(user_story=user_story, status=closed_task_status)
assert user_story.is_closed == True
user_story.status = open_status
user_story.save()
user_story = UserStory.objects.get(pk=user_story.pk)
assert user_story.is_closed == True
def test_us_with_tasks_open_close_us_status(data):
assert data.user_story1.is_closed is False
data.user_story1.status = data.us_closed_status
data.user_story1.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
data.user_story1.status = data.us_open_status
data.user_story1.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
def test_us_with_tasks_on_delete_empty_open():
closed_status = factories.UserStoryStatusFactory(is_closed=True)
open_status = factories.UserStoryStatusFactory(is_closed=False)
closed_task_status = factories.TaskStatusFactory(is_closed=True)
open_task_status = factories.TaskStatusFactory(is_closed=False)
user_story = factories.UserStoryFactory(status=open_status)
task1 = factories.TaskFactory(user_story=user_story, status=closed_task_status)
assert user_story.is_closed == True
task1.delete()
user_story = UserStory.objects.get(pk=user_story.pk)
assert user_story.is_closed == False
def test_us_on_task_delete_empty_close(data):
data.user_story1.status = data.us_closed_status
data.user_story1.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
data.task3.delete()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
data.task2.delete()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
data.task1.delete()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
def test_us_with_tasks_on_delete_empty_close():
closed_status = factories.UserStoryStatusFactory(is_closed=True)
open_status = factories.UserStoryStatusFactory(is_closed=False)
closed_task_status = factories.TaskStatusFactory(is_closed=True)
open_task_status = factories.TaskStatusFactory(is_closed=False)
user_story = factories.UserStoryFactory(status=closed_status)
task1 = factories.TaskFactory(user_story=user_story, status=open_task_status)
assert user_story.is_closed == False
task1.delete()
user_story = UserStory.objects.get(pk=user_story.pk)
assert user_story.is_closed == True
def test_us_on_task_delete_empty_open(data):
data.task1.status = data.task_closed_status
data.task1.save()
data.task2.status = data.task_closed_status
data.task2.save()
data.task3.status = data.task_closed_status
data.task3.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
data.task3.delete()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
data.task2.delete()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
data.task1.delete()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
def test_us_with_tasks_open():
closed_status = factories.UserStoryStatusFactory(is_closed=True)
open_status = factories.UserStoryStatusFactory(is_closed=False)
closed_task_status = factories.TaskStatusFactory(is_closed=True)
open_task_status = factories.TaskStatusFactory(is_closed=False)
user_story = factories.UserStoryFactory(status=open_status)
task1 = factories.TaskFactory(user_story=user_story, status=closed_task_status)
task2 = factories.TaskFactory(user_story=user_story, status=closed_task_status)
task3 = factories.TaskFactory(user_story=user_story, status=open_task_status)
assert user_story.is_closed == False
user_story.status = closed_status
user_story.save()
user_story = UserStory.objects.get(pk=user_story.pk)
assert user_story.is_closed == False
def test_us_with_tasks_on_move_empty_open(data):
data.task1.status = data.task_closed_status
data.task1.save()
data.task2.status = data.task_closed_status
data.task2.save()
data.task3.status = data.task_closed_status
data.task3.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
data.task3.user_story = data.user_story2
data.task3.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
data.task2.user_story = data.user_story2
data.task2.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
data.task1.user_story = data.user_story2
data.task1.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
def test_us_close_last_tasks():
closed_status = factories.TaskStatusFactory(is_closed=True)
open_status = factories.TaskStatusFactory(is_closed=False)
user_story = factories.UserStoryFactory()
task1 = factories.TaskFactory(user_story=user_story, status=closed_status)
task2 = factories.TaskFactory(user_story=user_story, status=closed_status)
task3 = factories.TaskFactory(user_story=user_story, status=open_status)
assert user_story.is_closed == False
task3.status = closed_status
task3.save()
user_story = UserStory.objects.get(pk=user_story.pk)
assert user_story.is_closed == True
def test_us_with_tasks_on_move_empty_close(data):
data.user_story1.status = data.us_closed_status
data.user_story1.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
data.task3.user_story = data.user_story2
data.task3.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
data.task2.user_story = data.user_story2
data.task2.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
data.task1.user_story = data.user_story2
data.task1.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
def test_us_with_all_closed_open_task():
closed_status = factories.TaskStatusFactory(is_closed=True)
open_status = factories.TaskStatusFactory(is_closed=False)
user_story = factories.UserStoryFactory()
task1 = factories.TaskFactory(user_story=user_story, status=closed_status)
task2 = factories.TaskFactory(user_story=user_story, status=closed_status)
task3 = factories.TaskFactory(user_story=user_story, status=closed_status)
assert user_story.is_closed == True
task3.status = open_status
task3.save()
user_story = UserStory.objects.get(pk=user_story.pk)
assert user_story.is_closed == False
def test_us_close_last_tasks(data):
assert data.user_story1.is_closed is False
data.task3.status = data.task_closed_status
data.task3.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
data.task2.status = data.task_closed_status
data.task2.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
data.task1.status = data.task_closed_status
data.task1.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
def test_us_delete_task_then_all_closed():
closed_status = factories.TaskStatusFactory(is_closed=True)
open_status = factories.TaskStatusFactory(is_closed=False)
user_story = factories.UserStoryFactory()
task1 = factories.TaskFactory(user_story=user_story, status=closed_status)
task2 = factories.TaskFactory(user_story=user_story, status=closed_status)
task3 = factories.TaskFactory(user_story=user_story, status=open_status)
assert user_story.is_closed == False
task3.delete()
user_story = UserStory.objects.get(pk=user_story.pk)
assert user_story.is_closed == True
def test_us_reopen_tasks(data):
data.task1.status = data.task_closed_status
data.task1.save()
data.task2.status = data.task_closed_status
data.task2.save()
data.task3.status = data.task_closed_status
data.task3.save()
assert data.user_story1.is_closed is True
data.task3.status = data.task_open_status
data.task3.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
data.task2.status = data.task_open_status
data.task2.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
data.task1.status = data.task_open_status
data.task1.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
def test_us_change_task_us_then_all_closed():
closed_status = factories.TaskStatusFactory(is_closed=True)
open_status = factories.TaskStatusFactory(is_closed=False)
user_story1 = factories.UserStoryFactory()
user_story2 = factories.UserStoryFactory()
task1 = factories.TaskFactory(user_story=user_story1, status=closed_status)
task2 = factories.TaskFactory(user_story=user_story1, status=closed_status)
task3 = factories.TaskFactory(user_story=user_story1, status=open_status)
assert user_story1.is_closed == False
task3.user_story = user_story2
task3.save()
user_story1 = UserStory.objects.get(pk=user_story1.pk)
assert user_story1.is_closed == True
def test_us_delete_task_then_all_closed(data):
data.task1.status = data.task_closed_status
data.task1.save()
data.task2.status = data.task_closed_status
data.task2.save()
assert data.user_story1.is_closed is False
data.task3.delete()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
def test_us_change_task_us_to_us_with_all_closed():
closed_status = factories.TaskStatusFactory(is_closed=True)
open_status = factories.TaskStatusFactory(is_closed=False)
user_story1 = factories.UserStoryFactory()
user_story2 = factories.UserStoryFactory()
task1 = factories.TaskFactory(user_story=user_story1, status=closed_status)
task2 = factories.TaskFactory(user_story=user_story1, status=closed_status)
task3 = factories.TaskFactory(user_story=user_story2, status=open_status)
assert user_story1.is_closed == True
task3.user_story = user_story1
task3.save()
user_story1 = UserStory.objects.get(pk=user_story1.pk)
assert user_story1.is_closed == False
def test_us_change_task_us_then_all_closed(data):
data.task1.status = data.task_closed_status
data.task1.save()
data.task2.status = data.task_closed_status
data.task2.save()
assert data.user_story1.is_closed is False
data.task3.user_story = data.user_story2
data.task3.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
def test_us_change_task_us_then_any_open(data):
data.task1.status = data.task_closed_status
data.task1.save()
data.task2.status = data.task_closed_status
data.task2.save()
data.task3.user_story = data.user_story2
data.task3.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
data.task3.user_story = data.user_story1
data.task3.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False
def test_task_create(data):
data.task1.status = data.task_closed_status
data.task1.save()
data.task2.status = data.task_closed_status
data.task2.save()
data.task3.status = data.task_closed_status
data.task3.save()
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
f.TaskFactory(user_story=data.user_story1, status=data.task_closed_status)
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is True
f.TaskFactory(user_story=data.user_story1, status=data.task_open_status)
data.user_story1 = UserStory.objects.get(pk=data.user_story1.pk)
assert data.user_story1.is_closed is False