Changeset View
Changeset View
Standalone View
Standalone View
bid_api/tests/test_webhooks.py
| import datetime | import datetime | ||||
| import hashlib | import hashlib | ||||
| import hmac | import hmac | ||||
| import json | import json | ||||
| import pathlib | import pathlib | ||||
| import responses | |||||
| from django.test import TestCase | from django.test import TestCase | ||||
| from django.utils import timezone | from django.utils import timezone | ||||
| from freezegun import freeze_time | |||||
| import responses | |||||
| from .abstract import UserModel | from .abstract import UserModel | ||||
| from bid_main.models import Role | from bid_main.models import Role | ||||
| import bid_main.fields | import bid_main.fields | ||||
| from bid_api import models | from bid_api import models | ||||
| # Import for side-effects of registering the signals | # Import for side-effects of registering the signals | ||||
| # noinspection PyUnresolvedReferences | # noinspection PyUnresolvedReferences | ||||
| ▲ Show 20 Lines • Show All 41 Lines • ▼ Show 20 Lines | def test_modify_user_email_only(self): | ||||
| self.assertEqual( | self.assertEqual( | ||||
| { | { | ||||
| "id": user.id, | "id": user.id, | ||||
| "old_email": "test@user.com", | "old_email": "test@user.com", | ||||
| "full_name": "", | "full_name": "", | ||||
| "email": "new@user.com", | "email": "new@user.com", | ||||
| "roles": [], | "roles": [], | ||||
| "avatar_changed": False, | "avatar_changed": False, | ||||
| "date_deletion_requested": None, | |||||
| }, | }, | ||||
| payload, | payload, | ||||
| ) | ) | ||||
| # This is just a stupid copy of the actual code, but I have no other way to | # This is just a stupid copy of the actual code, but I have no other way to | ||||
| # test the actual HMAC, apart from hardcoding a specific value (which is very | # test the actual HMAC, apart from hardcoding a specific value (which is very | ||||
| # fragile, and frankly also simply copied from the output of the actual code). | # fragile, and frankly also simply copied from the output of the actual code). | ||||
| mac = hmac.new( | mac = hmac.new( | ||||
| Show All 22 Lines | def test_modify_user_email_and_full_name(self): | ||||
| self.assertEqual( | self.assertEqual( | ||||
| { | { | ||||
| "id": user.id, | "id": user.id, | ||||
| "old_email": "test@user.com", | "old_email": "test@user.com", | ||||
| "full_name": "ဖန်စီဘောင်းဘီ", | "full_name": "ဖန်စီဘောင်းဘီ", | ||||
| "email": "new@user.com", | "email": "new@user.com", | ||||
| "roles": [], | "roles": [], | ||||
| "avatar_changed": False, | "avatar_changed": False, | ||||
| "date_deletion_requested": None, | |||||
| }, | }, | ||||
| payload, | payload, | ||||
| ) | ) | ||||
| @responses.activate | @responses.activate | ||||
| def test_modify_user_avatar(self): | def test_modify_user_avatar(self): | ||||
| responses.add( | responses.add( | ||||
| responses.POST, self.HOOK_URL, json={"status": "success"}, status=200 | responses.POST, self.HOOK_URL, json={"status": "success"}, status=200 | ||||
| Show All 12 Lines | def test_modify_user_avatar(self): | ||||
| self.assertEqual( | self.assertEqual( | ||||
| { | { | ||||
| "id": user.id, | "id": user.id, | ||||
| "full_name": "", | "full_name": "", | ||||
| "email": "test@user.com", | "email": "test@user.com", | ||||
| "old_email": "test@user.com", | "old_email": "test@user.com", | ||||
| "roles": [], | "roles": [], | ||||
| "avatar_changed": True, | "avatar_changed": True, | ||||
| "date_deletion_requested": None, | |||||
| }, | }, | ||||
| payload, | payload, | ||||
| ) | ) | ||||
| @responses.activate | @responses.activate | ||||
| def test_modify_user_add_role(self): | def test_modify_user_add_role(self): | ||||
| responses.add( | responses.add( | ||||
| responses.POST, self.HOOK_URL, json={"status": "success"}, status=200 | responses.POST, self.HOOK_URL, json={"status": "success"}, status=200 | ||||
| Show All 33 Lines | def test_modify_user_add_role(self): | ||||
| self.assertEqual( | self.assertEqual( | ||||
| { | { | ||||
| "id": user.id, | "id": user.id, | ||||
| "old_email": "test@user.com", | "old_email": "test@user.com", | ||||
| "full_name": "ဖန်စီဘောင်းဘီ", | "full_name": "ဖန်စီဘောင်းဘီ", | ||||
| "email": "test@user.com", | "email": "test@user.com", | ||||
| "avatar_changed": False, | "avatar_changed": False, | ||||
| "roles": ["cloud_subscriber"], | "roles": ["cloud_subscriber"], | ||||
| "date_deletion_requested": None, | |||||
| }, | }, | ||||
| payload, | payload, | ||||
| ) | ) | ||||
| payload = json.loads(responses.calls[1].request.body) | payload = json.loads(responses.calls[1].request.body) | ||||
| self.assertEqual( | self.assertEqual( | ||||
| { | { | ||||
| "id": user.id, | "id": user.id, | ||||
| "old_email": "test@user.com", | "old_email": "test@user.com", | ||||
| "full_name": "ဖန်စီဘောင်းဘီ", | "full_name": "ဖန်စီဘောင်းဘီ", | ||||
| "email": "test@user.com", | "email": "test@user.com", | ||||
| "avatar_changed": False, | "avatar_changed": False, | ||||
| "roles": [], | "roles": [], | ||||
| "date_deletion_requested": None, | |||||
| }, | |||||
| payload, | |||||
| ) | |||||
| @responses.activate | |||||
| @freeze_time('2021-01-01') | |||||
sybren: I didn't know freezegun, seems like a really useful tool!
It's easy to make mistakes with… | |||||
Done Inline ActionsTried that, but it gives strange results that don't match the offset from UTC (the formatted time is the same): "date_deletion_requested": "2021-01-01T01:02:03+00:00", Offset given as part of the time string @freeze_time('2021-01-01T01:02:03+02:00') works as expected, however: "date_deletion_requested": "2020-12-31T23:02:03+00:00", (There are a few of open issues with tz_offset) railla: Tried that, but it gives strange results that don't match the offset from UTC (the formatted… | |||||
Done Inline ActionsEek, well I guess that shows that it's a good idea to test with timezones ;-) sybren: Eek, well I guess that shows that it's a good idea to test with timezones ;-)
Your current… | |||||
| def test_modify_date_deletion_requested(self): | |||||
| responses.add( | |||||
| responses.POST, self.HOOK_URL, json={"status": "success"}, status=200 | |||||
| ) | |||||
| user = UserModel.objects.create_user("test@user.com", "123456") | |||||
| user.date_deletion_requested = timezone.now() | |||||
| user.save(update_fields={"date_deletion_requested"}) | |||||
| self.assertTrue(user.webhook_user_modified) | |||||
| self.assertEqual(1, len(responses.calls)) | |||||
| call = responses.calls[0] | |||||
| self.assertEqual(self.HOOK_URL, call.request.url) | |||||
| payload = json.loads(call.request.body) | |||||
| self.assertEqual( | |||||
| { | |||||
| "id": user.id, | |||||
| "old_email": "test@user.com", | |||||
| "full_name": "", | |||||
| "email": "test@user.com", | |||||
| "roles": [], | |||||
| "avatar_changed": False, | |||||
| "date_deletion_requested": '2021-01-01T00:00:00+00:00', | |||||
| }, | }, | ||||
| payload, | payload, | ||||
| ) | ) | ||||
| @responses.activate | @responses.activate | ||||
| def test_queue_after_error_500(self): | def test_queue_after_error_500(self): | ||||
| responses.add( | responses.add( | ||||
| responses.POST, self.HOOK_URL, json={"status": "error"}, status=500 | responses.POST, self.HOOK_URL, json={"status": "error"}, status=500 | ||||
| Show All 15 Lines | def test_queue_after_error_500(self): | ||||
| self.assertEqual( | self.assertEqual( | ||||
| { | { | ||||
| "id": user.id, | "id": user.id, | ||||
| "old_email": "test@user.com", | "old_email": "test@user.com", | ||||
| "full_name": "ဖန်စီဘောင်းဘီ", | "full_name": "ဖန်စီဘောင်းဘီ", | ||||
| "email": "test@user.com", | "email": "test@user.com", | ||||
| "avatar_changed": False, | "avatar_changed": False, | ||||
| "roles": [], | "roles": [], | ||||
| "date_deletion_requested": None, | |||||
| }, | }, | ||||
| payload, | payload, | ||||
| "The payload in the queue should be the POSTed JSON", | "The payload in the queue should be the POSTed JSON", | ||||
| ) | ) | ||||
| @responses.activate | @responses.activate | ||||
| def test_queue_after_ioerror(self): | def test_queue_after_ioerror(self): | ||||
| # Explicitly do not call responses.add(), so that it'll cause an error. | # Explicitly do not call responses.add(), so that it'll cause an error. | ||||
| Show All 18 Lines | def test_queue_after_ioerror(self): | ||||
| self.assertEqual( | self.assertEqual( | ||||
| { | { | ||||
| "id": user.id, | "id": user.id, | ||||
| "old_email": "test@user.com", | "old_email": "test@user.com", | ||||
| "full_name": "ဖန်စီဘောင်းဘီ", | "full_name": "ဖန်စီဘောင်းဘီ", | ||||
| "email": "test@user.com", | "email": "test@user.com", | ||||
| "avatar_changed": False, | "avatar_changed": False, | ||||
| "roles": [], | "roles": [], | ||||
| "date_deletion_requested": None, | |||||
| }, | }, | ||||
| payload, | payload, | ||||
| "The payload in the queue should be the POSTed JSON", | "The payload in the queue should be the POSTed JSON", | ||||
| ) | ) | ||||
| @responses.activate | @responses.activate | ||||
| def test_flushing_queue_happy(self): | def test_flushing_queue_happy(self): | ||||
| # Expect the hook at a new URL. | # Expect the hook at a new URL. | ||||
| Show All 27 Lines | def test_flushing_queue_happy(self): | ||||
| self.assertEqual( | self.assertEqual( | ||||
| { | { | ||||
| "id": user.id, | "id": user.id, | ||||
| "old_email": "test@user.com", | "old_email": "test@user.com", | ||||
| "full_name": "ဖန်စီဘောင်းဘီ", | "full_name": "ဖန်စီဘောင်းဘီ", | ||||
| "email": "test@user.com", | "email": "test@user.com", | ||||
| "avatar_changed": False, | "avatar_changed": False, | ||||
| "roles": [], | "roles": [], | ||||
| "date_deletion_requested": None, | |||||
| }, | }, | ||||
| payload, | payload, | ||||
| ) | ) | ||||
| mac = hmac.new(b"new-secret", call.request.body, hashlib.sha256) | mac = hmac.new(b"new-secret", call.request.body, hashlib.sha256) | ||||
| hexdigest = mac.hexdigest() | hexdigest = mac.hexdigest() | ||||
| self.assertEqual(hexdigest, call.request.headers["X-Webhook-HMAC"]) | self.assertEqual(hexdigest, call.request.headers["X-Webhook-HMAC"]) | ||||
| self.assertEqual("application/json", call.request.headers["Content-Type"]) | self.assertEqual("application/json", call.request.headers["Content-Type"]) | ||||
| ▲ Show 20 Lines • Show All 89 Lines • ▼ Show 20 Lines | def test_flushing_queue_via_cli_cmd_happy(self): | ||||
| self.assertEqual( | self.assertEqual( | ||||
| { | { | ||||
| "id": user.id, | "id": user.id, | ||||
| "old_email": "test@user.com", | "old_email": "test@user.com", | ||||
| "full_name": "ဖန်စီဘောင်းဘီ", | "full_name": "ဖန်စီဘောင်းဘီ", | ||||
| "email": "test@user.com", | "email": "test@user.com", | ||||
| "avatar_changed": False, | "avatar_changed": False, | ||||
| "roles": [], | "roles": [], | ||||
| "date_deletion_requested": None, | |||||
| }, | }, | ||||
| payload, | payload, | ||||
| ) | ) | ||||
| mac = hmac.new(b"new-secret", call.request.body, hashlib.sha256) | mac = hmac.new(b"new-secret", call.request.body, hashlib.sha256) | ||||
| hexdigest = mac.hexdigest() | hexdigest = mac.hexdigest() | ||||
| self.assertEqual(hexdigest, call.request.headers["X-Webhook-HMAC"]) | self.assertEqual(hexdigest, call.request.headers["X-Webhook-HMAC"]) | ||||
| self.assertEqual("application/json", call.request.headers["Content-Type"]) | self.assertEqual("application/json", call.request.headers["Content-Type"]) | ||||
| ▲ Show 20 Lines • Show All 222 Lines • Show Last 20 Lines | |||||
I didn't know freezegun, seems like a really useful tool!
It's easy to make mistakes with timezones, so I'd always suggest testing with non-zero ones. According to their docs freezegun.freeze_time('2021-01-01 01:02:03', tz_offset=-9) would do the trick.