Completed
Push — master ( 0a6e76...6b11d9 )
by Fox
01:00
created

WikiPage.change()   A

Complexity

Conditions 2

Size

Total Lines 10

Duplication

Lines 0
Ratio 0 %

Importance

Changes 1
Bugs 0 Features 0
Metric Value
cc 2
c 1
b 0
f 0
dl 0
loc 10
rs 9.4285
1
import hmac
2
import hashlib
3
4
from rest_framework.response import Response
5
from rest_framework.decorators import api_view
6
7
from django_th.models import TriggerService
8
from django_th.services import default_provider
9
10
from th_taiga.models import Taiga
11
12
13
class TaigaDomain:
14
15
    def factory(type):
16
        if type == "epic":
17
            return Epic()
18
        if type == "issue":
19
            return Issue()
20
        if type == "task":
21
            return Task()
22
        if type == "userstory":
23
            return UserStory()
24
        if type == "wikipage":
25
            return WikiPage()
26
        if type == "relateduserstory":
27
            return RelatedUserStory()
28
        assert 0, "Bad type creation: " + type
29
30
    factory = staticmethod(factory)
31
32
33
class Epic(TaigaDomain):
34
35
    def create(self, taiga_obj, data):
36
        """
37
38
        :param taiga_obj: taiga object
39
        :param data: data to return
40
        :return: data
41
        """
42
        if taiga_obj.notify_epic_create:
43
            data['type_action'] = 'New Epic created'
44
        return data
45
46
    def change(self, taiga_obj, data):
47
        """
48
49
        :param taiga_obj: taiga object
50
        :param data: data to return
51
        :return: data
52
        """
53
        if taiga_obj.notify_epic_change:
54
            data['type_action'] = 'Changed Epic'
55
        return data
56
57
    def delete(self, taiga_obj, data):
58
        """
59
60
        :param taiga_obj: taiga object
61
        :param data: data to return
62
        :return: data
63
        """
64
        if taiga_obj.notify_epic_delete:
65
            data['type_action'] = 'Deleted Epic'
66
        return data
67
68
69
class Issue(TaigaDomain):
70
71
    def create(self, taiga_obj, data):
72
        """
73
74
        :param taiga_obj: taiga object
75
        :param action: craete/change/delete
76
        :param data: data to return
77
        :return: data
78
        """
79
        if taiga_obj.notify_issue_create:
80
            data['type_action'] = 'New Issue created'
81
        return data
82
83
    def change(self, taiga_obj, data):
84
        """
85
86
        :param taiga_obj: taiga object
87
        :param action: craete/change/delete
88
        :param data: data to return
89
        :return: data
90
        """
91
        if taiga_obj.notify_issue_change:
92
            data['type_action'] = 'Changed Issue'
93
        return data
94
95
    def delete(self, taiga_obj, data):
96
        """
97
98
        :param taiga_obj: taiga object
99
        :param action: craete/change/delete
100
        :param data: data to return
101
        :return: data
102
        """
103
        if taiga_obj.notify_issue_delete:
104
            data['type_action'] = 'Deleted Issue'
105
        return data
106
107
108
class UserStory(TaigaDomain):
109
110
    def create(self, taiga_obj, data):
111
        """
112
113
        :param taiga_obj: taiga object
114
        :param data: data to return
115
        :return: data
116
        """
117
        if taiga_obj.notify_userstory_create:
118
            data['type_action'] = 'New Userstory created'
119
120
        return data
121
122
    def change(self, taiga_obj, data):
123
        """
124
125
        :param taiga_obj: taiga object
126
        :param data: data to return
127
        :return: data
128
        """
129
        if taiga_obj.notify_userstory_change:
130
            data['type_action'] = 'Changed Userstory'
131
        return data
132
133
    def delete(self, taiga_obj, data):
134
        """
135
136
        :param taiga_obj: taiga object
137
        :param data: data to return
138
        :return: data
139
        """
140
        if taiga_obj.notify_userstory_delete:
141
            data['type_action'] = 'Deleted Userstory'
142
        return data
143
144 View Code Duplication
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
145
class Task(TaigaDomain):
146
    def create(self,  taiga_obj, data):
147
        """
148
149
        :param taiga_obj: taiga object
150
        :param data: data to return
151
        :return: data
152
        """
153
        if taiga_obj.notify_task_create:
154
            data['type_action'] = 'New Task created'
155
        return data
156
157
    def change(self,  taiga_obj, data):
158
        """
159
160
        :param taiga_obj: taiga object
161
        :param data: data to return
162
        :return: data
163
        """
164
        if taiga_obj.notify_task_change:
165
            data['type_action'] = 'Changed Task'
166
        return data
167
168
    def delete(self,  taiga_obj, data):
169
        """
170
171
        :param taiga_obj: taiga object
172
        :param data: data to return
173
        :return: data
174
        """
175
        if taiga_obj.notify_task_delete:
176
            data['type_action'] = 'Deleted Task'
177
        return data
178
179
180
class WikiPage(TaigaDomain):
181
182
    def create(self, taiga_obj, data):
183
        """
184
185
        :param taiga_obj: taiga object
186
        :param data: data to return
187
        :return: data
188
        """
189
        if taiga_obj.notify_wikipage_create:
190
            data['type_action'] = 'New Wikipage created'
191
        return data
192
193
    def change(self, taiga_obj, data):
194
        """
195
196
        :param taiga_obj: taiga object
197
        :param data: data to return
198
        :return: data
199
        """
200
        if taiga_obj.notify_wikipage_change:
201
            data['type_action'] = 'Changed Wikipage'
202
        return data
203
204
    def delete(self, taiga_obj, data):
205
        """
206
207
        :param taiga_obj: taiga object
208
        :param data: data to return
209
        :return: data
210
        """
211
        if taiga_obj.notify_wikipage_delete:
212
            data['type_action'] = 'Deleted Wikipage'
213
        return data
214
215
216
class RelatedUserStory(TaigaDomain):
217
218
    def create(self, taiga_obj, data):
219
        """
220
221
        :param taiga_obj: taiga object
222
        :param data: data to return
223
        :return: data
224
        """
225
        if taiga_obj.notify_relateduserstory_create:
226
            data['type_action'] = 'New Related Userstory created'
227
        return data
228
229
    def delete(self, taiga_obj, data):
230
        """
231
232
        :param taiga_obj: taiga object
233
        :param data: data to return
234
        :return: data
235
        """
236
        if taiga_obj.notify_relateduserstory_delete:
237
            data['type_action'] = 'Deleted Related Userstory'
238
        return data
239
240
241
def data_filter(trigger_id, **data):
242
    """
243
    check if we want to track event for a given action
244
    :param trigger_id:
245
    :param data:
246
    :return:
247
    """
248
    taiga_obj = Taiga.objects.get(trigger_id=trigger_id)
249
250
    action = data.get('action')
251
    domain = data.get('type')
252
    data = data.get('data')
253
254
    t = TaigaDomain.factory(domain)
255
    if action == 'create':
256
        t.create(taiga_obj, data)
257
    elif action == 'change':
258
        t.change(taiga_obj, data)
259
    elif action == 'delete':
260
        t.delete(taiga_obj, data)
261
    return data
262
263
264
def consumer(trigger_id, data):
265
    """
266
        call the consumer and handle the data
267
        :param trigger_id:
268
        :param data:
269
        :return:
270
    """
271
    status = True
272
    # consumer - the service which uses the data
273
    default_provider.load_services()
274
    service = TriggerService.objects.get(id=trigger_id)
275
276
    service_consumer = default_provider.get_service(
277
        str(service.consumer.name.name))
278
    kwargs = {'user': service.user}
279
280
    data = data_filter(trigger_id, **data)
281
    if len(data) > 0:
282
283
        getattr(service_consumer, '__init__')(service.consumer.token,
284
                                              **kwargs)
285
        status = getattr(service_consumer, 'save_data')(service.id, **data)
286
287
    return status
288
289
290
def verify_signature(data, key, signature):
291
    mac = hmac.new(key.encode("utf-8"), msg=data, digestmod=hashlib.sha1)
292
    return mac.hexdigest() == signature
293
294
295
@api_view(['POST'])
296
def taiga(request, trigger_id, key):
297
    signature = request.META.get('HTTP_X_TAIGA_WEBHOOK_SIGNATURE')
298
    # check that the data are ok with the provided signature
299
    if verify_signature(request._request.body, key, signature):
300
        status = consumer(trigger_id, request.data)
301
        if status:
302
            return Response({"message": "Success"})
303
        else:
304
            return Response({"message": "Failed!"})
305