Differential D9035 Diff 29389 profiles/tests/test_webhooks.py

Changeset View

Standalone View

profiles/tests/test_webhooks.py

This file was added.

				from typing import Dict, Union
				from unittest.mock import patch
				import hashlib
				import hmac
				import json
				import responses

				from django.contrib.auth.models import User, Group
				from django.test import RequestFactory, TestCase, override_settings
				from django.urls import reverse

				from common.tests.factories.users import UserFactory
				from profiles.models import Profile

				BLENDER_ID_BASE_URL = 'http://id.local:8000/'


				def prepare_hmac_header(body: Union[str, dict], secret: str = 'testsecret') -> Dict[str, str]:
				sybrenUnsubmitted Done Inline Actions Add type declarations. sybren: Add type declarations.
				if isinstance(body, dict):
				body = json.dumps(body).encode()
				return {'HTTP_X-Webhook-HMAC': hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()}


				sybrenUnsubmitted Not Done Inline Actions There is a lot happening on this one line. It'll probably make debugging easier if it's split up into a few smaller lines. sybren: There is a lot happening on this one line. It'll probably make debugging easier if it's split…
				@override_settings(BLENDER_ID={
				'BASE_URL': BLENDER_ID_BASE_URL,
				'OAUTH_CLIENT': 'testoauthclient',
				'OAUTH_SECRET': 'testoathsecret',
				'WEBHOOK_USER_MODIFIED_SECRET': b'testsecret',
				})
				class WebhooksTest(TestCase):
				webhook_payload = {
				'avatar_changed': False,
				'email': 'newmail@example.com',
				'full_name': 'Иван Васильевич Doe',
				'id': 2,
				'old_email': 'mail@example.com',
				'roles': [],
				}

				def setUp(self):
				self.factory = RequestFactory()
				self.url = reverse('webhook-user-modified')
				# Mock Blender ID responses
				responses.add(
				sybrenUnsubmitted Done Inline Actions Include non-ASCII characters in names. sybren: Include non-ASCII characters in names.
				responses.GET, f'{BLENDER_ID_BASE_URL}api/user/2/avatar',
				status=302, headers={
				'Location': f'{BLENDER_ID_BASE_URL}media/cache/1c/da/1cda54d605799b1f4b0dc080cb9a18ad.jpg',
				})
				responses.add(
				responses.GET, f'{BLENDER_ID_BASE_URL}api/me',
				json={
				'id': 2,
				'full_name': 'Jane Doe',
				'email': 'jane@example.com',
				'nickname': 'janedoe',
				# 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/1cda54d605799b1f4b0dc080cb9a18ad.jpg',
				body=out, stream=True)
				sybrenUnsubmitted Not Done Inline Actions Nicknames in Blender ID aren't limited to ASCII, so be sure to test with non-ASCII nicknames too. sybren: Nicknames in Blender ID aren't limited to ASCII, so be sure to test with non-ASCII nicknames…

				# Prepare a user
				self.user = UserFactory(
				email='mail@example.com',
				oauth_info__oauth_user_id='2',
				oauth_tokens__oauth_user_id='2',
				oauth_tokens__access_token='testaccesstoken',
				oauth_tokens__refresh_token='testrefreshtoken',
				)

				def test_user_modified_missing_hmac(self):
				response = self.client.post(self.url, {}, content_type='application/json')

				self.assertEquals(response.status_code, 400)
				self.assertEquals(response.content, b'Invalid HMAC')

				def test_user_modified_invalid_hmac(self):
				url = reverse('webhook-user-modified')
				headers = {'HTTP_X-Webhook-HMAC': 'deadbeef'}
				response = self.client.post(url, {}, content_type='application/json', **headers)

				self.assertEquals(response.status_code, 400)
				self.assertEquals(response.content, b'Invalid HMAC')

				@patch('profiles.views.webhooks.WEBHOOK_MAX_BODY_SIZE', 1)
				def test_user_modified_request_body_too_large(self):
				body = {"deadbeef": "foobar"}
				response = self.client.post(self.url, body, content_type='application/json', **prepare_hmac_header(body))

				self.assertEquals(response.status_code, 413)

				def test_user_modified_unexpected_content_type(self):
				response = self.client.post(self.url, 'text', content_type='text/plain', **prepare_hmac_header(b'text'))

				self.assertEquals(response.status_code, 400)
				self.assertEquals(response.content, b'Unsupported Content-Type')

				def test_user_modified_malformed_json(self):
				body = b'{"":"",}'
				response = self.client.post(self.url, body, content_type='application/json', **prepare_hmac_header(body))

				self.assertEquals(response.status_code, 400)
				self.assertEquals(response.content, b'Malformed JSON')

				@responses.activate
				def test_user_modified_updates_profile_and_user(self):
				body = self.webhook_payload
				response = self.client.post(self.url, body, content_type='application/json', **prepare_hmac_header(body))

				self.assertEquals(response.status_code, 204)
				self.assertEquals(response.content, b'')
				profile = Profile.objects.get(user_id=self.user.pk)
				self.assertEquals(profile.full_name, 'Иван Васильевич Doe')
				self.assertEquals(profile.user.email, 'newmail@example.com')

				@responses.activate
				def test_user_modified_avatar_changed(self):
				body = {
				**self.webhook_payload,
				'avatar_changed': True,
				}
				response = self.client.post(self.url, body, content_type='application/json', **prepare_hmac_header(body))

				self.assertEquals(response.status_code, 204)
				self.assertEquals(response.content, b'')
				profile = Profile.objects.get(user_id=self.user.pk)
				self.assertTrue(profile.avatar.name.endswith('.jpg'))

				@responses.activate
				sybrenUnsubmitted Not Done Inline Actions Test the entire name. Now a bug that ignores the name and turns it into `".jpg"` won't be detected. sybren: Test the entire name. Now a bug that ignores the name and turns it into `".jpg"` won't be…
				raillaAuthorUnsubmitted Done Inline Actions The name is generated by the storage and IRL will never match the original name because it's hashed (which is tested elsewhere) and naming/storing logic is not specific to the avatar in any way. Checking the full name would require mocking the storage and/or the hashing, and it seems out of scope of this change, which should assume that storage just works. I've added a todo https://developer.blender.org/T81290 about configuring test storage in the backlog, because currently it's using the same configuration staging is using, which is not ideal. railla: The name is generated by the storage and IRL will never match the original name because it's…
				sybrenUnsubmitted Not Done Inline Actions Fair enough, but why then bother checking the avatar name at all? sybren: Fair enough, but why then bother checking the avatar name at all?
				raillaAuthorUnsubmitted Done Inline Actions It's just a sanity check: the goal of the test is to test that avatar change is being handled in some meaningful way ("`avatar` used to be empty, check that it's not empty after the webhook was called"). Afaik, the only way to check that a `FileField` is "not empty", is to check either `bool(file_field)` or `file_field.name`, and out of these two I'd prefer the explicit latter one, and just to remind whomever reading the test that it's an image, also check for the extension. railla: It's just a sanity check: the goal of the test is to test that avatar change is being handled…
				sybrenUnsubmitted Not Done Inline Actions Ok, fair enough. sybren: Ok, fair enough.
				def test_user_modified_roles_added_removed_adds_removes_user_groups(self):
				# No groups ("roles") assigned yet
				self.assertEquals(list(self.user.groups.all()), [])

				# Two new roles added
				body = {
				**self.webhook_payload,
				'roles': ['cloud_has_subscription', 'cloud_subscriber'],
				}
				response = self.client.post(self.url, body, content_type='application/json', **prepare_hmac_header(body))

				self.assertEquals(response.status_code, 204)
				self.assertEquals(response.content, b'')
				user = User.objects.get(pk=self.user.pk)
				self.assertEquals(
				sorted([g.name for g in user.groups.all()]),
				['cloud_has_subscription', 'cloud_subscriber']
				)

				# One role removed
				body = {
				**self.webhook_payload,
				'roles': ['cloud_has_subscription'],
				}
				response = self.client.post(self.url, body, content_type='application/json', **prepare_hmac_header(body))

				self.assertEquals(response.status_code, 204)
				user = User.objects.get(pk=self.user.pk)
				self.assertEquals(
				sorted([g.name for g in user.groups.all()]),
				['cloud_has_subscription']
				)
				# Check that the group itself still exists
				self.assertEquals(Group.objects.filter(name='cloud_subscriber').count(), 1)

				def test_user_modified_missing_oauth_info(self):
				body = {
				**self.webhook_payload,
				'id': '999',
				}
				with self.assertLogs('', level='ERROR') as logs:
				response = self.client.post(self.url, body, content_type='application/json', **prepare_hmac_header(body))
				self.assertRegex(logs.output[0], 'Cannot update profile: no OAuth info found for ID 999')

				self.assertEquals(response.status_code, 204)

				@responses.activate
				def test_user_modified_logs_errors_when_blender_id_user_info_broken(self):
				body = self.webhook_payload
				# Mock a "broken" user info response
				responses.replace(
				responses.GET, f'{BLENDER_ID_BASE_URL}api/me',
				status=403, body='Unauthorized')

				with self.assertLogs('', level='ERROR') as logs:
				response = self.client.post(self.url, body, content_type='application/json', **prepare_hmac_header(body))
				self.assertRegex(logs.output[0], 'Unable to update username for Profile')

				self.assertEquals(response.status_code, 204)

				@responses.activate
				def test_user_modified_logs_error_when_blender_id_avatar_broken(self):
				body = {
				**self.webhook_payload,
				'avatar_changed': True,
				}
				# Mock a "broken" avatar response
				responses.replace(
				responses.GET, f'{BLENDER_ID_BASE_URL}api/user/2/avatar',
				status=500, body='Houston, we have a problem')

				with self.assertLogs('', level='ERROR') as logs:
				response = self.client.post(self.url, body, content_type='application/json', **prepare_hmac_header(body))
				self.assertRegex(logs.output[0], 'Failed to copy avatar for Profile')

				self.assertEquals(response.status_code, 204)