|
1
|
import json |
|
2
|
from unittest.mock import patch |
|
3
|
|
|
4
|
import pytest |
|
5
|
from django.contrib.auth.models import User |
|
6
|
from django.test import Client |
|
7
|
|
|
8
|
from fossil.models import FossilRepository |
|
9
|
from fossil.webhooks import Webhook, WebhookDelivery |
|
10
|
from organization.models import Team |
|
11
|
from projects.models import ProjectTeam |
|
12
|
|
|
13
|
|
|
14
|
@pytest.fixture |
|
15
|
def fossil_repo_obj(sample_project): |
|
16
|
"""Return the auto-created FossilRepository for sample_project.""" |
|
17
|
return FossilRepository.objects.get(project=sample_project, deleted_at__isnull=True) |
|
18
|
|
|
19
|
|
|
20
|
@pytest.fixture |
|
21
|
def webhook(fossil_repo_obj, admin_user): |
|
22
|
return Webhook.objects.create( |
|
23
|
repository=fossil_repo_obj, |
|
24
|
url="https://example.com/webhook", |
|
25
|
secret="test-secret", |
|
26
|
events="all", |
|
27
|
is_active=True, |
|
28
|
created_by=admin_user, |
|
29
|
) |
|
30
|
|
|
31
|
|
|
32
|
@pytest.fixture |
|
33
|
def inactive_webhook(fossil_repo_obj, admin_user): |
|
34
|
return Webhook.objects.create( |
|
35
|
repository=fossil_repo_obj, |
|
36
|
url="https://example.com/inactive", |
|
37
|
secret="", |
|
38
|
events="checkin", |
|
39
|
is_active=False, |
|
40
|
created_by=admin_user, |
|
41
|
) |
|
42
|
|
|
43
|
|
|
44
|
@pytest.fixture |
|
45
|
def delivery(webhook): |
|
46
|
return WebhookDelivery.objects.create( |
|
47
|
webhook=webhook, |
|
48
|
event_type="checkin", |
|
49
|
payload={"hash": "abc123", "user": "dev"}, |
|
50
|
response_status=200, |
|
51
|
response_body="OK", |
|
52
|
success=True, |
|
53
|
duration_ms=150, |
|
54
|
attempt=1, |
|
55
|
) |
|
56
|
|
|
57
|
|
|
58
|
@pytest.fixture |
|
59
|
def writer_user(db, admin_user, sample_project): |
|
60
|
"""User with write access but not admin.""" |
|
61
|
writer = User.objects.create_user(username="writer", password="testpass123") |
|
62
|
team = Team.objects.create(name="Writers", organization=sample_project.organization, created_by=admin_user) |
|
63
|
team.members.add(writer) |
|
64
|
ProjectTeam.objects.create(project=sample_project, team=team, role="write", created_by=admin_user) |
|
65
|
return writer |
|
66
|
|
|
67
|
|
|
68
|
@pytest.fixture |
|
69
|
def writer_client(writer_user): |
|
70
|
client = Client() |
|
71
|
client.login(username="writer", password="testpass123") |
|
72
|
return client |
|
73
|
|
|
74
|
|
|
75
|
# --- Webhook Model Tests --- |
|
76
|
|
|
77
|
|
|
78
|
@pytest.mark.django_db |
|
79
|
class TestWebhookModel: |
|
80
|
def test_create_webhook(self, webhook): |
|
81
|
assert webhook.pk is not None |
|
82
|
assert str(webhook) == "https://example.com/webhook (all)" |
|
83
|
|
|
84
|
def test_soft_delete(self, webhook, admin_user): |
|
85
|
webhook.soft_delete(user=admin_user) |
|
86
|
assert webhook.is_deleted |
|
87
|
assert Webhook.objects.filter(pk=webhook.pk).count() == 0 |
|
88
|
assert Webhook.all_objects.filter(pk=webhook.pk).count() == 1 |
|
89
|
|
|
90
|
def test_secret_encrypted_at_rest(self, webhook): |
|
91
|
"""EncryptedTextField encrypts the value in the DB.""" |
|
92
|
# Read raw value from DB bypassing the field's from_db_value |
|
93
|
from django.db import connection |
|
94
|
|
|
95
|
with connection.cursor() as cursor: |
|
96
|
cursor.execute("SELECT secret FROM fossil_webhook WHERE id = %s", [webhook.pk]) |
|
97
|
raw = cursor.fetchone()[0] |
|
98
|
# Raw DB value should NOT be the plaintext |
|
99
|
assert raw != "test-secret" |
|
100
|
# But accessing via the model decrypts it |
|
101
|
webhook.refresh_from_db() |
|
102
|
assert webhook.secret == "test-secret" |
|
103
|
|
|
104
|
def test_ordering(self, fossil_repo_obj, admin_user): |
|
105
|
w1 = Webhook.objects.create(repository=fossil_repo_obj, url="https://a.com/hook", events="all", created_by=admin_user) |
|
106
|
w2 = Webhook.objects.create(repository=fossil_repo_obj, url="https://b.com/hook", events="all", created_by=admin_user) |
|
107
|
hooks = list(Webhook.objects.filter(repository=fossil_repo_obj)) |
|
108
|
# Ordered by -created_at, so newest first |
|
109
|
assert hooks[0] == w2 |
|
110
|
assert hooks[1] == w1 |
|
111
|
|
|
112
|
|
|
113
|
@pytest.mark.django_db |
|
114
|
class TestWebhookDeliveryModel: |
|
115
|
def test_create_delivery(self, delivery): |
|
116
|
assert delivery.pk is not None |
|
117
|
assert delivery.success is True |
|
118
|
assert delivery.response_status == 200 |
|
119
|
assert "abc123" in json.dumps(delivery.payload) |
|
120
|
|
|
121
|
def test_delivery_str(self, delivery): |
|
122
|
assert "example.com/webhook" in str(delivery) |
|
123
|
|
|
124
|
def test_ordering(self, webhook): |
|
125
|
d1 = WebhookDelivery.objects.create(webhook=webhook, event_type="checkin", payload={}, success=True) |
|
126
|
d2 = WebhookDelivery.objects.create(webhook=webhook, event_type="ticket", payload={}, success=False) |
|
127
|
deliveries = list(WebhookDelivery.objects.filter(webhook=webhook)) |
|
128
|
# Ordered by -delivered_at, so newest first |
|
129
|
assert deliveries[0] == d2 |
|
130
|
assert deliveries[1] == d1 |
|
131
|
|
|
132
|
|
|
133
|
# --- Webhook List View Tests --- |
|
134
|
|
|
135
|
|
|
136
|
@pytest.mark.django_db |
|
137
|
class TestWebhookListView: |
|
138
|
def test_list_webhooks(self, admin_client, sample_project, webhook): |
|
139
|
response = admin_client.get(f"/projects/{sample_project.slug}/fossil/webhooks/") |
|
140
|
assert response.status_code == 200 |
|
141
|
content = response.content.decode() |
|
142
|
assert "example.com/webhook" in content |
|
143
|
assert "Active" in content |
|
144
|
|
|
145
|
def test_list_empty(self, admin_client, sample_project, fossil_repo_obj): |
|
146
|
response = admin_client.get(f"/projects/{sample_project.slug}/fossil/webhooks/") |
|
147
|
assert response.status_code == 200 |
|
148
|
assert "No webhooks configured" in response.content.decode() |
|
149
|
|
|
150
|
def test_list_denied_for_writer(self, writer_client, sample_project, webhook): |
|
151
|
"""Webhook management requires admin, not just write.""" |
|
152
|
response = writer_client.get(f"/projects/{sample_project.slug}/fossil/webhooks/") |
|
153
|
assert response.status_code == 403 |
|
154
|
|
|
155
|
def test_list_denied_for_no_perm(self, no_perm_client, sample_project): |
|
156
|
response = no_perm_client.get(f"/projects/{sample_project.slug}/fossil/webhooks/") |
|
157
|
assert response.status_code == 403 |
|
158
|
|
|
159
|
def test_list_denied_for_anon(self, client, sample_project): |
|
160
|
response = client.get(f"/projects/{sample_project.slug}/fossil/webhooks/") |
|
161
|
assert response.status_code == 302 # redirect to login |
|
162
|
|
|
163
|
|
|
164
|
# --- Webhook Create View Tests --- |
|
165
|
|
|
166
|
|
|
167
|
@pytest.mark.django_db |
|
168
|
class TestWebhookCreateView: |
|
169
|
def test_get_form(self, admin_client, sample_project, fossil_repo_obj): |
|
170
|
response = admin_client.get(f"/projects/{sample_project.slug}/fossil/webhooks/create/") |
|
171
|
assert response.status_code == 200 |
|
172
|
assert "Create Webhook" in response.content.decode() |
|
173
|
|
|
174
|
@patch("core.url_validation.is_safe_outbound_url", return_value=(True, "")) |
|
175
|
def test_create_webhook(self, mock_url_check, admin_client, sample_project, fossil_repo_obj): |
|
176
|
response = admin_client.post( |
|
177
|
f"/projects/{sample_project.slug}/fossil/webhooks/create/", |
|
178
|
{"url": "https://hooks.example.com/test", "secret": "s3cret", "events": ["checkin", "ticket"], "is_active": "on"}, |
|
179
|
) |
|
180
|
assert response.status_code == 302 |
|
181
|
hook = Webhook.objects.get(url="https://hooks.example.com/test") |
|
182
|
assert hook.secret == "s3cret" |
|
183
|
assert hook.events == "checkin,ticket" |
|
184
|
assert hook.is_active is True |
|
185
|
|
|
186
|
@patch("core.url_validation.is_safe_outbound_url", return_value=(True, "")) |
|
187
|
def test_create_webhook_all_events(self, mock_url_check, admin_client, sample_project, fossil_repo_obj): |
|
188
|
response = admin_client.post( |
|
189
|
f"/projects/{sample_project.slug}/fossil/webhooks/create/", |
|
190
|
{"url": "https://hooks.example.com/all", "is_active": "on"}, |
|
191
|
) |
|
192
|
assert response.status_code == 302 |
|
193
|
hook = Webhook.objects.get(url="https://hooks.example.com/all") |
|
194
|
assert hook.events == "all" |
|
195
|
|
|
196
|
def test_create_denied_for_writer(self, writer_client, sample_project): |
|
197
|
response = writer_client.post( |
|
198
|
f"/projects/{sample_project.slug}/fossil/webhooks/create/", |
|
199
|
{"url": "https://evil.com/hook"}, |
|
200
|
) |
|
201
|
assert response.status_code == 403 |
|
202
|
|
|
203
|
|
|
204
|
# --- Webhook Edit View Tests --- |
|
205
|
|
|
206
|
|
|
207
|
@pytest.mark.django_db |
|
208
|
class TestWebhookEditView: |
|
209
|
def test_get_edit_form(self, admin_client, sample_project, webhook): |
|
210
|
response = admin_client.get(f"/projects/{sample_project.slug}/fossil/webhooks/{webhook.pk}/edit/") |
|
211
|
assert response.status_code == 200 |
|
212
|
content = response.content.decode() |
|
213
|
assert "example.com/webhook" in content |
|
214
|
assert "Update Webhook" in content |
|
215
|
|
|
216
|
@patch("core.url_validation.is_safe_outbound_url", return_value=(True, "")) |
|
217
|
def test_edit_webhook(self, mock_url_check, admin_client, sample_project, webhook): |
|
218
|
response = admin_client.post( |
|
219
|
f"/projects/{sample_project.slug}/fossil/webhooks/{webhook.pk}/edit/", |
|
220
|
{"url": "https://new-url.example.com/hook", "events": ["wiki"], "is_active": "on"}, |
|
221
|
) |
|
222
|
assert response.status_code == 302 |
|
223
|
webhook.refresh_from_db() |
|
224
|
assert webhook.url == "https://new-url.example.com/hook" |
|
225
|
assert webhook.events == "wiki" |
|
226
|
|
|
227
|
def test_edit_preserves_secret_when_blank(self, admin_client, sample_project, webhook): |
|
228
|
"""Editing without providing a new secret should keep the old one.""" |
|
229
|
old_secret = webhook.secret |
|
230
|
response = admin_client.post( |
|
231
|
f"/projects/{sample_project.slug}/fossil/webhooks/{webhook.pk}/edit/", |
|
232
|
{"url": "https://example.com/webhook", "secret": "", "events": ["all"], "is_active": "on"}, |
|
233
|
) |
|
234
|
assert response.status_code == 302 |
|
235
|
webhook.refresh_from_db() |
|
236
|
assert webhook.secret == old_secret |
|
237
|
|
|
238
|
def test_edit_denied_for_writer(self, writer_client, sample_project, webhook): |
|
239
|
response = writer_client.post( |
|
240
|
f"/projects/{sample_project.slug}/fossil/webhooks/{webhook.pk}/edit/", |
|
241
|
{"url": "https://evil.com/hook"}, |
|
242
|
) |
|
243
|
assert response.status_code == 403 |
|
244
|
|
|
245
|
def test_edit_nonexistent_webhook(self, admin_client, sample_project, fossil_repo_obj): |
|
246
|
response = admin_client.get(f"/projects/{sample_project.slug}/fossil/webhooks/99999/edit/") |
|
247
|
assert response.status_code == 404 |
|
248
|
|
|
249
|
|
|
250
|
# --- Webhook Delete View Tests --- |
|
251
|
|
|
252
|
|
|
253
|
@pytest.mark.django_db |
|
254
|
class TestWebhookDeleteView: |
|
255
|
def test_delete_webhook(self, admin_client, sample_project, webhook): |
|
256
|
response = admin_client.post(f"/projects/{sample_project.slug}/fossil/webhooks/{webhook.pk}/delete/") |
|
257
|
assert response.status_code == 302 |
|
258
|
webhook.refresh_from_db() |
|
259
|
assert webhook.is_deleted |
|
260
|
|
|
261
|
def test_delete_get_redirects(self, admin_client, sample_project, webhook): |
|
262
|
response = admin_client.get(f"/projects/{sample_project.slug}/fossil/webhooks/{webhook.pk}/delete/") |
|
263
|
assert response.status_code == 302 # GET redirects to list |
|
264
|
|
|
265
|
def test_delete_denied_for_writer(self, writer_client, sample_project, webhook): |
|
266
|
response = writer_client.post(f"/projects/{sample_project.slug}/fossil/webhooks/{webhook.pk}/delete/") |
|
267
|
assert response.status_code == 403 |
|
268
|
|
|
269
|
|
|
270
|
# --- Webhook Deliveries View Tests --- |
|
271
|
|
|
272
|
|
|
273
|
@pytest.mark.django_db |
|
274
|
class TestWebhookDeliveriesView: |
|
275
|
def test_view_deliveries(self, admin_client, sample_project, webhook, delivery): |
|
276
|
response = admin_client.get(f"/projects/{sample_project.slug}/fossil/webhooks/{webhook.pk}/deliveries/") |
|
277
|
assert response.status_code == 200 |
|
278
|
content = response.content.decode() |
|
279
|
assert "checkin" in content |
|
280
|
assert "200" in content |
|
281
|
assert "150ms" in content |
|
282
|
|
|
283
|
def test_view_empty_deliveries(self, admin_client, sample_project, webhook): |
|
284
|
response = admin_client.get(f"/projects/{sample_project.slug}/fossil/webhooks/{webhook.pk}/deliveries/") |
|
285
|
assert response.status_code == 200 |
|
286
|
assert "No deliveries yet" in response.content.decode() |
|
287
|
|
|
288
|
def test_deliveries_denied_for_writer(self, writer_client, sample_project, webhook): |
|
289
|
response = writer_client.get(f"/projects/{sample_project.slug}/fossil/webhooks/{webhook.pk}/deliveries/") |
|
290
|
assert response.status_code == 403 |
|
291
|
|
|
292
|
|
|
293
|
# --- Webhook Dispatch Task Tests --- |
|
294
|
|
|
295
|
|
|
296
|
@pytest.mark.django_db |
|
297
|
class TestDispatchWebhookTask: |
|
298
|
def test_successful_delivery(self, webhook): |
|
299
|
"""Test dispatch_webhook task with a successful response.""" |
|
300
|
from fossil.tasks import dispatch_webhook |
|
301
|
|
|
302
|
mock_response = type("Response", (), {"status_code": 200, "text": "OK"})() |
|
303
|
|
|
304
|
with patch("requests.post", return_value=mock_response): |
|
305
|
dispatch_webhook.apply(args=[webhook.pk, "checkin", {"hash": "abc"}]) |
|
306
|
|
|
307
|
delivery = WebhookDelivery.objects.get(webhook=webhook) |
|
308
|
assert delivery.success is True |
|
309
|
assert delivery.response_status == 200 |
|
310
|
assert delivery.event_type == "checkin" |
|
311
|
|
|
312
|
def test_failed_delivery_logs(self, webhook): |
|
313
|
"""Test that failed HTTP responses are logged and delivery is recorded.""" |
|
314
|
from fossil.tasks import dispatch_webhook |
|
315
|
|
|
316
|
mock_response = type("Response", (), {"status_code": 500, "text": "Internal Server Error"})() |
|
317
|
|
|
318
|
with patch("requests.post", return_value=mock_response): |
|
319
|
dispatch_webhook.apply(args=[webhook.pk, "checkin", {"hash": "abc"}]) |
|
320
|
|
|
321
|
# The task retries on non-2xx, so apply() catches the Retry internally |
|
322
|
delivery = WebhookDelivery.objects.filter(webhook=webhook).first() |
|
323
|
assert delivery is not None |
|
324
|
assert delivery.success is False |
|
325
|
assert delivery.response_status == 500 |
|
326
|
|
|
327
|
def test_nonexistent_webhook_no_crash(self): |
|
328
|
"""Task should handle missing webhook gracefully.""" |
|
329
|
from fossil.tasks import dispatch_webhook |
|
330
|
|
|
331
|
# Should return without error (logs warning) |
|
332
|
dispatch_webhook.apply(args=[99999, "checkin", {"hash": "abc"}]) |
|
333
|
# No deliveries created |
|
334
|
assert WebhookDelivery.objects.count() == 0 |
|
335
|
|
|
336
|
def test_hmac_signature_set_with_secret(self, webhook): |
|
337
|
"""When webhook has a secret, X-Fossilrepo-Signature header is set.""" |
|
338
|
from fossil.tasks import dispatch_webhook |
|
339
|
|
|
340
|
mock_response = type("Response", (), {"status_code": 200, "text": "OK"})() |
|
341
|
captured_kwargs = {} |
|
342
|
|
|
343
|
def capture_post(*args, **kwargs): |
|
344
|
captured_kwargs.update(kwargs) |
|
345
|
return mock_response |
|
346
|
|
|
347
|
with patch("requests.post", side_effect=capture_post): |
|
348
|
dispatch_webhook.apply(args=[webhook.pk, "checkin", {"hash": "abc"}]) |
|
349
|
|
|
350
|
headers = captured_kwargs.get("headers", {}) |
|
351
|
assert "X-Fossilrepo-Signature" in headers |
|
352
|
assert headers["X-Fossilrepo-Signature"].startswith("sha256=") |
|
353
|
|
|
354
|
def test_no_signature_without_secret(self, fossil_repo_obj, admin_user): |
|
355
|
"""When webhook has no secret, no signature header is sent.""" |
|
356
|
from fossil.tasks import dispatch_webhook |
|
357
|
|
|
358
|
hook = Webhook.objects.create( |
|
359
|
repository=fossil_repo_obj, |
|
360
|
url="https://no-secret.example.com/hook", |
|
361
|
secret="", |
|
362
|
events="all", |
|
363
|
is_active=True, |
|
364
|
created_by=admin_user, |
|
365
|
) |
|
366
|
|
|
367
|
mock_response = type("Response", (), {"status_code": 200, "text": "OK"})() |
|
368
|
captured_kwargs = {} |
|
369
|
|
|
370
|
def capture_post(*args, **kwargs): |
|
371
|
captured_kwargs.update(kwargs) |
|
372
|
return mock_response |
|
373
|
|
|
374
|
with patch("requests.post", side_effect=capture_post): |
|
375
|
dispatch_webhook.apply(args=[hook.pk, "checkin", {"hash": "abc"}]) |
|
376
|
|
|
377
|
headers = captured_kwargs.get("headers", {}) |
|
378
|
assert "X-Fossilrepo-Signature" not in headers |
|
379
|
|