FossilRepo

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

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button