|
2 | 2 | from unittest.mock import patch |
3 | 3 |
|
4 | 4 | import pytest |
| 5 | +from django.core.exceptions import ObjectDoesNotExist |
5 | 6 | from django.urls import reverse |
6 | 7 | from rest_framework import status |
7 | 8 | from rest_framework.response import Response |
@@ -1253,3 +1254,99 @@ def test_webhook_trigger_manual( |
1253 | 1254 | ) |
1254 | 1255 | assert response.status_code == status.HTTP_404_NOT_FOUND |
1255 | 1256 | assert mock_execute.apply_async.call_count == 0 |
| 1257 | + |
| 1258 | + |
| 1259 | +@pytest.mark.django_db |
| 1260 | +def test_current_personal_notification( |
| 1261 | + make_organization_and_user_with_plugin_token, |
| 1262 | + make_custom_webhook, |
| 1263 | + make_user_auth_headers, |
| 1264 | + make_personal_notification_webhook, |
| 1265 | +): |
| 1266 | + organization, user, token = make_organization_and_user_with_plugin_token() |
| 1267 | + with pytest.raises(ObjectDoesNotExist): |
| 1268 | + user.personal_webhook |
| 1269 | + |
| 1270 | + webhook = make_custom_webhook(organization, trigger_type=Webhook.TRIGGER_PERSONAL_NOTIFICATION) |
| 1271 | + |
| 1272 | + client = APIClient() |
| 1273 | + url = reverse("api-internal:webhooks-current-personal-notification") |
| 1274 | + |
| 1275 | + # no webhook setup |
| 1276 | + response = client.get(url, **make_user_auth_headers(user, token)) |
| 1277 | + assert response.status_code == status.HTTP_200_OK |
| 1278 | + assert response.json() == {"webhook": None, "context": None} |
| 1279 | + |
| 1280 | + # setup personal webhook |
| 1281 | + personal_webhook = make_personal_notification_webhook(user, webhook) |
| 1282 | + response = client.get(url, **make_user_auth_headers(user, token)) |
| 1283 | + assert response.status_code == status.HTTP_200_OK |
| 1284 | + assert response.json() == {"webhook": webhook.public_primary_key, "context": {}} |
| 1285 | + |
| 1286 | + # update context data |
| 1287 | + personal_webhook.context_data = {"test": "test"} |
| 1288 | + response = client.get(url, **make_user_auth_headers(user, token)) |
| 1289 | + assert response.status_code == status.HTTP_200_OK |
| 1290 | + assert response.json() == {"webhook": webhook.public_primary_key, "context": {"test": "test"}} |
| 1291 | + |
| 1292 | + |
| 1293 | +@pytest.mark.django_db |
| 1294 | +def test_set_personal_notification( |
| 1295 | + make_organization_and_user_with_plugin_token, |
| 1296 | + make_custom_webhook, |
| 1297 | + make_user_auth_headers, |
| 1298 | +): |
| 1299 | + organization, user, token = make_organization_and_user_with_plugin_token() |
| 1300 | + with pytest.raises(ObjectDoesNotExist): |
| 1301 | + user.personal_webhook |
| 1302 | + |
| 1303 | + webhook = make_custom_webhook(organization, trigger_type=Webhook.TRIGGER_PERSONAL_NOTIFICATION) |
| 1304 | + other_webhook = make_custom_webhook(organization, trigger_type=Webhook.TRIGGER_MANUAL) |
| 1305 | + |
| 1306 | + client = APIClient() |
| 1307 | + url = reverse("api-internal:webhooks-set-personal-notification") |
| 1308 | + |
| 1309 | + # webhook id is required |
| 1310 | + data = {} |
| 1311 | + response = client.post( |
| 1312 | + url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(user, token) |
| 1313 | + ) |
| 1314 | + assert response.status_code == status.HTTP_400_BAD_REQUEST |
| 1315 | + assert response.json()["webhook"] == "This field is required." |
| 1316 | + |
| 1317 | + # invalid webhook type |
| 1318 | + data = {"webhook": other_webhook.public_primary_key} |
| 1319 | + response = client.post( |
| 1320 | + url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(user, token) |
| 1321 | + ) |
| 1322 | + assert response.status_code == status.HTTP_400_BAD_REQUEST |
| 1323 | + assert response.json()["webhook"] == "Webhook not found." |
| 1324 | + |
| 1325 | + # check backend info |
| 1326 | + data = {"webhook": webhook.public_primary_key} |
| 1327 | + response = client.post( |
| 1328 | + url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(user, token) |
| 1329 | + ) |
| 1330 | + assert response.status_code == status.HTTP_200_OK |
| 1331 | + user.refresh_from_db() |
| 1332 | + assert user.personal_webhook.webhook == webhook |
| 1333 | + assert user.personal_webhook.context_data == {} |
| 1334 | + |
| 1335 | + # update context data |
| 1336 | + data = {"webhook": webhook.public_primary_key, "context": {"test": "test"}} |
| 1337 | + response = client.post( |
| 1338 | + url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(user, token) |
| 1339 | + ) |
| 1340 | + assert response.status_code == status.HTTP_200_OK |
| 1341 | + user.refresh_from_db() |
| 1342 | + assert user.personal_webhook.context_data == {"test": "test"} |
| 1343 | + |
| 1344 | + # invalid context |
| 1345 | + data = {"webhook": webhook.public_primary_key, "context": "not-json"} |
| 1346 | + response = client.post( |
| 1347 | + url, data=json.dumps(data), content_type="application/json", **make_user_auth_headers(user, token) |
| 1348 | + ) |
| 1349 | + assert response.status_code == status.HTTP_400_BAD_REQUEST |
| 1350 | + assert response.json()["context"] == "Invalid context." |
| 1351 | + user.refresh_from_db() |
| 1352 | + assert user.personal_webhook.context_data == {"test": "test"} |
0 commit comments