mirror of
https://gitlab.com/allianceauth/allianceauth.git
synced 2025-07-10 13:00:16 +02:00
8987cf2199
Use django-redis-cache backend for locking get_or_set No longer require group-related tasks to be locked to one simultaneous execution. Remove legacy service group cache models. Truncate Discord nicknames to 32 characters Correct Discourse group name extension using only valid leading characters. Prevent name slicing from ending with illegal character Closes #801 Closes #847 Closes #835 Closes #852
456 lines
18 KiB
Python
456 lines
18 KiB
Python
from __future__ import unicode_literals
|
|
|
|
try:
|
|
# Py3
|
|
from unittest import mock
|
|
except ImportError:
|
|
# Py2
|
|
import mock
|
|
|
|
from django.test import TestCase, RequestFactory
|
|
from django.conf import settings
|
|
from django.contrib.auth.models import User, Group, Permission
|
|
from django.core.exceptions import ObjectDoesNotExist
|
|
|
|
from alliance_auth.tests.auth_utils import AuthUtils
|
|
|
|
from .auth_hooks import DiscordService
|
|
from .models import DiscordUser
|
|
from .tasks import DiscordTasks
|
|
from .manager import DiscordOAuthManager
|
|
|
|
import requests_mock
|
|
import datetime
|
|
|
|
MODULE_PATH = 'services.modules.discord'
|
|
|
|
|
|
def add_permissions():
|
|
permission = Permission.objects.get(codename='access_discord')
|
|
members = Group.objects.get(name=settings.DEFAULT_AUTH_GROUP)
|
|
blues = Group.objects.get(name=settings.DEFAULT_BLUE_GROUP)
|
|
AuthUtils.add_permissions_to_groups([permission], [members, blues])
|
|
|
|
|
|
class DiscordHooksTestCase(TestCase):
|
|
def setUp(self):
|
|
self.member = 'member_user'
|
|
member = AuthUtils.create_member(self.member)
|
|
DiscordUser.objects.create(user=member, uid='12345')
|
|
self.blue = 'blue_user'
|
|
blue = AuthUtils.create_blue(self.blue)
|
|
DiscordUser.objects.create(user=blue, uid='67891')
|
|
self.none_user = 'none_user'
|
|
none_user = AuthUtils.create_user(self.none_user)
|
|
self.service = DiscordService
|
|
add_permissions()
|
|
|
|
def test_has_account(self):
|
|
member = User.objects.get(username=self.member)
|
|
blue = User.objects.get(username=self.blue)
|
|
none_user = User.objects.get(username=self.none_user)
|
|
self.assertTrue(DiscordTasks.has_account(member))
|
|
self.assertTrue(DiscordTasks.has_account(blue))
|
|
self.assertFalse(DiscordTasks.has_account(none_user))
|
|
|
|
def test_service_enabled(self):
|
|
service = self.service()
|
|
member = User.objects.get(username=self.member)
|
|
blue = User.objects.get(username=self.blue)
|
|
none_user = User.objects.get(username=self.none_user)
|
|
|
|
self.assertTrue(service.service_active_for_user(member))
|
|
self.assertTrue(service.service_active_for_user(blue))
|
|
self.assertFalse(service.service_active_for_user(none_user))
|
|
|
|
@mock.patch(MODULE_PATH + '.tasks.DiscordOAuthManager')
|
|
def test_update_all_groups(self, manager):
|
|
service = self.service()
|
|
service.update_all_groups()
|
|
# Check member and blue user have groups updated
|
|
self.assertTrue(manager.update_groups.called)
|
|
self.assertEqual(manager.update_groups.call_count, 2)
|
|
|
|
def test_update_groups(self):
|
|
# Check member has Member group updated
|
|
with mock.patch(MODULE_PATH + '.tasks.DiscordOAuthManager') as manager:
|
|
service = self.service()
|
|
member = User.objects.get(username=self.member)
|
|
service.update_groups(member)
|
|
self.assertTrue(manager.update_groups.called)
|
|
args, kwargs = manager.update_groups.call_args
|
|
user_id, groups = args
|
|
self.assertIn(settings.DEFAULT_AUTH_GROUP, groups)
|
|
self.assertEqual(user_id, member.discord.uid)
|
|
|
|
# Check none user does not have groups updated
|
|
with mock.patch(MODULE_PATH + '.tasks.DiscordOAuthManager') as manager:
|
|
service = self.service()
|
|
none_user = User.objects.get(username=self.none_user)
|
|
service.update_groups(none_user)
|
|
self.assertFalse(manager.update_groups.called)
|
|
|
|
@mock.patch(MODULE_PATH + '.tasks.DiscordOAuthManager')
|
|
def test_validate_user(self, manager):
|
|
service = self.service()
|
|
# Test member is not deleted
|
|
member = User.objects.get(username=self.member)
|
|
service.validate_user(member)
|
|
self.assertTrue(member.discord)
|
|
|
|
# Test none user is deleted
|
|
none_user = User.objects.get(username=self.none_user)
|
|
DiscordUser.objects.create(user=none_user, uid='abc123')
|
|
service.validate_user(none_user)
|
|
self.assertTrue(manager.delete_user.called)
|
|
with self.assertRaises(ObjectDoesNotExist):
|
|
none_discord = User.objects.get(username=self.none_user).discord
|
|
|
|
@mock.patch(MODULE_PATH + '.tasks.DiscordOAuthManager')
|
|
def test_sync_nickname(self, manager):
|
|
service = self.service()
|
|
member = User.objects.get(username=self.member)
|
|
AuthUtils.add_main_character(member, 'test user', '12345', corp_ticker='AAUTH')
|
|
|
|
service.sync_nickname(member)
|
|
|
|
self.assertTrue(manager.update_nickname.called)
|
|
args, kwargs = manager.update_nickname.call_args
|
|
self.assertEqual(args[0], member.discord.uid)
|
|
self.assertEqual(args[1], 'test user')
|
|
|
|
@mock.patch(MODULE_PATH + '.tasks.DiscordOAuthManager')
|
|
def test_delete_user(self, manager):
|
|
member = User.objects.get(username=self.member)
|
|
|
|
service = self.service()
|
|
result = service.delete_user(member)
|
|
|
|
self.assertTrue(result)
|
|
self.assertTrue(manager.delete_user.called)
|
|
with self.assertRaises(ObjectDoesNotExist):
|
|
discord_user = User.objects.get(username=self.member).discord
|
|
|
|
def test_render_services_ctrl(self):
|
|
service = self.service()
|
|
member = User.objects.get(username=self.member)
|
|
request = RequestFactory().get('/en/services/')
|
|
request.user = member
|
|
|
|
response = service.render_services_ctrl(request)
|
|
self.assertTemplateUsed(service.service_ctrl_template)
|
|
self.assertIn('/discord/reset/', response)
|
|
self.assertIn('/discord/deactivate/', response)
|
|
|
|
# Test register becomes available
|
|
member.discord.delete()
|
|
member = User.objects.get(username=self.member)
|
|
request.user = member
|
|
response = service.render_services_ctrl(request)
|
|
self.assertIn('/discord/activate/', response)
|
|
|
|
# TODO: Test update nicknames
|
|
|
|
|
|
class DiscordViewsTestCase(TestCase):
|
|
def setUp(self):
|
|
self.member = AuthUtils.create_member('auth_member')
|
|
self.member.set_password('password')
|
|
self.member.save()
|
|
add_permissions()
|
|
|
|
def login(self):
|
|
self.client.login(username=self.member.username, password='password')
|
|
|
|
@mock.patch(MODULE_PATH + '.views.DiscordOAuthManager')
|
|
def test_activate(self, manager):
|
|
self.login()
|
|
manager.generate_oauth_redirect_url.return_value = '/example.com/oauth/'
|
|
response = self.client.get('/discord/activate/', follow=False)
|
|
self.assertRedirects(response, expected_url='/example.com/oauth/', target_status_code=404)
|
|
|
|
@mock.patch(MODULE_PATH + '.tasks.DiscordOAuthManager')
|
|
def test_callback(self, manager):
|
|
self.login()
|
|
manager.add_user.return_value = '1234'
|
|
response = self.client.get('/discord/callback/', data={'code': '1234'})
|
|
|
|
self.assertTrue(manager.add_user.called)
|
|
self.assertEqual(manager.update_nickname.called, settings.DISCORD_SYNC_NAMES)
|
|
self.assertEqual(self.member.discord.uid, '1234')
|
|
self.assertRedirects(response, expected_url='/en/services/', target_status_code=200)
|
|
|
|
@mock.patch(MODULE_PATH + '.tasks.DiscordOAuthManager')
|
|
def test_reset(self, manager):
|
|
self.login()
|
|
DiscordUser.objects.create(user=self.member, uid='12345')
|
|
manager.delete_user.return_value = True
|
|
|
|
response = self.client.get('/discord/reset/')
|
|
|
|
self.assertRedirects(response, expected_url='/discord/activate/', target_status_code=302)
|
|
|
|
@mock.patch(MODULE_PATH + '.tasks.DiscordOAuthManager')
|
|
def test_deactivate(self, manager):
|
|
self.login()
|
|
DiscordUser.objects.create(user=self.member, uid='12345')
|
|
manager.delete_user.return_value = True
|
|
|
|
response = self.client.get('/discord/deactivate/')
|
|
|
|
self.assertTrue(manager.delete_user.called)
|
|
self.assertRedirects(response, expected_url='/en/services/', target_status_code=200)
|
|
with self.assertRaises(ObjectDoesNotExist):
|
|
discord_user = User.objects.get(pk=self.member.pk).discord
|
|
|
|
|
|
class DiscordManagerTestCase(TestCase):
|
|
def setUp(self):
|
|
pass
|
|
|
|
def test__sanitize_groupname(self):
|
|
test_group_name = ' Group Name_Test_'
|
|
group_name = DiscordOAuthManager._sanitize_groupname(test_group_name)
|
|
|
|
self.assertEqual(group_name, 'GroupName_Test')
|
|
|
|
def test_generate_Bot_add_url(self):
|
|
from . import manager
|
|
bot_add_url = DiscordOAuthManager.generate_bot_add_url()
|
|
|
|
auth_url = manager.AUTH_URL
|
|
real_bot_add_url = '{}?client_id=appid&scope=bot&permissions={}'.format(auth_url, manager.BOT_PERMISSIONS)
|
|
self.assertEqual(bot_add_url, real_bot_add_url)
|
|
|
|
def test_generate_oauth_redirect_url(self):
|
|
from . import manager
|
|
import urllib
|
|
import sys
|
|
oauth_url = DiscordOAuthManager.generate_oauth_redirect_url()
|
|
|
|
self.assertIn(manager.AUTH_URL, oauth_url)
|
|
self.assertIn('+'.join(manager.SCOPES), oauth_url)
|
|
self.assertIn(settings.DISCORD_APP_ID, oauth_url)
|
|
if sys.version_info[0] < 3:
|
|
# Py2
|
|
self.assertIn(urllib.quote_plus(settings.DISCORD_CALLBACK_URL), oauth_url)
|
|
else: # Py3
|
|
self.assertIn(urllib.parse.quote_plus(settings.DISCORD_CALLBACK_URL), oauth_url)
|
|
|
|
@mock.patch(MODULE_PATH + '.manager.OAuth2Session')
|
|
def test__process_callback_code(self, oauth):
|
|
from . import manager
|
|
instance = oauth.return_value
|
|
instance.fetch_token.return_value = {'access_token': 'mywonderfultoken'}
|
|
|
|
token = DiscordOAuthManager._process_callback_code('12345')
|
|
|
|
self.assertTrue(oauth.called)
|
|
args, kwargs = oauth.call_args
|
|
self.assertEqual(args[0], settings.DISCORD_APP_ID)
|
|
self.assertEqual(kwargs['redirect_uri'], settings.DISCORD_CALLBACK_URL)
|
|
self.assertTrue(instance.fetch_token.called)
|
|
args, kwargs = instance.fetch_token.call_args
|
|
self.assertEqual(args[0], manager.TOKEN_URL)
|
|
self.assertEqual(kwargs['client_secret'], settings.DISCORD_APP_SECRET)
|
|
self.assertEqual(kwargs['code'], '12345')
|
|
self.assertEqual(token['access_token'], 'mywonderfultoken')
|
|
|
|
@mock.patch(MODULE_PATH + '.manager.DiscordOAuthManager._process_callback_code')
|
|
@requests_mock.Mocker()
|
|
def test_add_user(self, oauth_token, m):
|
|
from . import manager
|
|
import json
|
|
|
|
# Arrange
|
|
oauth_token.return_value = {'access_token': 'accesstoken'}
|
|
|
|
headers = {'accept': 'application/json', 'authorization': 'Bearer accesstoken'}
|
|
|
|
m.register_uri('POST',
|
|
manager.DISCORD_URL + '/invites/'+str(settings.DISCORD_INVITE_CODE),
|
|
request_headers=headers,
|
|
text='{}')
|
|
|
|
m.register_uri('GET',
|
|
manager.DISCORD_URL + "/users/@me",
|
|
request_headers=headers,
|
|
text=json.dumps({'id': "123456"}))
|
|
|
|
# Act
|
|
return_value = DiscordOAuthManager.add_user('abcdef')
|
|
|
|
# Assert
|
|
self.assertEqual(return_value, '123456')
|
|
self.assertEqual(m.call_count, 2)
|
|
|
|
@requests_mock.Mocker()
|
|
def test_delete_user(self, m):
|
|
from . import manager
|
|
import json
|
|
|
|
# Arrange
|
|
headers = {'accept': 'application/json', 'authorization': 'Bot ' + settings.DISCORD_BOT_TOKEN}
|
|
|
|
user_id = 12345
|
|
request_url = '{}/guilds/{}/members/{}'.format(manager.DISCORD_URL, settings.DISCORD_GUILD_ID, user_id)
|
|
m.register_uri('DELETE',
|
|
request_url,
|
|
request_headers=headers,
|
|
text=json.dumps({}))
|
|
|
|
# Act
|
|
result = DiscordOAuthManager.delete_user(user_id)
|
|
|
|
# Assert
|
|
self.assertTrue(result)
|
|
|
|
###
|
|
# Test 404 (already deleted)
|
|
# Arrange
|
|
m.register_uri('DELETE',
|
|
request_url,
|
|
request_headers=headers,
|
|
status_code=404)
|
|
|
|
# Act
|
|
result = DiscordOAuthManager.delete_user(user_id)
|
|
|
|
# Assert
|
|
self.assertTrue(result)
|
|
|
|
###
|
|
# Test 500 (some random API error)
|
|
# Arrange
|
|
m.register_uri('DELETE',
|
|
request_url,
|
|
request_headers=headers,
|
|
status_code=500)
|
|
|
|
# Act
|
|
result = DiscordOAuthManager.delete_user(user_id)
|
|
|
|
# Assert
|
|
self.assertFalse(result)
|
|
|
|
@requests_mock.Mocker()
|
|
def test_update_nickname(self, m):
|
|
from . import manager
|
|
import json
|
|
# Arrange
|
|
headers = {'content-type': 'application/json', 'authorization': 'Bot ' + settings.DISCORD_BOT_TOKEN}
|
|
|
|
user_id = 12345
|
|
request_url = '{}/guilds/{}/members/{}'.format(manager.DISCORD_URL, settings.DISCORD_GUILD_ID, user_id)
|
|
m.patch(request_url,
|
|
request_headers=headers)
|
|
|
|
# Act
|
|
result = DiscordOAuthManager.update_nickname(user_id, 'somenick')
|
|
|
|
# Assert
|
|
self.assertTrue(result)
|
|
|
|
@mock.patch(MODULE_PATH + '.manager.DiscordOAuthManager._get_groups')
|
|
@requests_mock.Mocker()
|
|
def test_update_groups(self, group_cache, m):
|
|
from . import manager
|
|
import json
|
|
|
|
# Arrange
|
|
groups = ['Member', 'Blue', 'Special Group']
|
|
|
|
group_cache.return_value = [{'id': 111, 'name': 'Member'},
|
|
{'id': 222, 'name': 'Blue'},
|
|
{'id': 333, 'name': 'SpecialGroup'},
|
|
{'id': 444, 'name': 'NotYourGroup'}]
|
|
|
|
headers = {'content-type': 'application/json', 'authorization': 'Bot ' + settings.DISCORD_BOT_TOKEN}
|
|
user_id = 12345
|
|
request_url = '{}/guilds/{}/members/{}'.format(manager.DISCORD_URL, settings.DISCORD_GUILD_ID, user_id)
|
|
|
|
m.patch(request_url,
|
|
request_headers=headers)
|
|
|
|
# Act
|
|
DiscordOAuthManager.update_groups(user_id, groups)
|
|
|
|
# Assert
|
|
self.assertEqual(len(m.request_history), 1, 'Must be one HTTP call made')
|
|
history = json.loads(m.request_history[0].text)
|
|
self.assertIn('roles', history, "'The request must send JSON object with the 'roles' key")
|
|
self.assertIn(111, history['roles'], 'The group id 111 must be added to the request')
|
|
self.assertIn(222, history['roles'], 'The group id 222 must be added to the request')
|
|
self.assertIn(333, history['roles'], 'The group id 333 must be added to the request')
|
|
self.assertNotIn(444, history['roles'], 'The group id 444 must NOT be added to the request')
|
|
|
|
@mock.patch(MODULE_PATH + '.manager.cache')
|
|
@mock.patch(MODULE_PATH + '.manager.DiscordOAuthManager._get_groups')
|
|
@requests_mock.Mocker()
|
|
def test_update_groups_backoff(self, group_cache, djcache, m):
|
|
from . import manager
|
|
|
|
# Arrange
|
|
groups = ['Member']
|
|
group_cache.return_value = [{'id': 111, 'name': 'Member'}]
|
|
|
|
headers = {'content-type': 'application/json', 'authorization': 'Bot ' + settings.DISCORD_BOT_TOKEN}
|
|
user_id = 12345
|
|
request_url = '{}/guilds/{}/members/{}'.format(manager.DISCORD_URL, settings.DISCORD_GUILD_ID, user_id)
|
|
|
|
djcache.get.return_value = None # No existing backoffs in cache
|
|
|
|
m.patch(request_url,
|
|
request_headers=headers,
|
|
headers={'Retry-After': '200'},
|
|
status_code=429)
|
|
|
|
# Act & Assert
|
|
with self.assertRaises(manager.DiscordApiBackoff) as bo:
|
|
try:
|
|
DiscordOAuthManager.update_groups(user_id, groups, blocking=False)
|
|
except manager.DiscordApiBackoff as bo:
|
|
self.assertEqual(bo.retry_after, 200, 'Retry-After time must be equal to Retry-After set in header')
|
|
self.assertFalse(bo.global_ratelimit, 'global_ratelimit must be False')
|
|
raise bo
|
|
|
|
self.assertTrue(djcache.set.called)
|
|
args, kwargs = djcache.set.call_args
|
|
self.assertEqual(args[0], 'DISCORD_BACKOFF_update_groups')
|
|
self.assertTrue(datetime.datetime.strptime(args[1], manager.cache_time_format) > datetime.datetime.now())
|
|
|
|
@mock.patch(MODULE_PATH + '.manager.cache')
|
|
@mock.patch(MODULE_PATH + '.manager.DiscordOAuthManager._get_groups')
|
|
@requests_mock.Mocker()
|
|
def test_update_groups_global_backoff(self, group_cache, djcache, m):
|
|
from . import manager
|
|
|
|
# Arrange
|
|
groups = ['Member']
|
|
group_cache.return_value = [{'id': 111, 'name': 'Member'}]
|
|
|
|
headers = {'content-type': 'application/json', 'authorization': 'Bot ' + settings.DISCORD_BOT_TOKEN}
|
|
user_id = 12345
|
|
request_url = '{}/guilds/{}/members/{}'.format(manager.DISCORD_URL, settings.DISCORD_GUILD_ID, user_id)
|
|
|
|
djcache.get.return_value = None # No existing backoffs in cache
|
|
|
|
m.patch(request_url,
|
|
request_headers=headers,
|
|
headers={'Retry-After': '200', 'X-RateLimit-Global': 'true'},
|
|
status_code=429)
|
|
|
|
# Act & Assert
|
|
with self.assertRaises(manager.DiscordApiBackoff) as bo:
|
|
try:
|
|
DiscordOAuthManager.update_groups(user_id, groups, blocking=False)
|
|
except manager.DiscordApiBackoff as bo:
|
|
self.assertEqual(bo.retry_after, 200, 'Retry-After time must be equal to Retry-After set in header')
|
|
self.assertTrue(bo.global_ratelimit, 'global_ratelimit must be True')
|
|
raise bo
|
|
|
|
self.assertTrue(djcache.set.called)
|
|
args, kwargs = djcache.set.call_args
|
|
self.assertEqual(args[0], 'DISCORD_BACKOFF_GLOBAL')
|
|
self.assertTrue(datetime.datetime.strptime(args[1], manager.cache_time_format) > datetime.datetime.now())
|