Page MenuHome
Differential D9035 Diff 29335 profiles/tests/test_webhooks.py

Changeset View

Standalone View

View Options

profiles/tests/test_webhooks.py

  • This file was added.
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, secret='testsecret'):
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()}
@override_settings(BLENDER_ID={
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…
'BASE_URL': BLENDER_ID_BASE_URL,
'OAUTH_CLIENT': 'testoauthclient',
'OAUTH_SECRET': 'testoathsecret',
'WEBHOOK_USER_MODIFIED_SECRET': 'testsecret',
})
class WebhooksTest(TestCase):
def setUp(self):
self.factory = RequestFactory()
self.url = reverse('webhook-user-modified')
# 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/1cda54d605799b1f4b0dc080cb9a18ad.jpg',
})
responses.add(
responses.GET, f'{BLENDER_ID_BASE_URL}api/me',
json={
'id': 2,
'full_name': 'Jane Doe',
sybrenUnsubmitted
Done
Inline Actions

Include non-ASCII characters in names.

sybren: Include non-ASCII characters in names.
'email': 'jane@example.com',
'nickname': 'janedoe',
'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)
# 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):
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…
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(self):
body = {
'avatar_changed': False,
'email': 'newmail@example.com',
'full_name': 'Jane Doe',
'id': 2,
'old_email': 'mail@example.com',
'roles': {},
}
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, 'Jane Doe')
self.assertEquals(profile.user.email, 'newmail@example.com')
@responses.activate
def test_user_modified_avatar_changed(self):
body = {
'avatar_changed': True,
'email': 'newmail@example.com',
'full_name': 'Jane Doe',
'id': 2,
'old_email': 'mail@example.com',
'roles': [],
}
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'))
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.
@responses.activate
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 = {
'avatar_changed': False,
'email': 'mail@example.com',
'full_name': 'Jane Doe',
'id': 2,
'old_email': 'mail@example.com',
'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 = {
'avatar_changed': False,
'email': 'mail@example.com',
'full_name': 'Jane Doe',
'id': 2,
'old_email': 'mail@example.com',
'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)