diff --git a/tests/integration/test_project_notifications.py b/tests/integration/test_project_notifications.py new file mode 100644 index 00000000..af71c25b --- /dev/null +++ b/tests/integration/test_project_notifications.py @@ -0,0 +1,229 @@ +import json +import pytest +from unittest.mock import MagicMock, patch + +from django.core.urlresolvers import reverse +from django.db.models.loading import get_model +from .. import factories as f + +from taiga.projects.notifications import services +from taiga.projects.notifications.choices import NotifyLevel +from taiga.projects.history.choices import HistoryType + + +pytestmark = pytest.mark.django_db + + +@pytest.fixture +def mail(): + from django.core import mail + mail.outbox = [] + return mail + + +def test_attach_notify_policy_to_project_queryset(): + project1 = f.ProjectFactory.create() + project2 = f.ProjectFactory.create() + + qs = project1.__class__.objects.order_by("id") + qs = services.attach_notify_policy_to_project_queryset(project1.owner, qs) + + assert len(qs) == 2 + assert qs[0].notify_level == NotifyLevel.notwatch + assert qs[1].notify_level == NotifyLevel.notwatch + + services.create_notify_policy(project1, project1.owner, NotifyLevel.watch) + qs = project1.__class__.objects.order_by("id") + qs = services.attach_notify_policy_to_project_queryset(project1.owner, qs) + assert qs[0].notify_level == NotifyLevel.watch + assert qs[1].notify_level == NotifyLevel.notwatch + + +def test_create_retrieve_notify_policy(): + project = f.ProjectFactory.create() + + policy_model_cls = get_model("notifications", "NotifyPolicy") + current_number = policy_model_cls.objects.all().count() + assert current_number == 0 + + policy = services.get_notify_policy(project, project.owner) + + current_number = policy_model_cls.objects.all().count() + assert current_number == 1 + assert policy.notify_level == NotifyLevel.notwatch + + +def test_notify_policy_existence(): + project = f.ProjectFactory.create() + assert not services.notify_policy_exists(project, project.owner) + + services.create_notify_policy(project, project.owner, NotifyLevel.watch) + assert services.notify_policy_exists(project, project.owner) + + +def test_analize_object_for_watchers(): + user1 = f.UserFactory.create() + user2 = f.UserFactory.create() + + issue = MagicMock() + issue.description = "Foo @{0} @{1} ".format(user1.username, + user2.username) + issue.content = "" + + history = MagicMock() + history.comment = "" + + services.analize_object_for_watchers(issue, history) + assert issue.watchers.add.call_count == 2 + + +def test_users_to_notify(): + project = f.ProjectFactory.create() + issue = f.IssueFactory.create(project=project) + + member1 = f.MembershipFactory.create(project=project) + member2 = f.MembershipFactory.create(project=project) + member3 = f.MembershipFactory.create(project=project) + + policy1 = services.create_notify_policy(project, member1.user) + policy2 = services.create_notify_policy(project, member2.user) + policy3 = services.create_notify_policy(project, member3.user) + + history = MagicMock() + history.owner = member2.user + history.comment = "" + + # Test basic description modifications + issue.description = "test1" + issue.save() + users = services.get_users_to_notify(issue, history=history) + assert len(users) == 1 + assert tuple(users)[0] == issue.get_owner() + + # Test watch notify level in one member + policy1.notify_level = NotifyLevel.watch + policy1.save() + + users = services.get_users_to_notify(issue, history=history) + assert len(users) == 2 + assert users == {member1.user, issue.get_owner()} + + # Test with watchers + issue.watchers.add(member3.user) + users = services.get_users_to_notify(issue, history=history) + assert len(users) == 3 + assert users == {member1.user, member3.user, issue.get_owner()} + + # Test with watchers with ignore policy + policy3.notify_level = NotifyLevel.ignore + policy3.save() + + issue.watchers.add(member3.user) + users = services.get_users_to_notify(issue, history=history) + assert len(users) == 2 + assert users == {member1.user, issue.get_owner()} + + +def test_send_notifications_using_services_method(mail): + project = f.ProjectFactory.create() + member1 = f.MembershipFactory.create(project=project) + member2 = f.MembershipFactory.create(project=project) + + history_change = MagicMock() + history_change.owner = member1.user + history_change.comment = "" + history_change.type = HistoryType.change + + history_create = MagicMock() + history_create.owner = member1.user + history_create.comment = "" + history_create.type = HistoryType.create + + history_delete = MagicMock() + history_delete.owner = member1.user + history_delete.comment = "" + history_delete.type = HistoryType.delete + + # Issues + issue = f.IssueFactory.create(project=project) + services.send_notifications(issue, + history=history_create, + users={member1.user, member2.user}) + + services.send_notifications(issue, + history=history_change, + users={member1.user, member2.user}) + + services.send_notifications(issue, + history=history_delete, + users={member1.user, member2.user}) + + # Userstories + us = f.UserStoryFactory.create() + services.send_notifications(us, + history=history_create, + users={member1.user, member2.user}) + + services.send_notifications(us, + history=history_change, + users={member1.user, member2.user}) + + services.send_notifications(us, + history=history_delete, + users={member1.user, member2.user}) + # Tasks + task = f.TaskFactory.create() + services.send_notifications(task, + history=history_create, + users={member1.user, member2.user}) + + services.send_notifications(task, + history=history_change, + users={member1.user, member2.user}) + + services.send_notifications(task, + history=history_delete, + users={member1.user, member2.user}) + + # Wiki pages + wiki = f.WikiPageFactory.create() + services.send_notifications(wiki, + history=history_create, + users={member1.user, member2.user}) + + services.send_notifications(wiki, + history=history_change, + users={member1.user, member2.user}) + + services.send_notifications(wiki, + history=history_delete, + users={member1.user, member2.user}) + + assert len(mail.outbox) == 24 + + + +def test_resource_notification_test(client, mail): + user1 = f.UserFactory.create() + user2 = f.UserFactory.create() + project = f.ProjectFactory.create(owner=user1) + role = f.RoleFactory.create(project=project) + member1 = f.MembershipFactory.create(project=project, user=user1, role=role) + member2 = f.MembershipFactory.create(project=project, user=user2, role=role) + issue = f.IssueFactory.create(owner=user2, project=project) + + mock_path = "taiga.projects.issues.api.IssueViewSet.pre_conditions_on_save" + url = reverse("issues-detail", args=[issue.pk]) + + client.login(user1) + + with patch(mock_path) as m: + data = {"subject": "Fooooo"} + response = client.patch(url, json.dumps(data), content_type="application/json") + assert len(mail.outbox) == 1 + assert response.status_code == 200 + + with patch(mock_path) as m: + response = client.delete(url) + assert response.status_code == 204 + assert len(mail.outbox) == 2