Changeset View
Changeset View
Standalone View
Standalone View
profiles/tests/test_webhooks.py
| from typing import Dict, Union | from typing import Dict, Union | ||||
| from unittest.mock import patch | from unittest.mock import patch | ||||
| import hashlib | import hashlib | ||||
| import hmac | import hmac | ||||
| import json | import json | ||||
| import responses | import responses | ||||
| from django.contrib.auth.models import User, Group | from django.contrib.auth.models import User, Group | ||||
| from django.test import TestCase, override_settings | from django.test import TestCase, override_settings | ||||
| from django.urls import reverse | from django.urls import reverse | ||||
| from common.tests.factories.users import UserFactory | from common.tests.factories.users import UserFactory | ||||
| from profiles.models import Profile | from profiles.models import Profile | ||||
| import profiles.tests.util as util | |||||
| BLENDER_ID_BASE_URL = 'http://id.local:8000/' | BLENDER_ID_BASE_URL = 'http://id.local:8000/' | ||||
| def prepare_hmac_header(body: Union[str, dict], secret: str = 'testsecret') -> Dict[str, str]: | def prepare_hmac_header(body: Union[str, dict], secret: str = 'testsecret') -> Dict[str, str]: | ||||
| """Return a dict containing an HMAC header matching a given request body.""" | """Return a dict containing an HMAC header matching a given request body.""" | ||||
| if isinstance(body, dict): | if isinstance(body, dict): | ||||
| body = json.dumps(body).encode() | body = json.dumps(body).encode() | ||||
| Show All 17 Lines | webhook_payload = { | ||||
| 'full_name': 'Иван Васильевич Doe', | 'full_name': 'Иван Васильевич Doe', | ||||
| 'id': 2, | 'id': 2, | ||||
| 'old_email': 'mail@example.com', | 'old_email': 'mail@example.com', | ||||
| 'roles': [], | 'roles': [], | ||||
| } | } | ||||
| def setUp(self): | def setUp(self): | ||||
| self.url = reverse('webhook-user-modified') | self.url = reverse('webhook-user-modified') | ||||
| # Mock Blender ID responses | util.mock_blender_id_responses() | ||||
| responses.add( | |||||
| responses.GET, | |||||
| f'{BLENDER_ID_BASE_URL}api/user/2/avatar', | |||||
| status=302, | |||||
| headers={ | |||||
| 'Location': f'{BLENDER_ID_BASE_URL}media/cache/1c/da/1cda54d605799b1f4b0dc080.jpg', | |||||
| }, | |||||
| ) | |||||
| responses.add( | |||||
| responses.GET, | |||||
| f'{BLENDER_ID_BASE_URL}api/me', | |||||
| json={ | |||||
| 'id': 2, | |||||
| 'full_name': 'Jane Doe', | |||||
| 'email': 'jane@example.com', | |||||
| 'nickname': 'ⅉanedoe', | |||||
| # N.B.: roles format here differs from one in user-modified webhook payload. | |||||
| 'roles': {'dev_core': True}, | |||||
| }, | |||||
| ) | |||||
| with open('common/static/common/images/blank-profile-pic.jpg', 'rb') as out: | |||||
| responses.add( | |||||
| responses.GET, | |||||
| 'http://id.local:8000/media/cache/1c/da/1cda54d605799b1f4b0dc080.jpg', | |||||
| body=out, | |||||
| stream=True, | |||||
| ) | |||||
sybren: :+1: for moving this code into its own function. The comment can go, though, as it just repeats… | |||||
| # Prepare a user | # Prepare a user | ||||
| self.user = UserFactory( | self.user = UserFactory( | ||||
| email='mail@example.com', | email='mail@example.com', | ||||
| oauth_info__oauth_user_id='2', | oauth_info__oauth_user_id='2', | ||||
| oauth_tokens__oauth_user_id='2', | oauth_tokens__oauth_user_id='2', | ||||
| oauth_tokens__access_token='testaccesstoken', | oauth_tokens__access_token='testaccesstoken', | ||||
| oauth_tokens__refresh_token='testrefreshtoken', | oauth_tokens__refresh_token='testrefreshtoken', | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 83 Lines • ▼ Show 20 Lines | def test_user_modified_roles_added_removed_adds_removes_user_groups(self): | ||||
| self.url, body, content_type='application/json', **prepare_hmac_header(body) | self.url, body, content_type='application/json', **prepare_hmac_header(body) | ||||
| ) | ) | ||||
| self.assertEquals(response.status_code, 204) | self.assertEquals(response.status_code, 204) | ||||
| self.assertEquals(response.content, b'') | self.assertEquals(response.content, b'') | ||||
| user = User.objects.get(pk=self.user.pk) | user = User.objects.get(pk=self.user.pk) | ||||
| self.assertEquals( | self.assertEquals( | ||||
| sorted([g.name for g in user.groups.all()]), | sorted([g.name for g in user.groups.all()]), | ||||
| ['cloud_has_subscription', 'cloud_subscriber'], | ['has_subscription', 'subscriber'], | ||||
| ) | ) | ||||
| # One role removed | # One role removed | ||||
| body = { | body = { | ||||
| **self.webhook_payload, | **self.webhook_payload, | ||||
| 'roles': ['cloud_has_subscription'], | 'roles': ['cloud_has_subscription'], | ||||
| } | } | ||||
| response = self.client.post( | response = self.client.post( | ||||
| self.url, body, content_type='application/json', **prepare_hmac_header(body) | self.url, body, content_type='application/json', **prepare_hmac_header(body) | ||||
| ) | ) | ||||
| self.assertEquals(response.status_code, 204) | self.assertEquals(response.status_code, 204) | ||||
| user = User.objects.get(pk=self.user.pk) | user = User.objects.get(pk=self.user.pk) | ||||
| self.assertEquals( | self.assertEquals( | ||||
| sorted([g.name for g in user.groups.all()]), | sorted([g.name for g in user.groups.all()]), | ||||
| [ | [ | ||||
| 'cloud_has_subscription', | 'has_subscription', | ||||
| ], | ], | ||||
| ) | ) | ||||
| # Check that the group itself still exists | # Check that the group itself still exists | ||||
| self.assertEquals(Group.objects.filter(name='cloud_subscriber').count(), 1) | self.assertEquals(Group.objects.filter(name='subscriber').count(), 1) | ||||
| # Two roles added, one already exists | # Two roles added, one already exists | ||||
| body = { | body = { | ||||
| **self.webhook_payload, | **self.webhook_payload, | ||||
| 'roles': ['cloud_has_subscription', 'cloud_subscriber', 'dev_core'], | 'roles': ['cloud_has_subscription', 'cloud_subscriber', 'dev_core'], | ||||
| } | } | ||||
| response = self.client.post( | response = self.client.post( | ||||
| self.url, body, content_type='application/json', **prepare_hmac_header(body) | self.url, body, content_type='application/json', **prepare_hmac_header(body) | ||||
| ) | ) | ||||
| self.assertEquals(response.status_code, 204) | self.assertEquals(response.status_code, 204) | ||||
| user = User.objects.get(pk=self.user.pk) | user = User.objects.get(pk=self.user.pk) | ||||
| self.assertEquals( | self.assertEquals( | ||||
| sorted([g.name for g in user.groups.all()]), | sorted([g.name for g in user.groups.all()]), | ||||
| [ | [ | ||||
| 'cloud_has_subscription', | |||||
| 'cloud_subscriber', | |||||
| 'dev_core', | 'dev_core', | ||||
| 'has_subscription', | |||||
| 'subscriber', | |||||
| ], | ], | ||||
| ) | ) | ||||
| def test_user_modified_missing_oauth_info(self): | def test_user_modified_missing_oauth_info(self): | ||||
| body = { | body = { | ||||
| **self.webhook_payload, | **self.webhook_payload, | ||||
| 'id': '999', | 'id': '999', | ||||
| } | } | ||||
| ▲ Show 20 Lines • Show All 49 Lines • Show Last 20 Lines | |||||
👍 for moving this code into its own function. The comment can go, though, as it just repeats the function name now ;-)
(same for the other appearances of the comment)