Improve webhooks (fix #3688, #3755 and #3758)

remotes/origin/issue/4795/notification_even_they_are_disabled
David Barragán Merino 2016-04-11 21:30:00 +02:00 committed by Alejandro Alonso
parent 40eaa7e119
commit 9f4886963a
12 changed files with 1489 additions and 132 deletions

View File

@ -4,7 +4,14 @@
## 2.1.0 ??? (unreleased) ## 2.1.0 ??? (unreleased)
### Features ### Features
- ... - Webhooks: Improve webhook data:
- add permalinks
- owner, assigned_to, status, type, priority, severity, user_story, milestone, project are objects
- add role to 'points' object
- add the owner to every notification ('by' field)
- add the date of the notification ('date' field)
- show human diffs in 'changes'
- remove unnecessary data
### Misc ### Misc
- Add sprint name and slug on search results for user stories ((thanks to [@everblut](https://github.com/everblut))) - Add sprint name and slug on search results for user stories ((thanks to [@everblut](https://github.com/everblut)))

View File

@ -22,8 +22,8 @@ from taiga.base.api.utils import encoders
import json import json
def dumps(data, ensure_ascii=True, encoder_class=encoders.JSONEncoder): def dumps(data, ensure_ascii=True, encoder_class=encoders.JSONEncoder, indent=None):
return json.dumps(data, cls=encoder_class, indent=None, ensure_ascii=ensure_ascii) return json.dumps(data, cls=encoder_class, ensure_ascii=ensure_ascii, indent=indent)
def loads(data): def loads(data):

View File

@ -15,6 +15,7 @@
# You should have received a copy of the GNU Affero General Public License # 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/>. # along with this program. If not, see <http://www.gnu.org/licenses/>.
from django.utils import timezone
from django.utils.translation import ugettext as _ from django.utils.translation import ugettext as _
from taiga.base import filters from taiga.base import filters
@ -45,7 +46,7 @@ class WebhookViewSet(BlockedByProjectMixin, ModelCrudViewSet):
self.check_permissions(request, 'test', webhook) self.check_permissions(request, 'test', webhook)
self.pre_conditions_blocked(webhook) self.pre_conditions_blocked(webhook)
webhooklog = tasks.test_webhook(webhook.id, webhook.url, webhook.key) webhooklog = tasks.test_webhook(webhook.id, webhook.url, webhook.key, request.user, timezone.now())
log = serializers.WebhookLogSerializer(webhooklog) log = serializers.WebhookLogSerializer(webhooklog)
return response.Ok(log.data) return response.Ok(log.data)

View File

@ -20,21 +20,26 @@ from django.core.exceptions import ObjectDoesNotExist
from taiga.base.api import serializers from taiga.base.api import serializers
from taiga.base.fields import TagsField, PgArrayField, JsonField from taiga.base.fields import TagsField, PgArrayField, JsonField
from taiga.projects.userstories import models as us_models from taiga.front.templatetags.functions import resolve as resolve_front_url
from taiga.projects.tasks import models as task_models
from taiga.projects.history import models as history_models
from taiga.projects.issues import models as issue_models from taiga.projects.issues import models as issue_models
from taiga.projects.milestones import models as milestone_models from taiga.projects.milestones import models as milestone_models
from taiga.projects.wiki import models as wiki_models
from taiga.projects.history import models as history_models
from taiga.projects.notifications.mixins import EditableWatchedResourceModelSerializer from taiga.projects.notifications.mixins import EditableWatchedResourceModelSerializer
from taiga.projects.services import get_logo_big_thumbnail_url
from taiga.projects.tasks import models as task_models
from taiga.projects.userstories import models as us_models
from taiga.projects.wiki import models as wiki_models
from taiga.users.gravatar import get_gravatar_url
from taiga.users.services import get_photo_or_gravatar_url
from .models import Webhook, WebhookLog from .models import Webhook, WebhookLog
class HistoryDiffField(serializers.Field): ########################################################################
def to_native(self, obj): ## WebHooks
return {key: {"from": value[0], "to": value[1]} for key, value in obj.items()} ########################################################################
class WebhookSerializer(serializers.ModelSerializer): class WebhookSerializer(serializers.ModelSerializer):
logs_counter = serializers.SerializerMethodField("get_logs_counter") logs_counter = serializers.SerializerMethodField("get_logs_counter")
@ -55,16 +60,93 @@ class WebhookLogSerializer(serializers.ModelSerializer):
model = WebhookLog model = WebhookLog
########################################################################
## User
########################################################################
class UserSerializer(serializers.Serializer): class UserSerializer(serializers.Serializer):
id = serializers.SerializerMethodField("get_pk") id = serializers.SerializerMethodField("get_pk")
name = serializers.SerializerMethodField("get_name") permalink = serializers.SerializerMethodField("get_permalink")
gravatar_url = serializers.SerializerMethodField("get_gravatar_url")
username = serializers.SerializerMethodField("get_username")
full_name = serializers.SerializerMethodField("get_full_name")
photo = serializers.SerializerMethodField("get_photo")
def get_pk(self, obj): def get_pk(self, obj):
return obj.pk return obj.pk
def get_name(self, obj): def get_permalink(self, obj):
return obj.full_name return resolve_front_url("user", obj.username)
def get_gravatar_url(self, obj):
return get_gravatar_url(obj.email)
def get_username(self, obj):
return obj.get_username
def get_full_name(self, obj):
return obj.get_full_name()
def get_photo(self, obj):
return get_photo_or_gravatar_url(obj)
########################################################################
## Project
########################################################################
class ProjectSerializer(serializers.Serializer):
id = serializers.SerializerMethodField("get_pk")
permalink = serializers.SerializerMethodField("get_permalink")
name = serializers.SerializerMethodField("get_name")
logo_big_url = serializers.SerializerMethodField("get_logo_big_url")
def get_pk(self, obj):
return obj.pk
def get_permalink(self, obj):
return resolve_front_url("project", obj.slug)
def get_name(self, obj):
return obj.name
def get_logo_big_url(self, obj):
return get_logo_big_thumbnail_url(obj)
########################################################################
## History Serializer
########################################################################
class HistoryDiffField(serializers.Field):
def to_native(self, value):
# Tip: 'value' is the object returned by
# taiga.projects.history.models.HistoryEntry.values_diff()
ret = {}
for key, val in value.items():
if key in ["attachments", "custom_attributes"]:
ret[key] = val
elif key == "points":
ret[key] = {k: {"from": v[0], "to": v[1]} for k, v in val.items()}
else:
ret[key] = {"from": val[0], "to": val[1]}
return ret
class HistoryEntrySerializer(serializers.ModelSerializer):
diff = HistoryDiffField(source="values_diff")
class Meta:
model = history_models.HistoryEntry
exclude = ("id", "type", "key", "is_hidden", "is_snapshot", "snapshot", "user", "delete_comment_user",
"values", "created_at")
########################################################################
## _Misc_
########################################################################
class CustomAttributesValuesWebhookSerializerMixin(serializers.ModelSerializer): class CustomAttributesValuesWebhookSerializerMixin(serializers.ModelSerializer):
custom_attributes_values = serializers.SerializerMethodField("get_custom_attributes_values") custom_attributes_values = serializers.SerializerMethodField("get_custom_attributes_values")
@ -90,86 +172,251 @@ class CustomAttributesValuesWebhookSerializerMixin(serializers.ModelSerializer):
except ObjectDoesNotExist: except ObjectDoesNotExist:
return None return None
class PointSerializer(serializers.Serializer):
id = serializers.SerializerMethodField("get_pk") class RolePointsSerializer(serializers.Serializer):
role = serializers.SerializerMethodField("get_role")
name = serializers.SerializerMethodField("get_name") name = serializers.SerializerMethodField("get_name")
value = serializers.SerializerMethodField("get_value") value = serializers.SerializerMethodField("get_value")
def get_role(self, obj):
return obj.role.name
def get_name(self, obj):
return obj.points.name
def get_value(self, obj):
return obj.points.value
class UserStoryStatusSerializer(serializers.Serializer):
id = serializers.SerializerMethodField("get_pk")
name = serializers.SerializerMethodField("get_name")
slug = serializers.SerializerMethodField("get_slug")
color = serializers.SerializerMethodField("get_color")
is_closed = serializers.SerializerMethodField("get_is_closed")
is_archived = serializers.SerializerMethodField("get_is_archived")
def get_pk(self, obj): def get_pk(self, obj):
return obj.pk return obj.pk
def get_name(self, obj): def get_name(self, obj):
return obj.name return obj.name
def get_value(self, obj): def get_slug(self, obj):
return obj.value return obj.slug
def get_color(self, obj):
return obj.color
def get_is_closed(self, obj):
return obj.is_closed
def get_is_archived(self, obj):
return obj.is_archived
class UserStorySerializer(CustomAttributesValuesWebhookSerializerMixin, EditableWatchedResourceModelSerializer, class TaskStatusSerializer(serializers.Serializer):
serializers.ModelSerializer): id = serializers.SerializerMethodField("get_pk")
tags = TagsField(default=[], required=False) name = serializers.SerializerMethodField("get_name")
external_reference = PgArrayField(required=False) slug = serializers.SerializerMethodField("get_slug")
owner = UserSerializer() color = serializers.SerializerMethodField("get_color")
assigned_to = UserSerializer() is_closed = serializers.SerializerMethodField("get_is_closed")
points = PointSerializer(many=True)
class Meta: def get_pk(self, obj):
model = us_models.UserStory return obj.pk
exclude = ("backlog_order", "sprint_order", "kanban_order", "version")
def custom_attributes_queryset(self, project): def get_name(self, obj):
return project.userstorycustomattributes.all() return obj.name
def get_slug(self, obj):
return obj.slug
def get_color(self, obj):
return obj.color
def get_is_closed(self, obj):
return obj.is_closed
class TaskSerializer(CustomAttributesValuesWebhookSerializerMixin, EditableWatchedResourceModelSerializer, class IssueStatusSerializer(serializers.Serializer):
serializers.ModelSerializer): id = serializers.SerializerMethodField("get_pk")
tags = TagsField(default=[], required=False) name = serializers.SerializerMethodField("get_name")
owner = UserSerializer() slug = serializers.SerializerMethodField("get_slug")
assigned_to = UserSerializer() color = serializers.SerializerMethodField("get_color")
is_closed = serializers.SerializerMethodField("get_is_closed")
class Meta: def get_pk(self, obj):
model = task_models.Task return obj.pk
def custom_attributes_queryset(self, project): def get_name(self, obj):
return project.taskcustomattributes.all() return obj.name
def get_slug(self, obj):
return obj.slug
def get_color(self, obj):
return obj.color
def get_is_closed(self, obj):
return obj.is_closed
class IssueSerializer(CustomAttributesValuesWebhookSerializerMixin, EditableWatchedResourceModelSerializer, class IssueTypeSerializer(serializers.Serializer):
serializers.ModelSerializer): id = serializers.SerializerMethodField("get_pk")
tags = TagsField(default=[], required=False) name = serializers.SerializerMethodField("get_name")
owner = UserSerializer() color = serializers.SerializerMethodField("get_color")
assigned_to = UserSerializer()
class Meta: def get_pk(self, obj):
model = issue_models.Issue return obj.pk
def custom_attributes_queryset(self, project): def get_name(self, obj):
return project.issuecustomattributes.all() return obj.name
def get_color(self, obj):
return obj.color
class WikiPageSerializer(serializers.ModelSerializer): class PrioritySerializer(serializers.Serializer):
owner = UserSerializer() id = serializers.SerializerMethodField("get_pk")
last_modifier = UserSerializer() name = serializers.SerializerMethodField("get_name")
color = serializers.SerializerMethodField("get_color")
class Meta: def get_pk(self, obj):
model = wiki_models.WikiPage return obj.pk
exclude = ("watchers", "version")
def get_name(self, obj):
return obj.name
def get_color(self, obj):
return obj.color
class SeveritySerializer(serializers.Serializer):
id = serializers.SerializerMethodField("get_pk")
name = serializers.SerializerMethodField("get_name")
color = serializers.SerializerMethodField("get_color")
def get_pk(self, obj):
return obj.pk
def get_name(self, obj):
return obj.name
def get_color(self, obj):
return obj.color
########################################################################
## Milestone
########################################################################
class MilestoneSerializer(serializers.ModelSerializer): class MilestoneSerializer(serializers.ModelSerializer):
permalink = serializers.SerializerMethodField("get_permalink")
project = ProjectSerializer()
owner = UserSerializer() owner = UserSerializer()
class Meta: class Meta:
model = milestone_models.Milestone model = milestone_models.Milestone
exclude = ("order", "watchers") exclude = ("order", "watchers")
def get_permalink(self, obj):
return resolve_front_url("taskboard", obj.project.slug, obj.slug)
class HistoryEntrySerializer(serializers.ModelSerializer):
diff = HistoryDiffField() ########################################################################
snapshot = JsonField() ## User Story
values = JsonField() ########################################################################
user = JsonField()
delete_comment_user = JsonField() class UserStorySerializer(CustomAttributesValuesWebhookSerializerMixin, EditableWatchedResourceModelSerializer,
serializers.ModelSerializer):
permalink = serializers.SerializerMethodField("get_permalink")
tags = TagsField(default=[], required=False)
external_reference = PgArrayField(required=False)
project = ProjectSerializer()
owner = UserSerializer()
assigned_to = UserSerializer()
points = RolePointsSerializer(source="role_points", many=True)
status = UserStoryStatusSerializer()
milestone = MilestoneSerializer()
class Meta: class Meta:
model = history_models.HistoryEntry model = us_models.UserStory
exclude = ("backlog_order", "sprint_order", "kanban_order", "version", "total_watchers", "is_watcher")
def get_permalink(self, obj):
return resolve_front_url("userstory", obj.project.slug, obj.ref)
def custom_attributes_queryset(self, project):
return project.userstorycustomattributes.all()
########################################################################
## Task
########################################################################
class TaskSerializer(CustomAttributesValuesWebhookSerializerMixin, EditableWatchedResourceModelSerializer,
serializers.ModelSerializer):
permalink = serializers.SerializerMethodField("get_permalink")
tags = TagsField(default=[], required=False)
project = ProjectSerializer()
owner = UserSerializer()
assigned_to = UserSerializer()
status = TaskStatusSerializer()
user_story = UserStorySerializer()
milestone = MilestoneSerializer()
class Meta:
model = task_models.Task
exclude = ("version", "total_watchers", "is_watcher")
def get_permalink(self, obj):
return resolve_front_url("task", obj.project.slug, obj.ref)
def custom_attributes_queryset(self, project):
return project.taskcustomattributes.all()
########################################################################
## Issue
########################################################################
class IssueSerializer(CustomAttributesValuesWebhookSerializerMixin, EditableWatchedResourceModelSerializer,
serializers.ModelSerializer):
permalink = serializers.SerializerMethodField("get_permalink")
tags = TagsField(default=[], required=False)
project = ProjectSerializer()
milestone = MilestoneSerializer()
owner = UserSerializer()
assigned_to = UserSerializer()
status = IssueStatusSerializer()
type = IssueTypeSerializer()
priority = PrioritySerializer()
severity = SeveritySerializer()
class Meta:
model = issue_models.Issue
exclude = ("version", "total_watchers", "is_watcher")
def get_permalink(self, obj):
return resolve_front_url("issue", obj.project.slug, obj.ref)
def custom_attributes_queryset(self, project):
return project.issuecustomattributes.all()
########################################################################
## Wiki Page
########################################################################
class WikiPageSerializer(serializers.ModelSerializer):
permalink = serializers.SerializerMethodField("get_permalink")
project = ProjectSerializer()
owner = UserSerializer()
last_modifier = UserSerializer()
class Meta:
model = wiki_models.WikiPage
exclude = ("watchers", "total_watchers", "is_watcher", "version")
def get_permalink(self, obj):
return resolve_front_url("wiki", obj.project.slug, obj.slug)

View File

@ -61,10 +61,13 @@ def on_new_history_entry(sender, instance, created, **kwargs):
extra_args = [instance] extra_args = [instance]
elif instance.type == HistoryType.delete: elif instance.type == HistoryType.delete:
task = tasks.delete_webhook task = tasks.delete_webhook
extra_args = [timezone.now()] extra_args = []
by = instance.owner
date = timezone.now()
for webhook in webhooks: for webhook in webhooks:
args = [webhook["id"], webhook["url"], webhook["key"], obj] + extra_args args = [webhook["id"], webhook["url"], webhook["key"], by, date, obj] + extra_args
if settings.CELERY_ENABLED: if settings.CELERY_ENABLED:
connection.on_commit(lambda: task.delay(*args)) connection.on_commit(lambda: task.delay(*args))

View File

@ -26,7 +26,7 @@ from taiga.celery import app
from .serializers import (UserStorySerializer, IssueSerializer, TaskSerializer, from .serializers import (UserStorySerializer, IssueSerializer, TaskSerializer,
WikiPageSerializer, MilestoneSerializer, WikiPageSerializer, MilestoneSerializer,
HistoryEntrySerializer) HistoryEntrySerializer, UserSerializer)
from .models import WebhookLog from .models import WebhookLog
@ -67,58 +67,70 @@ def _send_request(webhook_id, url, key, data):
request = requests.Request('POST', url, data=serialized_data, headers=headers) request = requests.Request('POST', url, data=serialized_data, headers=headers)
prepared_request = request.prepare() prepared_request = request.prepare()
session = requests.Session() with requests.Session() as session:
try: try:
response = session.send(prepared_request) response = session.send(prepared_request)
webhook_log = WebhookLog.objects.create(webhook_id=webhook_id, url=url, except RequestException as e:
status=response.status_code, # Error sending the webhook
request_data=data, webhook_log = WebhookLog.objects.create(webhook_id=webhook_id, url=url, status=0,
request_headers=dict(prepared_request.headers), request_data=data,
response_data=response.content, request_headers=dict(prepared_request.headers),
response_headers=dict(response.headers), response_data="error-in-request: {}".format(str(e)),
duration=response.elapsed.total_seconds()) response_headers={},
except RequestException as e: duration=0)
webhook_log = WebhookLog.objects.create(webhook_id=webhook_id, url=url, status=0, else:
request_data=data, # Webhook was sent successfully
request_headers=dict(prepared_request.headers), webhook_log = WebhookLog.objects.create(webhook_id=webhook_id, url=url,
response_data="error-in-request: {}".format(str(e)), status=response.status_code,
response_headers={}, request_data=data,
duration=0) request_headers=dict(prepared_request.headers),
session.close() response_data=response.content,
response_headers=dict(response.headers),
duration=response.elapsed.total_seconds())
finally:
# Only the last ten webhook logs traces are required
# so remove the leftover
ids = (WebhookLog.objects.filter(webhook_id=webhook_id)
.order_by("-id")
.values_list('id', flat=True)[10:])
WebhookLog.objects.filter(id__in=ids).delete()
ids = [log.id for log in WebhookLog.objects.filter(webhook_id=webhook_id).order_by("-id")[10:]]
WebhookLog.objects.filter(id__in=ids).delete()
return webhook_log return webhook_log
@app.task @app.task
def change_webhook(webhook_id, url, key, obj, change): def create_webhook(webhook_id, url, key, by, date, obj):
data = {} data = {}
data['data'] = _serialize(obj)
data['action'] = "change"
data['type'] = _get_type(obj)
data['change'] = _serialize(change)
return _send_request(webhook_id, url, key, data)
@app.task
def create_webhook(webhook_id, url, key, obj):
data = {}
data['data'] = _serialize(obj)
data['action'] = "create" data['action'] = "create"
data['type'] = _get_type(obj) data['type'] = _get_type(obj)
data['by'] = UserSerializer(by).data
data['date'] = date
data['data'] = _serialize(obj)
return _send_request(webhook_id, url, key, data) return _send_request(webhook_id, url, key, data)
@app.task @app.task
def delete_webhook(webhook_id, url, key, obj, deleted_date): def delete_webhook(webhook_id, url, key, by, date, obj):
data = {} data = {}
data['data'] = _serialize(obj)
data['action'] = "delete" data['action'] = "delete"
data['type'] = _get_type(obj) data['type'] = _get_type(obj)
data['deleted_date'] = deleted_date data['by'] = UserSerializer(by).data
data['date'] = date
data['data'] = _serialize(obj)
return _send_request(webhook_id, url, key, data)
@app.task
def change_webhook(webhook_id, url, key, by, date, obj, change):
data = {}
data['action'] = "change"
data['type'] = _get_type(obj)
data['by'] = UserSerializer(by).data
data['date'] = date
data['data'] = _serialize(obj)
data['change'] = _serialize(change)
return _send_request(webhook_id, url, key, data) return _send_request(webhook_id, url, key, data)
@ -129,10 +141,12 @@ def resend_webhook(webhook_id, url, key, data):
@app.task @app.task
def test_webhook(webhook_id, url, key): def test_webhook(webhook_id, url, key, by, date):
data = {} data = {}
data['data'] = {"test": "test"}
data['action'] = "test" data['action'] = "test"
data['type'] = "test" data['type'] = "test"
data['by'] = UserSerializer(by).data
data['date'] = date
data['data'] = {"test": "test"}
return _send_request(webhook_id, url, key, data) return _send_request(webhook_id, url, key, data)

View File

@ -0,0 +1,248 @@
# Copyright (C) 2014-2016 Andrey Antukh <niwi@niwi.nz>
# Copyright (C) 2014-2016 Jesús Espino <jespinog@gmail.com>
# Copyright (C) 2014-2016 David Barragán <bameda@dbarragan.com>
# Copyright (C) 2014-2016 Alejandro Alonso <alejandro.alonso@kaleidos.net>
# Copyright (C) 2014-2016 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/>.
import pytest
from unittest.mock import patch
from unittest.mock import Mock
from .. import factories as f
from taiga.projects.history import services
pytestmark = pytest.mark.django_db(transaction=True)
from taiga.base.utils import json
def test_webhooks_when_create_issue(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.IssueFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "create"
assert data["type"] == "issue"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
def test_webhooks_when_update_issue(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.IssueFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
obj.subject = "test webhook update"
obj.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "issue"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["data"]["subject"] == obj.subject
assert data["change"]["comment"] == "test_comment"
assert data["change"]["diff"]["subject"]["to"] == data["data"]["subject"]
assert data["change"]["diff"]["subject"]["from"] != data["data"]["subject"]
def test_webhooks_when_delete_issue(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.IssueFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, delete=True)
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "delete"
assert data["type"] == "issue"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert "data" in data
def test_webhooks_when_update_issue_attachments(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.IssueFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
# Create attachments
attachment1 = f.IssueAttachmentFactory(project=obj.project, content_object=obj, owner=obj.owner)
attachment2 = f.IssueAttachmentFactory(project=obj.project, content_object=obj, owner=obj.owner)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "issue"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["attachments"]["new"]) == 2
assert len(data["change"]["diff"]["attachments"]["changed"]) == 0
assert len(data["change"]["diff"]["attachments"]["deleted"]) == 0
# Update attachment
attachment1.description = "new attachment description"
attachment1.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "issue"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["attachments"]["new"]) == 0
assert len(data["change"]["diff"]["attachments"]["changed"]) == 1
assert len(data["change"]["diff"]["attachments"]["deleted"]) == 0
# Delete attachment
attachment2.delete()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "issue"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["attachments"]["new"]) == 0
assert len(data["change"]["diff"]["attachments"]["changed"]) == 0
assert len(data["change"]["diff"]["attachments"]["deleted"]) == 1
def test_webhooks_when_update_issue_custom_attributes(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.IssueFactory.create(project=project)
custom_attr_1 = f.IssueCustomAttributeFactory(project=obj.project)
ct1_id = "{}".format(custom_attr_1.id)
custom_attr_2 = f.IssueCustomAttributeFactory(project=obj.project)
ct2_id = "{}".format(custom_attr_2.id)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
# Create custom attributes
obj.custom_attributes_values.attributes_values = {
ct1_id: "test_1_updated",
ct2_id: "test_2_updated"
}
obj.custom_attributes_values.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "issue"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["custom_attributes"]["new"]) == 2
assert len(data["change"]["diff"]["custom_attributes"]["changed"]) == 0
assert len(data["change"]["diff"]["custom_attributes"]["deleted"]) == 0
# Update custom attributes
obj.custom_attributes_values.attributes_values[ct1_id] = "test_2_updated"
obj.custom_attributes_values.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "issue"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["custom_attributes"]["new"]) == 0
assert len(data["change"]["diff"]["custom_attributes"]["changed"]) == 1
assert len(data["change"]["diff"]["custom_attributes"]["deleted"]) == 0
# Delete custom attributes
del obj.custom_attributes_values.attributes_values[ct1_id]
obj.custom_attributes_values.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "issue"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["custom_attributes"]["new"]) == 0
assert len(data["change"]["diff"]["custom_attributes"]["changed"]) == 0
assert len(data["change"]["diff"]["custom_attributes"]["deleted"]) == 1

View File

@ -0,0 +1,101 @@
# Copyright (C) 2014-2016 Andrey Antukh <niwi@niwi.nz>
# Copyright (C) 2014-2016 Jesús Espino <jespinog@gmail.com>
# Copyright (C) 2014-2016 David Barragán <bameda@dbarragan.com>
# Copyright (C) 2014-2016 Alejandro Alonso <alejandro.alonso@kaleidos.net>
# Copyright (C) 2014-2016 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/>.
import pytest
from unittest.mock import patch
from unittest.mock import Mock
from .. import factories as f
from taiga.projects.history import services
pytestmark = pytest.mark.django_db(transaction=True)
from taiga.base.utils import json
def test_webhooks_when_create_milestone(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.MilestoneFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "create"
assert data["type"] == "milestone"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
def test_webhooks_when_update_milestone(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.MilestoneFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
obj.name = "test webhook update"
obj.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "milestone"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["data"]["name"] == obj.name
assert data["change"]["comment"] == "test_comment"
assert data["change"]["diff"]["name"]["to"] == data["data"]["name"]
assert data["change"]["diff"]["name"]["from"] != data["data"]["name"]
def test_webhooks_when_delete_milestone(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.MilestoneFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, delete=True)
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "delete"
assert data["type"] == "milestone"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert "data" in data

View File

@ -18,6 +18,7 @@
import pytest import pytest
from unittest.mock import patch from unittest.mock import patch
from unittest.mock import Mock
from .. import factories as f from .. import factories as f
@ -26,7 +27,7 @@ from taiga.projects.history import services
pytestmark = pytest.mark.django_db(transaction=True) pytestmark = pytest.mark.django_db(transaction=True)
def test_new_object_with_one_webhook(settings): def test_new_object_with_one_webhook_signal(settings):
settings.WEBHOOKS_ENABLED = True settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory() project = f.ProjectFactory()
f.WebhookFactory.create(project=project) f.WebhookFactory.create(project=project)
@ -38,28 +39,31 @@ def test_new_object_with_one_webhook(settings):
f.WikiPageFactory.create(project=project) f.WikiPageFactory.create(project=project)
] ]
for obj in objects: response = Mock(status_code=200, headers={}, content="ok")
with patch('taiga.webhooks.tasks.create_webhook') as create_webhook_mock: response.elapsed.total_seconds.return_value = 100
services.take_snapshot(obj, user=obj.owner, comment="test")
assert create_webhook_mock.call_count == 1
for obj in objects: for obj in objects:
with patch('taiga.webhooks.tasks.change_webhook') as change_webhook_mock: with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
services.take_snapshot(obj, user=obj.owner, comment="test")
assert session_send_mock.call_count == 1
for obj in objects:
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
services.take_snapshot(obj, user=obj.owner) services.take_snapshot(obj, user=obj.owner)
assert change_webhook_mock.call_count == 0 assert session_send_mock.call_count == 0
for obj in objects: for obj in objects:
with patch('taiga.webhooks.tasks.change_webhook') as change_webhook_mock: with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
services.take_snapshot(obj, user=obj.owner, comment="test") services.take_snapshot(obj, user=obj.owner, comment="test")
assert change_webhook_mock.call_count == 1 assert session_send_mock.call_count == 1
for obj in objects: for obj in objects:
with patch('taiga.webhooks.tasks.delete_webhook') as delete_webhook_mock: with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
services.take_snapshot(obj, user=obj.owner, comment="test", delete=True) services.take_snapshot(obj, user=obj.owner, comment="test", delete=True)
assert delete_webhook_mock.call_count == 1 assert session_send_mock.call_count == 1
def test_new_object_with_two_webhook(settings): def test_new_object_with_two_webhook_signals(settings):
settings.WEBHOOKS_ENABLED = True settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory() project = f.ProjectFactory()
f.WebhookFactory.create(project=project) f.WebhookFactory.create(project=project)
@ -72,28 +76,31 @@ def test_new_object_with_two_webhook(settings):
f.WikiPageFactory.create(project=project) f.WikiPageFactory.create(project=project)
] ]
for obj in objects: response = Mock(status_code=200, headers={}, content="ok")
with patch('taiga.webhooks.tasks.create_webhook') as create_webhook_mock: response.elapsed.total_seconds.return_value = 100
services.take_snapshot(obj, user=obj.owner, comment="test")
assert create_webhook_mock.call_count == 2
for obj in objects: for obj in objects:
with patch('taiga.webhooks.tasks.change_webhook') as change_webhook_mock: with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
services.take_snapshot(obj, user=obj.owner, comment="test") services.take_snapshot(obj, user=obj.owner, comment="test")
assert change_webhook_mock.call_count == 2 assert session_send_mock.call_count == 2
for obj in objects: for obj in objects:
with patch('taiga.webhooks.tasks.change_webhook') as change_webhook_mock: with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
services.take_snapshot(obj, user=obj.owner, comment="test")
assert session_send_mock.call_count == 2
for obj in objects:
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
services.take_snapshot(obj, user=obj.owner) services.take_snapshot(obj, user=obj.owner)
assert change_webhook_mock.call_count == 0 assert session_send_mock.call_count == 0
for obj in objects: for obj in objects:
with patch('taiga.webhooks.tasks.delete_webhook') as delete_webhook_mock: with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
services.take_snapshot(obj, user=obj.owner, comment="test", delete=True) services.take_snapshot(obj, user=obj.owner, comment="test", delete=True)
assert delete_webhook_mock.call_count == 2 assert session_send_mock.call_count == 2
def test_send_request_one_webhook(settings): def test_send_request_one_webhook_signal(settings):
settings.WEBHOOKS_ENABLED = True settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory() project = f.ProjectFactory()
f.WebhookFactory.create(project=project) f.WebhookFactory.create(project=project)
@ -105,12 +112,15 @@ def test_send_request_one_webhook(settings):
f.WikiPageFactory.create(project=project) f.WikiPageFactory.create(project=project)
] ]
for obj in objects: response = Mock(status_code=200, headers={}, content="ok")
with patch('taiga.webhooks.tasks._send_request') as _send_request_mock: response.elapsed.total_seconds.return_value = 100
services.take_snapshot(obj, user=obj.owner, comment="test")
assert _send_request_mock.call_count == 1
for obj in objects: for obj in objects:
with patch('taiga.webhooks.tasks._send_request') as _send_request_mock: with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
services.take_snapshot(obj, user=obj.owner, comment="test")
assert session_send_mock.call_count == 1
for obj in objects:
with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock:
services.take_snapshot(obj, user=obj.owner, comment="test", delete=True) services.take_snapshot(obj, user=obj.owner, comment="test", delete=True)
assert _send_request_mock.call_count == 1 assert session_send_mock.call_count == 1

View File

@ -0,0 +1,248 @@
# Copyright (C) 2014-2016 Andrey Antukh <niwi@niwi.nz>
# Copyright (C) 2014-2016 Jesús Espino <jespinog@gmail.com>
# Copyright (C) 2014-2016 David Barragán <bameda@dbarragan.com>
# Copyright (C) 2014-2016 Alejandro Alonso <alejandro.alonso@kaleidos.net>
# Copyright (C) 2014-2016 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/>.
import pytest
from unittest.mock import patch
from unittest.mock import Mock
from .. import factories as f
from taiga.projects.history import services
pytestmark = pytest.mark.django_db(transaction=True)
from taiga.base.utils import json
def test_webhooks_when_create_task(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.TaskFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "create"
assert data["type"] == "task"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
def test_webhooks_when_update_task(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.TaskFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
obj.subject = "test webhook update"
obj.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "task"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["data"]["subject"] == obj.subject
assert data["change"]["comment"] == "test_comment"
assert data["change"]["diff"]["subject"]["to"] == data["data"]["subject"]
assert data["change"]["diff"]["subject"]["from"] != data["data"]["subject"]
def test_webhooks_when_delete_task(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.TaskFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, delete=True)
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "delete"
assert data["type"] == "task"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert "data" in data
def test_webhooks_when_update_task_attachments(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.TaskFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
# Create attachments
attachment1 = f.TaskAttachmentFactory(project=obj.project, content_object=obj, owner=obj.owner)
attachment2 = f.TaskAttachmentFactory(project=obj.project, content_object=obj, owner=obj.owner)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "task"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["attachments"]["new"]) == 2
assert len(data["change"]["diff"]["attachments"]["changed"]) == 0
assert len(data["change"]["diff"]["attachments"]["deleted"]) == 0
# Update attachment
attachment1.description = "new attachment description"
attachment1.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "task"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["attachments"]["new"]) == 0
assert len(data["change"]["diff"]["attachments"]["changed"]) == 1
assert len(data["change"]["diff"]["attachments"]["deleted"]) == 0
# Delete attachment
attachment2.delete()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "task"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["attachments"]["new"]) == 0
assert len(data["change"]["diff"]["attachments"]["changed"]) == 0
assert len(data["change"]["diff"]["attachments"]["deleted"]) == 1
def test_webhooks_when_update_task_custom_attributes(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.TaskFactory.create(project=project)
custom_attr_1 = f.TaskCustomAttributeFactory(project=obj.project)
ct1_id = "{}".format(custom_attr_1.id)
custom_attr_2 = f.TaskCustomAttributeFactory(project=obj.project)
ct2_id = "{}".format(custom_attr_2.id)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
# Create custom attributes
obj.custom_attributes_values.attributes_values = {
ct1_id: "test_1_updated",
ct2_id: "test_2_updated"
}
obj.custom_attributes_values.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "task"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["custom_attributes"]["new"]) == 2
assert len(data["change"]["diff"]["custom_attributes"]["changed"]) == 0
assert len(data["change"]["diff"]["custom_attributes"]["deleted"]) == 0
# Update custom attributes
obj.custom_attributes_values.attributes_values[ct1_id] = "test_2_updated"
obj.custom_attributes_values.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "task"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["custom_attributes"]["new"]) == 0
assert len(data["change"]["diff"]["custom_attributes"]["changed"]) == 1
assert len(data["change"]["diff"]["custom_attributes"]["deleted"]) == 0
# Delete custom attributes
del obj.custom_attributes_values.attributes_values[ct1_id]
obj.custom_attributes_values.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "task"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["custom_attributes"]["new"]) == 0
assert len(data["change"]["diff"]["custom_attributes"]["changed"]) == 0
assert len(data["change"]["diff"]["custom_attributes"]["deleted"]) == 1

View File

@ -0,0 +1,308 @@
# Copyright (C) 2014-2016 Andrey Antukh <niwi@niwi.nz>
# Copyright (C) 2014-2016 Jesús Espino <jespinog@gmail.com>
# Copyright (C) 2014-2016 David Barragán <bameda@dbarragan.com>
# Copyright (C) 2014-2016 Alejandro Alonso <alejandro.alonso@kaleidos.net>
# Copyright (C) 2014-2016 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/>.
import pytest
from unittest.mock import patch
from unittest.mock import Mock
from .. import factories as f
from taiga.projects.history import services
pytestmark = pytest.mark.django_db(transaction=True)
from taiga.base.utils import json
def test_webhooks_when_create_user_story(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.UserStoryFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "create"
assert data["type"] == "userstory"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
def test_webhooks_when_update_user_story(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.UserStoryFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
obj.subject = "test webhook update"
obj.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "userstory"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["data"]["subject"] == obj.subject
assert data["change"]["comment"] == "test_comment"
assert data["change"]["diff"]["subject"]["to"] == data["data"]["subject"]
assert data["change"]["diff"]["subject"]["from"] != data["data"]["subject"]
def test_webhooks_when_delete_user_story(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.UserStoryFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, delete=True)
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "delete"
assert data["type"] == "userstory"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert "data" in data
def test_webhooks_when_update_user_story_attachments(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.UserStoryFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
# Create attachments
attachment1 = f.UserStoryAttachmentFactory(project=obj.project, content_object=obj, owner=obj.owner)
attachment2 = f.UserStoryAttachmentFactory(project=obj.project, content_object=obj, owner=obj.owner)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "userstory"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["attachments"]["new"]) == 2
assert len(data["change"]["diff"]["attachments"]["changed"]) == 0
assert len(data["change"]["diff"]["attachments"]["deleted"]) == 0
# Update attachment
attachment1.description = "new attachment description"
attachment1.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "userstory"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["attachments"]["new"]) == 0
assert len(data["change"]["diff"]["attachments"]["changed"]) == 1
assert len(data["change"]["diff"]["attachments"]["deleted"]) == 0
# Delete attachment
attachment2.delete()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "userstory"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["attachments"]["new"]) == 0
assert len(data["change"]["diff"]["attachments"]["changed"]) == 0
assert len(data["change"]["diff"]["attachments"]["deleted"]) == 1
def test_webhooks_when_update_user_story_custom_attributes(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.UserStoryFactory.create(project=project)
custom_attr_1 = f.UserStoryCustomAttributeFactory(project=obj.project)
ct1_id = "{}".format(custom_attr_1.id)
custom_attr_2 = f.UserStoryCustomAttributeFactory(project=obj.project)
ct2_id = "{}".format(custom_attr_2.id)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
# Create custom attributes
obj.custom_attributes_values.attributes_values = {
ct1_id: "test_1_updated",
ct2_id: "test_2_updated"
}
obj.custom_attributes_values.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "userstory"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["custom_attributes"]["new"]) == 2
assert len(data["change"]["diff"]["custom_attributes"]["changed"]) == 0
assert len(data["change"]["diff"]["custom_attributes"]["deleted"]) == 0
# Update custom attributes
obj.custom_attributes_values.attributes_values[ct1_id] = "test_2_updated"
obj.custom_attributes_values.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "userstory"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["custom_attributes"]["new"]) == 0
assert len(data["change"]["diff"]["custom_attributes"]["changed"]) == 1
assert len(data["change"]["diff"]["custom_attributes"]["deleted"]) == 0
# Delete custom attributes
del obj.custom_attributes_values.attributes_values[ct1_id]
obj.custom_attributes_values.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "userstory"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["custom_attributes"]["new"]) == 0
assert len(data["change"]["diff"]["custom_attributes"]["changed"]) == 0
assert len(data["change"]["diff"]["custom_attributes"]["deleted"]) == 1
def test_webhooks_when_update_user_story_points(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
role1 = f.RoleFactory.create(project=project)
role2 = f.RoleFactory.create(project=project)
points1 = f.PointsFactory.create(project=project, value=None)
points2 = f.PointsFactory.create(project=project, value=1)
points3 = f.PointsFactory.create(project=project, value=2)
obj = f.UserStoryFactory.create(project=project)
obj.role_points.all().delete()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
# Set points
f.RolePointsFactory.create(user_story=obj, role=role1, points=points1)
f.RolePointsFactory.create(user_story=obj, role=role2, points=points2)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "userstory"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == ""
assert data["change"]["diff"]["points"][role1.name]["from"] == None
assert data["change"]["diff"]["points"][role1.name]["to"] == points1.name
assert data["change"]["diff"]["points"][role2.name]["from"] == None
assert data["change"]["diff"]["points"][role2.name]["to"] == points2.name
# Change points
obj.role_points.all().update(points=points3)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "userstory"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == ""
assert data["change"]["diff"]["points"][role1.name]["from"] == points1.name
assert data["change"]["diff"]["points"][role1.name]["to"] == points3.name
assert data["change"]["diff"]["points"][role2.name]["from"] == points2.name
assert data["change"]["diff"]["points"][role2.name]["to"] == points3.name

View File

@ -0,0 +1,170 @@
# Copyright (C) 2014-2016 Andrey Antukh <niwi@niwi.nz>
# Copyright (C) 2014-2016 Jesús Espino <jespinog@gmail.com>
# Copyright (C) 2014-2016 David Barragán <bameda@dbarragan.com>
# Copyright (C) 2014-2016 Alejandro Alonso <alejandro.alonso@kaleidos.net>
# Copyright (C) 2014-2016 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/>.
import pytest
from unittest.mock import patch
from unittest.mock import Mock
from .. import factories as f
from taiga.projects.history import services
pytestmark = pytest.mark.django_db(transaction=True)
from taiga.base.utils import json
def test_webhooks_when_create_wiki_page(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.WikiPageFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "create"
assert data["type"] == "wikipage"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
def test_webhooks_when_update_wiki_page(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.WikiPageFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
obj.content = "test webhook update"
obj.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "wikipage"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["data"]["content"] == obj.content
assert data["change"]["comment"] == "test_comment"
assert data["change"]["diff"]["content_html"]["from"] != data["change"]["diff"]["content_html"]["to"]
assert obj.content in data["change"]["diff"]["content_html"]["to"]
def test_webhooks_when_delete_wiki_page(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.WikiPageFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, delete=True)
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "delete"
assert data["type"] == "wikipage"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert "data" in data
def test_webhooks_when_update_wiki_page_attachments(settings):
settings.WEBHOOKS_ENABLED = True
project = f.ProjectFactory()
f.WebhookFactory.create(project=project)
f.WebhookFactory.create(project=project)
obj = f.WikiPageFactory.create(project=project)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner)
assert send_request_mock.call_count == 2
# Create attachments
attachment1 = f.WikiAttachmentFactory(project=obj.project, content_object=obj, owner=obj.owner)
attachment2 = f.WikiAttachmentFactory(project=obj.project, content_object=obj, owner=obj.owner)
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "wikipage"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["attachments"]["new"]) == 2
assert len(data["change"]["diff"]["attachments"]["changed"]) == 0
assert len(data["change"]["diff"]["attachments"]["deleted"]) == 0
# Update attachment
attachment1.description = "new attachment description"
attachment1.save()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "wikipage"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["attachments"]["new"]) == 0
assert len(data["change"]["diff"]["attachments"]["changed"]) == 1
assert len(data["change"]["diff"]["attachments"]["deleted"]) == 0
# Delete attachment
attachment2.delete()
with patch('taiga.webhooks.tasks._send_request') as send_request_mock:
services.take_snapshot(obj, user=obj.owner, comment="test_comment")
assert send_request_mock.call_count == 2
(webhook_id, url, key, data) = send_request_mock.call_args[0]
assert data["action"] == "change"
assert data["type"] == "wikipage"
assert data["by"]["id"] == obj.owner.id
assert "date" in data
assert data["data"]["id"] == obj.id
assert data["change"]["comment"] == "test_comment"
assert len(data["change"]["diff"]["attachments"]["new"]) == 0
assert len(data["change"]["diff"]["attachments"]["changed"]) == 0
assert len(data["change"]["diff"]["attachments"]["deleted"]) == 1