mirror of
https://gitlab.com/allianceauth/allianceauth.git
synced 2026-02-06 23:26:19 +01:00
The Great Services Refactor (#594)
* Hooks registration, discovery and retrieval module Will discover @hooks.register decorated functions inside the auth_hooks module in any installed django app. * Class to register modular service apps * Register service modules URLs * Example service module * Refactor services into modules Each service type has been split out into its own django app/module. A hook mechanism is provided to register a subclass of the ServiceHook class. The modules then overload functions defined in ServiceHook as required to provide interoperability with alliance auth. Service modules provide their own urls and views for user registration and account management and a partial template to display on the services page. Where possible, new modules should provide their own models for local data storage. * Added menu items hooks and template tags * Added menu item hook for broadcasts * Added str method to ServicesHook * Added exception handling to hook iterators * Refactor mumble migration and table name Upgrading will require `migrate mumble --fake-initial` to be run first and then `migrate mumble` to rename the table. * Refactor teamspeak3 migration and rename table Upgrading will require `migrate teamspeak3 --fake-initial` * Added module models and migrations for refactoring AuthServicesInfo * Migrate AuthServiceInfo fields to service modules models * Added helper for getting a users main character * Added new style celery instance * Changed Discord from AuthServicesInfo to DiscordUser model * Switch celery tasks to staticmethods * Changed Discourse from AuthServicesInfo to DiscourseUser model * Changed IPBoard from AuthServicesInfo to IpboardUser model * Changed Ips4 from AuthServicesInfo to Ips4User model Also added disable service task. This service still needs some love though. Was always missing a deactivate services hook (before refactoring) for reasons I'm unsure of so I'm reluctant to add it without knowing why. * Changed Market from AuthServicesInfo to MarketUser model * Changed Mumble from AuthServicesInfo to MumbleUser model Switched user foreign key to one to one relationship. Removed implicit password change on user exists. Combined regular and blue user creation. * Changed Openfire from AuthServicesInfo to OpenfireUser model * Changed SMF from AuthServicesInfo to SmfUser model Added disable task * Changed Phpbb3 from AuthServicesInfo to Phpbb3User model * Changed XenForo from AuthServicesInfo to XenforoUser model * Changed Teamspeak3 from AuthServicesInfo to Teamspeak3User model * Remove obsolete manager functions * Standardise URL format This will break some callback URLs Discord changes from /discord_callback/ to /discord/callback/ * Removed unnecessary imports * Mirror upstream decorator change * Setup for unit testing * Unit tests for discord service * Added add main character helper * Added Discourse unit tests * Added Ipboard unit tests * Added Ips4 unit tests * Fix naming of market manager, switch to use class methods * Remove unused hook functions * Added market service unit tests * Added corp ticker to add main character helper * Added mumble unit tests * Fix url name and remove namespace * Fix missing return and add missing URL * Added openfire unit tests * Added missing return * Added phpbb3 unit tests * Fix SmfManager naming inconsistency and switch to classmethods * Added smf unit tests * Remove unused functions, Added missing return * Added xenforo unit tests * Added missing return * Fixed reference to old model * Fixed error preventing groups from syncing on reset request * Added teamspeak3 unit tests * Added nose as test runner and some test settings * Added package requirements for running tests * Added unit tests for services signals and tasks * Remove unused tests file * Fix teamspeak3 service signals * Added unit tests for teamspeak3 signals Changed other unit tests setUp to inert signals * Fix password gen and hashing python3 compatibility Fixes #630 Adds unit tests to check the password functions run on both platforms. * Fix unit test to not rely on checking url params * Add Travis CI settings file * Remove default blank values from services models * Added dynamic user model admin actions for syncing service groups * Remove unused search fields * Add hook function for syncing nicknames * Added discord hook for sync nickname * Added user admin model menu actions for sync nickname hook * Remove obsolete code * Rename celery config app to avoid package name clash * Added new style celerybeat schedule configuration periodic_task decorator is depreciated * Added string representations * Added admin pages for services user models * Removed legacy code * Move link discord button to correct template * Remove blank default fields from example model * Disallow empty django setting * Fix typos * Added coverage configuration file * Add coverage and coveralls to travis config Should probably use nose's built in coverage, but this works for now. * Replace AuthServicesInfo get_or_create instances with get Reflects upstream changes to AuthServicesInfo behaviour. * Update mumble user table name * Split out mumble authenticator requirements zeroc-ice seems to cause long build times on travis-ci and isn't required for the core projects functionality or testing.
This commit is contained in:
0
services/modules/ipboard/__init__.py
Normal file
0
services/modules/ipboard/__init__.py
Normal file
10
services/modules/ipboard/admin.py
Normal file
10
services/modules/ipboard/admin.py
Normal file
@@ -0,0 +1,10 @@
|
||||
from __future__ import unicode_literals
|
||||
from django.contrib import admin
|
||||
from .models import IpboardUser
|
||||
|
||||
|
||||
class IpboardUserAdmin(admin.ModelAdmin):
|
||||
list_display = ('user', 'username')
|
||||
search_fields = ('user__username', 'username')
|
||||
|
||||
admin.site.register(IpboardUser, IpboardUserAdmin)
|
||||
7
services/modules/ipboard/apps.py
Normal file
7
services/modules/ipboard/apps.py
Normal file
@@ -0,0 +1,7 @@
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class IpboardServiceConfig(AppConfig):
|
||||
name = 'ipboard'
|
||||
68
services/modules/ipboard/auth_hooks.py
Normal file
68
services/modules/ipboard/auth_hooks.py
Normal file
@@ -0,0 +1,68 @@
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from django.conf import settings
|
||||
from django.template.loader import render_to_string
|
||||
|
||||
from services.hooks import ServicesHook
|
||||
from alliance_auth import hooks
|
||||
|
||||
from .urls import urlpatterns
|
||||
from .tasks import IpboardTasks
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class IpboardService(ServicesHook):
|
||||
def __init__(self):
|
||||
ServicesHook.__init__(self)
|
||||
self.name = 'ipboard'
|
||||
self.service_url = settings.IPBOARD_ENDPOINT
|
||||
self.urlpatterns = urlpatterns
|
||||
|
||||
@property
|
||||
def title(self):
|
||||
return 'IPBoard Forums'
|
||||
|
||||
def delete_user(self, user, notify_user=False):
|
||||
logger.debug('Deleting user %s %s account' % (user, self.name))
|
||||
return IpboardTasks.delete_user(user, notify_user=notify_user)
|
||||
|
||||
def update_groups(self, user):
|
||||
logger.debug("Updating %s groups for %s" % (self.name, user))
|
||||
if IpboardTasks.has_account(user):
|
||||
IpboardTasks.update_groups.delay(user.pk)
|
||||
|
||||
def validate_user(self, user):
|
||||
logger.debug('Validating user %s %s account' % (user, self.name))
|
||||
if IpboardTasks.has_account(user) and not self.service_active_for_user(user):
|
||||
self.delete_user(user, notify_user=True)
|
||||
|
||||
def update_all_groups(self):
|
||||
logger.debug('Update all %s groups called' % self.name)
|
||||
IpboardTasks.update_all_groups.delay()
|
||||
|
||||
def service_enabled_members(self):
|
||||
return settings.ENABLE_AUTH_IPBOARD or False
|
||||
|
||||
def service_enabled_blues(self):
|
||||
return settings.ENABLE_BLUE_IPBOARD or False
|
||||
|
||||
def render_services_ctrl(self, request):
|
||||
urls = self.Urls()
|
||||
urls.auth_activate = 'auth_activate_ipboard'
|
||||
urls.auth_deactivate = 'auth_deactivate_ipboard'
|
||||
urls.auth_reset_password = 'auth_reset_ipboard_password'
|
||||
urls.auth_set_password = 'auth_set_ipboard_password'
|
||||
return render_to_string(self.service_ctrl_template, {
|
||||
'service_name': self.title,
|
||||
'urls': urls,
|
||||
'service_url': self.service_url,
|
||||
'username': request.user.ipboard.username if IpboardTasks.has_account(request.user) else '',
|
||||
}, request=request)
|
||||
|
||||
|
||||
@hooks.register('services_hook')
|
||||
def register_service():
|
||||
return IpboardService()
|
||||
161
services/modules/ipboard/manager.py
Executable file
161
services/modules/ipboard/manager.py
Executable file
@@ -0,0 +1,161 @@
|
||||
from __future__ import unicode_literals
|
||||
import random
|
||||
import string
|
||||
import re
|
||||
from hashlib import md5
|
||||
try:
|
||||
from xmlrpclib import Server
|
||||
except ImportError:
|
||||
# python 3
|
||||
from xmlrpc import server as Server
|
||||
|
||||
|
||||
from django.conf import settings
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class IPBoardManager:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def __santatize_username(username):
|
||||
sanatized = username.replace(" ", "_")
|
||||
return sanatized.lower()
|
||||
|
||||
@staticmethod
|
||||
def __generate_random_pass():
|
||||
return ''.join([random.choice(string.ascii_letters + string.digits) for n in range(16)])
|
||||
|
||||
@staticmethod
|
||||
def _gen_pwhash(password):
|
||||
return md5(password.encode('utf-8')).hexdigest()
|
||||
|
||||
@staticmethod
|
||||
def _sanitize_groupname(name):
|
||||
name = name.strip(' _')
|
||||
return re.sub('[^\w.-]', '', name)
|
||||
|
||||
@staticmethod
|
||||
def exec_xmlrpc(func, **kwargs):
|
||||
""" Send a XMLRPC request """
|
||||
try:
|
||||
server = Server(settings.IPBOARD_ENDPOINT, verbose=False)
|
||||
params = {}
|
||||
for i in kwargs:
|
||||
params[i] = kwargs[i]
|
||||
params['api_key'] = settings.IPBOARD_APIKEY
|
||||
params['api_module'] = settings.IPBOARD_APIMODULE
|
||||
|
||||
return getattr(server, func)(params)
|
||||
except:
|
||||
return {}
|
||||
|
||||
@classmethod
|
||||
def add_user(cls, username, email):
|
||||
""" Add user to service """
|
||||
sanatized = str(IPBoardManager.__santatize_username(username))
|
||||
logger.debug("Adding user to IPBoard with username %s" % sanatized)
|
||||
plain_password = IPBoardManager.__generate_random_pass()
|
||||
password = cls._gen_pwhash(plain_password)
|
||||
IPBoardManager.exec_xmlrpc('createUser', username=sanatized, email=str(email), display_name=sanatized,
|
||||
md5_passwordHash=password)
|
||||
logger.info("Added IPBoard user with username %s" % sanatized)
|
||||
return sanatized, plain_password
|
||||
|
||||
@staticmethod
|
||||
def delete_user(username):
|
||||
""" Delete user """
|
||||
IPBoardManager.exec_xmlrpc('deleteUser', username=username)
|
||||
logger.info("Deleted IPBoard user with username %s" % username)
|
||||
return username
|
||||
|
||||
@staticmethod
|
||||
def disable_user(username):
|
||||
""" Disable user """
|
||||
IPBoardManager.exec_xmlrpc('disableUser', username=username)
|
||||
logger.info("Disabled IPBoard user with username %s" % username)
|
||||
return username
|
||||
|
||||
@classmethod
|
||||
def update_user(cls, username, email, password):
|
||||
""" Add user to service """
|
||||
password = cls._gen_pwhash(password)
|
||||
logger.debug("Updating IPBoard username %s with email %s and password hash starting with %s" % (
|
||||
username, email, password[0:5]))
|
||||
IPBoardManager.exec_xmlrpc('updateUser', username=username, email=email, md5_passwordHash=password)
|
||||
logger.info("Updated IPBoard user with username %s" % username)
|
||||
return username
|
||||
|
||||
@staticmethod
|
||||
def get_all_groups():
|
||||
groups = []
|
||||
ret = IPBoardManager.exec_xmlrpc('getAllGroups')
|
||||
for group in ret:
|
||||
groups.append(group["g_title"])
|
||||
logger.debug("Retrieved group list from IPBoard: %s" % groups)
|
||||
return groups
|
||||
|
||||
@staticmethod
|
||||
def get_user_groups(username):
|
||||
groups = []
|
||||
ret = IPBoardManager.exec_xmlrpc('getUserGroups', username=username)
|
||||
if type(ret) is list:
|
||||
for group in ret:
|
||||
groups.append(group["g_title"])
|
||||
logger.debug("Got user %s IPBoard groups %s" % (username, groups))
|
||||
return groups
|
||||
|
||||
@staticmethod
|
||||
def add_group(group):
|
||||
ret = IPBoardManager.exec_xmlrpc('addGroup', group=group)
|
||||
logger.info("Added IPBoard group %s" % group)
|
||||
return ret
|
||||
|
||||
@staticmethod
|
||||
def add_user_to_group(username, group):
|
||||
ret = IPBoardManager.exec_xmlrpc('addUserToGroup', username=username, group=group)
|
||||
logger.info("Added IPBoard user %s to group %s" % (username, group))
|
||||
return ret
|
||||
|
||||
@staticmethod
|
||||
def remove_user_from_group(username, group):
|
||||
ret = IPBoardManager.exec_xmlrpc('removeUserFromGroup', username=username, group=group)
|
||||
logger.info("Removed IPBoard user %s from group %s" % (username, group))
|
||||
return ret
|
||||
|
||||
@staticmethod
|
||||
def help_me():
|
||||
ret = IPBoardManager.exec_xmlrpc('helpMe')
|
||||
return ret
|
||||
|
||||
@staticmethod
|
||||
def update_groups(username, groups):
|
||||
logger.debug("Updating IPBoard user %s with groups %s" % (username, groups))
|
||||
forum_groups = IPBoardManager.get_all_groups()
|
||||
user_groups = set(IPBoardManager.get_user_groups(username))
|
||||
act_groups = set([IPBoardManager._sanitize_groupname(g) for g in groups])
|
||||
addgroups = act_groups - user_groups
|
||||
remgroups = user_groups - act_groups
|
||||
|
||||
logger.info("Updating IPBoard groups for user %s - adding %s, removing %s" % (username, addgroups, remgroups))
|
||||
for g in addgroups:
|
||||
if g not in forum_groups:
|
||||
IPBoardManager.add_group(g)
|
||||
logger.debug("Adding user %s to IPBoard group %s" % (username, g))
|
||||
IPBoardManager.add_user_to_group(username, g)
|
||||
|
||||
for g in remgroups:
|
||||
logger.debug("Removing user %s from IPBoard group %s" % (username, g))
|
||||
IPBoardManager.remove_user_from_group(username, g)
|
||||
|
||||
@staticmethod
|
||||
def update_user_password(username, email, plain_password=None):
|
||||
logger.debug("Settings new IPBoard password for user %s" % username)
|
||||
if not plain_password:
|
||||
plain_password = IPBoardManager.__generate_random_pass()
|
||||
IPBoardManager.update_user(username, email, plain_password)
|
||||
return plain_password
|
||||
26
services/modules/ipboard/migrations/0001_initial.py
Normal file
26
services/modules/ipboard/migrations/0001_initial.py
Normal file
@@ -0,0 +1,26 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# Generated by Django 1.10.2 on 2016-12-12 03:27
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
import django.db.models.deletion
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
initial = True
|
||||
|
||||
dependencies = [
|
||||
('auth', '0008_alter_user_username_max_length'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='IpboardUser',
|
||||
fields=[
|
||||
('user', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, primary_key=True, related_name='ipboard', serialize=False, to=settings.AUTH_USER_MODEL)),
|
||||
('username', models.CharField(max_length=254)),
|
||||
],
|
||||
),
|
||||
]
|
||||
0
services/modules/ipboard/migrations/__init__.py
Normal file
0
services/modules/ipboard/migrations/__init__.py
Normal file
16
services/modules/ipboard/models.py
Normal file
16
services/modules/ipboard/models.py
Normal file
@@ -0,0 +1,16 @@
|
||||
from __future__ import unicode_literals
|
||||
from django.utils.encoding import python_2_unicode_compatible
|
||||
from django.contrib.auth.models import User
|
||||
from django.db import models
|
||||
|
||||
|
||||
@python_2_unicode_compatible
|
||||
class IpboardUser(models.Model):
|
||||
user = models.OneToOneField(User,
|
||||
primary_key=True,
|
||||
on_delete=models.CASCADE,
|
||||
related_name='ipboard')
|
||||
username = models.CharField(max_length=254)
|
||||
|
||||
def __str__(self):
|
||||
return self.username
|
||||
73
services/modules/ipboard/tasks.py
Normal file
73
services/modules/ipboard/tasks.py
Normal file
@@ -0,0 +1,73 @@
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from alliance_auth.celeryapp import app
|
||||
from django.conf import settings
|
||||
from django.contrib.auth.models import User
|
||||
from django.core.exceptions import ObjectDoesNotExist
|
||||
from notifications import notify
|
||||
|
||||
from .manager import IPBoardManager
|
||||
from .models import IpboardUser
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class IpboardTasks:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def delete_user(cls, user, notify_user=False):
|
||||
if cls.has_account(user):
|
||||
if IPBoardManager.disable_user(user.ipboard.username):
|
||||
user.ipboard.delete()
|
||||
if notify_user:
|
||||
notify(user, 'IPBoard Account Disabled', level='danger')
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def has_account(user):
|
||||
try:
|
||||
return user.ipboard.username != ''
|
||||
except ObjectDoesNotExist:
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
@app.task(bind=True)
|
||||
def update_groups(self, pk):
|
||||
user = User.objects.get(pk=pk)
|
||||
logger.debug("Updating user %s ipboard groups." % user)
|
||||
groups = []
|
||||
for group in user.groups.all():
|
||||
groups.append(str(group.name))
|
||||
if len(groups) == 0:
|
||||
groups.append('empty')
|
||||
logger.debug("Updating user %s ipboard groups to %s" % (user, groups))
|
||||
try:
|
||||
IPBoardManager.update_groups(user.ipboard.username, groups)
|
||||
except:
|
||||
logger.exception("IPBoard group sync failed for %s, retrying in 10 mins" % user)
|
||||
raise self.retry(countdown=60 * 10)
|
||||
logger.debug("Updated user %s ipboard groups." % user)
|
||||
|
||||
@staticmethod
|
||||
@app.task
|
||||
def update_all_groups():
|
||||
logger.debug("Updating ALL ipboard groups")
|
||||
for ipboard_user in IpboardUser.objects.exclude(username__exact=''):
|
||||
IpboardTasks.update_groups.delay(ipboard_user.user.pk)
|
||||
|
||||
@staticmethod
|
||||
@app.task
|
||||
def disable():
|
||||
if settings.ENABLE_AUTH_IPBOARD:
|
||||
logger.warn(
|
||||
"ENABLE_AUTH_IPBOARD still True, after disabling users will still be able to create IPBoard accounts")
|
||||
if settings.ENABLE_BLUE_IPBOARD:
|
||||
logger.warn(
|
||||
"ENABLE_BLUE_IPBOARD still True, after disabling blues will still be able to create IPBoard accounts")
|
||||
logger.debug("Deleting all Ipboard Users")
|
||||
IpboardUser.objects.all().delete()
|
||||
207
services/modules/ipboard/tests.py
Normal file
207
services/modules/ipboard/tests.py
Normal file
@@ -0,0 +1,207 @@
|
||||
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
|
||||
from django import urls
|
||||
from django.core.exceptions import ObjectDoesNotExist
|
||||
|
||||
from alliance_auth.tests.auth_utils import AuthUtils
|
||||
|
||||
from .auth_hooks import IpboardService
|
||||
from .models import IpboardUser
|
||||
from .tasks import IpboardTasks
|
||||
from .manager import IPBoardManager
|
||||
|
||||
MODULE_PATH = 'services.modules.ipboard'
|
||||
|
||||
|
||||
class IpboardHooksTestCase(TestCase):
|
||||
def setUp(self):
|
||||
self.member = 'member_user'
|
||||
member = AuthUtils.create_member(self.member)
|
||||
IpboardUser.objects.create(user=member, username=self.member)
|
||||
self.blue = 'blue_user'
|
||||
blue = AuthUtils.create_blue(self.blue)
|
||||
IpboardUser.objects.create(user=blue, username=self.blue)
|
||||
self.none_user = 'none_user'
|
||||
none_user = AuthUtils.create_user(self.none_user)
|
||||
self.service = IpboardService
|
||||
|
||||
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(IpboardTasks.has_account(member))
|
||||
self.assertTrue(IpboardTasks.has_account(blue))
|
||||
self.assertFalse(IpboardTasks.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_enabled_members())
|
||||
self.assertTrue(service.service_enabled_blues())
|
||||
|
||||
self.assertEqual(service.service_active_for_user(member), settings.ENABLE_AUTH_IPBOARD)
|
||||
self.assertEqual(service.service_active_for_user(blue), settings.ENABLE_BLUE_IPBOARD)
|
||||
self.assertFalse(service.service_active_for_user(none_user))
|
||||
|
||||
@mock.patch(MODULE_PATH + '.tasks.IPBoardManager')
|
||||
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.IPBoardManager') as manager:
|
||||
service = self.service()
|
||||
member = User.objects.get(username=self.member)
|
||||
service.update_groups(member)
|
||||
self.assertTrue(manager.update_groups.called)
|
||||
|
||||
# Check none user does not have groups updated
|
||||
with mock.patch(MODULE_PATH + '.tasks.IPBoardManager') 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.IPBoardManager')
|
||||
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(User.objects.get(username=self.member).ipboard)
|
||||
|
||||
# Test none user is deleted
|
||||
none_user = User.objects.get(username=self.none_user)
|
||||
IpboardUser.objects.create(user=none_user, username='none_user')
|
||||
service.validate_user(none_user)
|
||||
self.assertTrue(manager.disable_user.called)
|
||||
with self.assertRaises(ObjectDoesNotExist):
|
||||
none_ipboard = User.objects.get(username=self.none_user).ipboard
|
||||
|
||||
@mock.patch(MODULE_PATH + '.tasks.IPBoardManager')
|
||||
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.disable_user.called)
|
||||
with self.assertRaises(ObjectDoesNotExist):
|
||||
Ipboard_user = User.objects.get(username=self.member).ipboard
|
||||
|
||||
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('/ipboard/set_password/', response)
|
||||
self.assertIn('/ipboard/reset_password/', response)
|
||||
self.assertIn('/ipboard/deactivate/', response)
|
||||
|
||||
# Test register becomes available
|
||||
member.ipboard.delete()
|
||||
member = User.objects.get(username=self.member)
|
||||
request.user = member
|
||||
response = service.render_services_ctrl(request)
|
||||
self.assertIn('/ipboard/activate/', response)
|
||||
|
||||
|
||||
class IpboardViewsTestCase(TestCase):
|
||||
def setUp(self):
|
||||
self.member = AuthUtils.create_member('auth_member')
|
||||
self.member.set_password('password')
|
||||
self.member.email = 'auth_member@example.com'
|
||||
self.member.save()
|
||||
AuthUtils.add_main_character(self.member, 'auth_member', '12345', corp_id='111', corp_name='Test Corporation')
|
||||
|
||||
def login(self):
|
||||
self.client.login(username=self.member.username, password='password')
|
||||
|
||||
@mock.patch(MODULE_PATH + '.views.IPBoardManager')
|
||||
def test_activate(self, manager):
|
||||
self.login()
|
||||
expected_username = 'auth_member'
|
||||
expected_password = 'abc123'
|
||||
manager.add_user.return_value = (expected_username, expected_password)
|
||||
response = self.client.get(urls.reverse('auth_activate_ipboard'), follow=False)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertContains(response, expected_username)
|
||||
self.assertContains(response, expected_password)
|
||||
self.assertTrue(manager.add_user.called)
|
||||
args, kwargs = manager.add_user.call_args
|
||||
self.assertEqual(args[0], 'auth_member') # Character name
|
||||
self.assertEqual(args[1], self.member.email)
|
||||
self.assertEqual(self.member.ipboard.username, expected_username)
|
||||
|
||||
@mock.patch(MODULE_PATH + '.tasks.IPBoardManager')
|
||||
def test_deactivate(self, manager):
|
||||
self.login()
|
||||
IpboardUser.objects.create(user=self.member, username='12345')
|
||||
manager.disable_user.return_value = True
|
||||
|
||||
response = self.client.get(urls.reverse('auth_deactivate_ipboard'))
|
||||
|
||||
self.assertTrue(manager.disable_user.called)
|
||||
self.assertRedirects(response, expected_url=urls.reverse('auth_services'), target_status_code=200)
|
||||
with self.assertRaises(ObjectDoesNotExist):
|
||||
ipboard_user = User.objects.get(pk=self.member.pk).ipboard
|
||||
|
||||
@mock.patch(MODULE_PATH + '.views.IPBoardManager')
|
||||
def test_set_password(self, manager):
|
||||
self.login()
|
||||
IpboardUser.objects.create(user=self.member, username='12345')
|
||||
expected_password = 'password'
|
||||
manager.update_user_password.return_value = expected_password
|
||||
|
||||
response = self.client.post(urls.reverse('auth_set_ipboard_password'), data={'password': expected_password})
|
||||
|
||||
self.assertTrue(manager.update_user_password.called)
|
||||
args, kwargs = manager.update_user_password.call_args
|
||||
self.assertEqual(kwargs['plain_password'], expected_password)
|
||||
self.assertRedirects(response, expected_url=urls.reverse('auth_services'), target_status_code=200)
|
||||
|
||||
@mock.patch(MODULE_PATH + '.views.IPBoardManager')
|
||||
def test_reset_password(self, manager):
|
||||
self.login()
|
||||
IpboardUser.objects.create(user=self.member, username='12345')
|
||||
|
||||
response = self.client.get(urls.reverse('auth_reset_ipboard_password'))
|
||||
|
||||
self.assertTrue(manager.update_user_password.called)
|
||||
self.assertTemplateUsed(response, 'registered/service_credentials.html')
|
||||
|
||||
|
||||
class IpboardManagerTestCase(TestCase):
|
||||
def setUp(self):
|
||||
self.manager = IPBoardManager
|
||||
|
||||
def test_generate_random_password(self):
|
||||
password = self.manager._IPBoardManager__generate_random_pass()
|
||||
|
||||
self.assertEqual(len(password), 16)
|
||||
self.assertIsInstance(password, type(''))
|
||||
|
||||
def test_gen_pwhash(self):
|
||||
pwhash = self.manager._gen_pwhash('test')
|
||||
|
||||
self.assertEqual(pwhash, '098f6bcd4621d373cade4e832627b4f6')
|
||||
16
services/modules/ipboard/urls.py
Normal file
16
services/modules/ipboard/urls.py
Normal file
@@ -0,0 +1,16 @@
|
||||
from __future__ import unicode_literals
|
||||
from django.conf.urls import url, include
|
||||
|
||||
from . import views
|
||||
|
||||
module_urls = [
|
||||
# Ipboard service control
|
||||
url(r'^activate/$', views.activate_ipboard_forum, name='auth_activate_ipboard'),
|
||||
url(r'^deactivate/$', views.deactivate_ipboard_forum, name='auth_deactivate_ipboard'),
|
||||
url(r'^reset_password/$', views.reset_ipboard_password, name='auth_reset_ipboard_password'),
|
||||
url(r'^set_password/$', views.set_ipboard_password, name='auth_set_ipboard_password'),
|
||||
]
|
||||
|
||||
urlpatterns = [
|
||||
url(r'^ipboard/', include(module_urls))
|
||||
]
|
||||
110
services/modules/ipboard/views.py
Normal file
110
services/modules/ipboard/views.py
Normal file
@@ -0,0 +1,110 @@
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from django.contrib import messages
|
||||
from django.contrib.auth.decorators import login_required
|
||||
from django.shortcuts import render, redirect
|
||||
|
||||
from authentication.decorators import members_and_blues
|
||||
from services.forms import ServicePasswordForm
|
||||
from eveonline.managers import EveManager
|
||||
|
||||
from .manager import IPBoardManager
|
||||
from .tasks import IpboardTasks
|
||||
from .models import IpboardUser
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@login_required
|
||||
@members_and_blues()
|
||||
def activate_ipboard_forum(request):
|
||||
logger.debug("activate_ipboard_forum called by user %s" % request.user)
|
||||
character = EveManager.get_main_character(request.user)
|
||||
logger.debug("Adding ipboard user for user %s with main character %s" % (request.user, character))
|
||||
result = IPBoardManager.add_user(character.character_name, request.user.email)
|
||||
if result[0] != "":
|
||||
ipboard_user = IpboardUser()
|
||||
ipboard_user.user = request.user
|
||||
ipboard_user.username = result[0]
|
||||
ipboard_user.save()
|
||||
logger.debug("Updated authserviceinfo for user %s with ipboard credentials. Updating groups." % request.user)
|
||||
IpboardTasks.update_groups.delay(request.user.pk)
|
||||
logger.info("Successfully activated ipboard for user %s" % request.user)
|
||||
messages.success(request, 'Activated IPBoard account.')
|
||||
credentials = {
|
||||
'username': result[0],
|
||||
'password': result[1],
|
||||
}
|
||||
return render(request, 'registered/service_credentials.html',
|
||||
context={'credentials': credentials, 'service': 'IPBoard'})
|
||||
else:
|
||||
logger.error("Unsuccessful attempt to activate ipboard for user %s" % request.user)
|
||||
messages.error(request, 'An error occurred while processing your IPBoard account.')
|
||||
return redirect("auth_services")
|
||||
|
||||
|
||||
@login_required
|
||||
@members_and_blues()
|
||||
def deactivate_ipboard_forum(request):
|
||||
logger.debug("deactivate_ipboard_forum called by user %s" % request.user)
|
||||
# false we failed
|
||||
if IpboardTasks.delete_user(request.user):
|
||||
logger.info("Successfully deactivated ipboard for user %s" % request.user)
|
||||
messages.success(request, 'Deactivated IPBoard account.')
|
||||
else:
|
||||
logger.error("Unsuccessful attempt to deactviate ipboard for user %s" % request.user)
|
||||
messages.error(request, 'An error occurred while processing your IPBoard account.')
|
||||
return redirect("auth_services")
|
||||
|
||||
|
||||
@login_required
|
||||
@members_and_blues()
|
||||
def set_ipboard_password(request):
|
||||
logger.debug("set_ipboard_password called by user %s" % request.user)
|
||||
error = None
|
||||
if request.method == 'POST':
|
||||
logger.debug("Received POST request with form.")
|
||||
form = ServicePasswordForm(request.POST)
|
||||
logger.debug("Form is valid: %s" % form.is_valid())
|
||||
if form.is_valid() and IpboardTasks.has_account(request.user):
|
||||
password = form.cleaned_data['password']
|
||||
logger.debug("Form contains password of length %s" % len(password))
|
||||
result = IPBoardManager.update_user_password(request.user.ipboard.username, request.user.email,
|
||||
plain_password=password)
|
||||
if result != "":
|
||||
logger.info("Successfully set IPBoard password for user %s" % request.user)
|
||||
messages.success(request, 'Set IPBoard password.')
|
||||
else:
|
||||
logger.error("Failed to install custom ipboard password for user %s" % request.user)
|
||||
messages.error(request, 'An error occurred while processing your IPBoard account.')
|
||||
return redirect("auth_services")
|
||||
else:
|
||||
logger.debug("Request is not type POST - providing empty form.")
|
||||
form = ServicePasswordForm()
|
||||
|
||||
logger.debug("Rendering form for user %s" % request.user)
|
||||
context = {'form': form, 'service': 'IPBoard', 'error': error}
|
||||
return render(request, 'registered/service_password.html', context=context)
|
||||
|
||||
|
||||
@login_required
|
||||
@members_and_blues()
|
||||
def reset_ipboard_password(request):
|
||||
logger.debug("reset_ipboard_password called by user %s" % request.user)
|
||||
if IpboardTasks.has_account(request.user):
|
||||
result = IPBoardManager.update_user_password(request.user.ipboard.username, request.user.email)
|
||||
if result != "":
|
||||
logger.info("Successfully reset ipboard password for user %s" % request.user)
|
||||
messages.success(request, 'Reset IPBoard password.')
|
||||
credentials = {
|
||||
'username': request.user.ipboard.username,
|
||||
'password': result,
|
||||
}
|
||||
return render(request, 'registered/service_credentials.html',
|
||||
context={'credentials': credentials, 'service': 'IPBoard'})
|
||||
|
||||
logger.error("Unsuccessful attempt to reset ipboard password for user %s" % request.user)
|
||||
messages.error(request, 'An error occurred while processing your IPBoard account.')
|
||||
return redirect("auth_services")
|
||||
Reference in New Issue
Block a user