A lot of refactoring on close us code signals code
Jesús Espino
parent
a9453b97b7
commit
d32a9562be
|
|
@ -94,72 +94,74 @@ def milestone_has_open_userstories(milestone):
|
|||
qs = milestone.user_stories.exclude(is_closed=True)
|
||||
return qs.exists()
|
||||
|
||||
@receiver(models.signals.post_delete, sender=Task, dispatch_uid="tasks_close_handler_on_delete")
|
||||
def tasks_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):
|
||||
instance.user_story.is_closed = True
|
||||
instance.user_story.finish_date = timezone.now()
|
||||
instance.user_story.save(update_fields=["is_closed", "finish_date"])
|
||||
|
||||
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):
|
||||
if not milestone_has_open_userstories(instance.milestone):
|
||||
instance.milestone.closed = True
|
||||
instance.milestone.save(update_fields=["closed"])
|
||||
|
||||
@receiver(models.signals.pre_save, sender=Task, dispatch_uid="tasks_close_handler")
|
||||
def tasks_close_handler(sender, instance, **kwargs):
|
||||
|
||||
@receiver(models.signals.pre_save, sender=Task, dispatch_uid="tasks_us_close_handler")
|
||||
def tasks_us_close_handler(sender, instance, **kwargs):
|
||||
if instance.id:
|
||||
orig_instance = sender.objects.get(id=instance.id)
|
||||
else:
|
||||
orig_instance = instance
|
||||
|
||||
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:
|
||||
instance.finished_date = None
|
||||
if instance.user_story_id:
|
||||
instance.user_story.is_closed = False
|
||||
instance.user_story.finish_date = None
|
||||
instance.user_story.save(update_fields=["is_closed", "finish_date"])
|
||||
else:
|
||||
instance.finished_date = timezone.now()
|
||||
if instance.user_story_id and not us_has_open_tasks(us=instance.user_story,
|
||||
exclude_task=instance):
|
||||
instance.user_story.is_closed = True
|
||||
instance.user_story.finish_date = timezone.now()
|
||||
instance.user_story.save(update_fields=["is_closed", "finish_date"])
|
||||
elif not instance.id:
|
||||
if not orig_instance.status.is_closed:
|
||||
instance.finished_date = None
|
||||
if instance.user_story_id:
|
||||
instance.user_story.is_closed = False
|
||||
instance.user_story.finish_date = None
|
||||
instance.user_story.save(update_fields=["is_closed", "finish_date"])
|
||||
else:
|
||||
instance.finished_date = timezone.now()
|
||||
if instance.user_story_id and not us_has_open_tasks(us=instance.user_story,
|
||||
exclude_task=instance):
|
||||
instance.user_story.is_closed = True
|
||||
instance.user_story.finish_date = timezone.now()
|
||||
instance.user_story.save(update_fields=["is_closed", "finish_date"])
|
||||
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 the task change its US
|
||||
if instance.user_story_id != orig_instance.user_story_id:
|
||||
if (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)):
|
||||
orig_instance.user_story.is_closed = True
|
||||
orig_instance.user_story.finish_date = timezone.now()
|
||||
orig_instance.user_story.save(update_fields=["is_closed", "finish_date"])
|
||||
|
||||
if instance.user_story_id:
|
||||
if instance.user_story.is_closed:
|
||||
if instance.user_story_id != orig_instance.user_story_id and instance.user_story.is_closed:
|
||||
if instance.status.is_closed:
|
||||
instance.user_story.finish_date = timezone.now()
|
||||
instance.user_story.save(update_fields=["finish_date"])
|
||||
close_user_story(instance.user_story)
|
||||
else:
|
||||
instance.user_story.is_closed = False
|
||||
instance.user_story.finish_date = None
|
||||
instance.user_story.save(update_fields=["is_closed", "finish_date"])
|
||||
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)
|
||||
else:
|
||||
open_user_story(instance.user_story)
|
||||
|
||||
|
||||
@receiver(models.signals.pre_save, sender=Task, dispatch_uid="tasks_milestone_close_handler")
|
||||
def tasks_milestone_close_handler(sender, instance, **kwargs):
|
||||
if instance.milestone_id:
|
||||
if instance.status.is_closed and not instance.milestone.closed:
|
||||
if not milestone_has_open_userstories(instance.milestone):
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@
|
|||
|
||||
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 _
|
||||
|
|
@ -155,3 +156,15 @@ def us_task_reassignation(sender, instance, created, **kwargs):
|
|||
def us_tags_normalization(sender, instance, **kwargs):
|
||||
if isinstance(instance.tags, (list, tuple)):
|
||||
instance.tags = list(map(str.lower, instance.tags))
|
||||
|
||||
@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'])
|
||||
|
|
|
|||
|
|
@ -121,6 +121,14 @@ class UserStoryFactory(Factory):
|
|||
owner = factory.SubFactory("tests.factories.UserFactory")
|
||||
subject = factory.Sequence(lambda n: "User Story {}".format(n))
|
||||
description = factory.Sequence(lambda n: "User Story {} description".format(n))
|
||||
status = factory.SubFactory("tests.factories.UserStoryStatusFactory")
|
||||
|
||||
|
||||
class UserStoryStatusFactory(Factory):
|
||||
FACTORY_FOR = get_model("projects", "UserStoryStatus")
|
||||
|
||||
name = factory.Sequence(lambda n: "User Story status {}".format(n))
|
||||
project = factory.SubFactory("tests.factories.ProjectFactory")
|
||||
|
||||
|
||||
class TaskFactory(Factory):
|
||||
|
|
|
|||
|
|
@ -0,0 +1,187 @@
|
|||
# Copyright (C) 2014 Andrey Antukh <niwi@niwi.be>
|
||||
# Copyright (C) 2014 Jesús Espino <jespinog@gmail.com>
|
||||
# Copyright (C) 2014 David Barragán <bameda@dbarragan.com>
|
||||
# Copyright (C) 2014 Anler Hernández <hello@anler.me>
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as
|
||||
# published by the Free Software Foundation, either version 3 of the
|
||||
# License, or (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
#
|
||||
# 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 taiga.projects.userstories.models import UserStory
|
||||
|
||||
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
|
||||
|
||||
|
||||
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_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_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_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_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_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_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_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_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_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
|
||||
Loading…
Reference in New Issue