Mathis/allianceauth
1
0
Fork 0
mirror of https://gitlab.com/allianceauth/allianceauth.git synced 2025-07-13 22:40:16 +02:00
Code Issues Packages Projects Releases Wiki Activity

Use integer teamspeak group IDs when filtering.

Browse Source
This commit is contained in:
Adarnof 2021-12-15 23:54:53 -05:00
parent 72740b9e4d
commit 6688f73565
2 changed files with 31 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
allianceauth/services/modules/teamspeak3/manager.py
View File

@ -159,7 +159,7 @@ class Teamspeak3Manager:
def _sync_ts_group_db(self): def _sync_ts_group_db(self):
try: try:
remote_groups = self._group_list() remote_groups = self._group_list()
managed_groups = {g:remote_groups[g] for g in remote_groups if g in set(remote_groups.keys()) - set(ReservedGroupName.objects.values_list('name', flat=True))} managed_groups = {g:int(remote_groups[g]) for g in remote_groups if g in set(remote_groups.keys()) - set(ReservedGroupName.objects.values_list('name', flat=True))}
remove = TSgroup.objects.exclude(ts_group_id__in=managed_groups.values()) remove = TSgroup.objects.exclude(ts_group_id__in=managed_groups.values())
if remove: if remove:
@ -174,8 +174,8 @@ class Teamspeak3Manager:
except TeamspeakError as e: except TeamspeakError as e:
logger.error(f"Error occurred while syncing TS group db: {str(e)}") logger.error(f"Error occurred while syncing TS group db: {str(e)}")
except: except Exception:
logger.exception("An unhandled exception has occurred while syncing TS groups.") logger.exception(f"An unhandled exception has occurred while syncing TS groups.")
def add_user(self, user, fmt_name): def add_user(self, user, fmt_name):
username_clean = self.__santatize_username(fmt_name[:30]) username_clean = self.__santatize_username(fmt_name[:30])
@ -234,7 +234,7 @@ class Teamspeak3Manager:
logger.exception(f"Failed to delete user id {uid} from TS3 - received response {ret}") logger.exception(f"Failed to delete user id {uid} from TS3 - received response {ret}")
return False return False
else: else:
logger.warn("User with id %s not found on TS3 server. Assuming succesful deletion." % uid) logger.warning("User with id %s not found on TS3 server. Assuming succesful deletion." % uid)
return True return True
def check_user_exists(self, uid): def check_user_exists(self, uid):

37
allianceauth/services/modules/teamspeak3/tests.py
View File

@ -366,7 +366,7 @@ class Teamspeak3ManagerTestCase(TestCase):
def test_update_groups_remove(self, remove, add, groups, userid): def test_update_groups_remove(self, remove, add, groups, userid):
"""Remove from one group""" """Remove from one group"""
userid.return_value = 1 userid.return_value = 1
groups.return_value = {'test': 1, 'dummy': 2} groups.return_value = {'test': '1', 'dummy': '2'}
Teamspeak3Manager().update_groups(1, {'test': 1}) Teamspeak3Manager().update_groups(1, {'test': 1})
self.assertEqual(add.call_count, 0) self.assertEqual(add.call_count, 0)
@ -390,7 +390,7 @@ class Teamspeak3ManagerTestCase(TestCase):
@mock.patch.object(Teamspeak3Manager, '_group_list') @mock.patch.object(Teamspeak3Manager, '_group_list')
def test_sync_group_db_create(self, group_list): def test_sync_group_db_create(self, group_list):
"""Populate the list of all TSgroups""" """Populate the list of all TSgroups"""
group_list.return_value = {'allowed':1, 'also allowed': 2} group_list.return_value = {'allowed':'1', 'also allowed':'2'}
Teamspeak3Manager()._sync_ts_group_db() Teamspeak3Manager()._sync_ts_group_db()
self.assertEqual(TSgroup.objects.all().count(), 2) self.assertEqual(TSgroup.objects.all().count(), 2)
@ -398,15 +398,15 @@ class Teamspeak3ManagerTestCase(TestCase):
def test_sync_group_db_delete(self, group_list): def test_sync_group_db_delete(self, group_list):
"""Populate the list of all TSgroups, and delete one which no longer exists""" """Populate the list of all TSgroups, and delete one which no longer exists"""
TSgroup.objects.create(ts_group_name='deleted', ts_group_id=3) TSgroup.objects.create(ts_group_name='deleted', ts_group_id=3)
group_list.return_value = {'allowed': 1, 'also allowed': 2} group_list.return_value = {'allowed': '1'}
Teamspeak3Manager()._sync_ts_group_db() Teamspeak3Manager()._sync_ts_group_db()
self.assertEqual(TSgroup.objects.all().count(), 2) self.assertEqual(TSgroup.objects.all().count(), 1)
self.assertFalse(TSgroup.objects.filter(ts_group_name='deleted').exists()) self.assertFalse(TSgroup.objects.filter(ts_group_name='deleted').exists())
@mock.patch.object(Teamspeak3Manager, '_group_list') @mock.patch.object(Teamspeak3Manager, '_group_list')
def test_sync_group_db_dont_create_reserved(self, group_list): def test_sync_group_db_dont_create_reserved(self, group_list):
"""Populate the list of all TSgroups, ignoring a reserved group name""" """Populate the list of all TSgroups, ignoring a reserved group name"""
group_list.return_value = {'allowed': 1, 'reserved': 4} group_list.return_value = {'allowed': '1', 'reserved': '4'}
Teamspeak3Manager()._sync_ts_group_db() Teamspeak3Manager()._sync_ts_group_db()
self.assertEqual(TSgroup.objects.all().count(), 1) self.assertEqual(TSgroup.objects.all().count(), 1)
self.assertFalse(TSgroup.objects.filter(ts_group_name='reserved').exists()) self.assertFalse(TSgroup.objects.filter(ts_group_name='reserved').exists())
@ -415,11 +415,28 @@ class Teamspeak3ManagerTestCase(TestCase):
def test_sync_group_db_delete_reserved(self, group_list): def test_sync_group_db_delete_reserved(self, group_list):
"""Populate the list of all TSgroups, deleting the TSgroup model for one which has become reserved""" """Populate the list of all TSgroups, deleting the TSgroup model for one which has become reserved"""
TSgroup.objects.create(ts_group_name='reserved', ts_group_id=4) TSgroup.objects.create(ts_group_name='reserved', ts_group_id=4)
group_list.return_value = {'allowed': 1, 'reserved': 4} group_list.return_value = {'allowed': '1', 'reserved': '4'}
Teamspeak3Manager()._sync_ts_group_db() Teamspeak3Manager()._sync_ts_group_db()
self.assertEqual(TSgroup.objects.all().count(), 1) self.assertEqual(TSgroup.objects.all().count(), 1)
self.assertFalse(TSgroup.objects.filter(ts_group_name='reserved').exists()) self.assertFalse(TSgroup.objects.filter(ts_group_name='reserved').exists())
@mock.patch.object(Teamspeak3Manager, '_group_list')
def test_sync_group_db_partial_addition(self, group_list):
"""Some TSgroups already exist in database, add new ones"""
TSgroup.objects.create(ts_group_name='allowed', ts_group_id=1)
group_list.return_value = {'allowed': '1', 'also allowed': '2'}
Teamspeak3Manager()._sync_ts_group_db()
self.assertEqual(TSgroup.objects.all().count(), 2)
@mock.patch.object(Teamspeak3Manager, '_group_list')
def test_sync_group_db_partial_removal(self, group_list):
"""One TSgroup has been deleted on server, so remove its model"""
TSgroup.objects.create(ts_group_name='allowed', ts_group_id=1)
TSgroup.objects.create(ts_group_name='also allowed', ts_group_id=2)
group_list.return_value = {'allowed': '1'}
Teamspeak3Manager()._sync_ts_group_db()
self.assertEqual(TSgroup.objects.all().count(), 1)
class MockRequest: class MockRequest:
pass pass
@ -445,12 +462,12 @@ class Teamspeak3AdminTestCase(TestCase):
def test_field_queryset_no_reserved_names(self): def test_field_queryset_no_reserved_names(self):
"""Ensure all groups are listed when no reserved names""" """Ensure all groups are listed when no reserved names"""
form = self.admin.get_form(request) form = self.admin.get_form(request)
self.assertEqual(form.base_fields['auth_group']._get_queryset().count(), 1) self.assertQuerysetEqual(form.base_fields['auth_group']._get_queryset(), Group.objects.all())
self.assertEqual(form.base_fields['ts_group']._get_queryset().count(), 1) self.assertQuerysetEqual(form.base_fields['ts_group']._get_queryset(), TSgroup.objects.all())
def test_field_queryset_reserved_names(self): def test_field_queryset_reserved_names(self):
"""Ensure reserved group names are filtered out""" """Ensure reserved group names are filtered out"""
ReservedGroupName.objects.bulk_create([ReservedGroupName(name='test', reason='tests', created_by='Bob')]) ReservedGroupName.objects.bulk_create([ReservedGroupName(name='test', reason='tests', created_by='Bob')])
form = self.admin.get_form(request) form = self.admin.get_form(request)
self.assertEqual(form.base_fields['auth_group']._get_queryset().count(), 0) self.assertQuerysetEqual(form.base_fields['auth_group']._get_queryset(), Group.objects.none())
self.assertEqual(form.base_fields['ts_group']._get_queryset().count(), 0) self.assertQuerysetEqual(form.base_fields['ts_group']._get_queryset(), TSgroup.objects.none())