Finish inner workings, start tests work

parent 7fc54a3d
This diff is collapsed.
......@@ -3,7 +3,7 @@ import logging
from django.core.management.base import BaseCommand
from mailinglists import gsuite
from mailinglists.gsuite import GSuiteSyncService
logger = logging.getLogger(__name__)
......@@ -12,4 +12,5 @@ class Command(BaseCommand):
def handle(self, *args, **options):
"""Sync all mailing lists"""
gsuite.sync_mailinglists()
sync_service = GSuiteSyncService()
sync_service.sync_mailinglists()
......@@ -23,6 +23,14 @@ class Migration(migrations.Migration):
model_name='mailinglist',
name='autoresponse_text',
),
migrations.RemoveField(
model_name='mailinglist',
name='prefix',
),
migrations.RemoveField(
model_name='mailinglist',
name='archived',
),
migrations.AlterField(
model_name='listalias',
name='mailinglist',
......
......@@ -41,25 +41,11 @@ class MailingList(models.Model):
help_text=_('Enter the name for the list (i.e. name@thalia.nu).'),
)
prefix = models.CharField(
verbose_name=_("Prefix"),
blank=True,
max_length=200,
help_text=_('Enter a prefix that should be prefixed to subjects '
'of all emails sent via this mailinglist.'),
)
description = models.TextField(
verbose_name=_("Description"),
help_text=_('Write a description for the mailinglist.'),
)
archived = models.BooleanField(
verbose_name=_("Archived"),
default=True,
help_text=_('Indicate whether an archive should be kept.')
)
moderated = models.BooleanField(
verbose_name=_("Moderated"),
default=False,
......
from django.db.models.signals import pre_save, \
pre_delete
from django.db.models.signals import pre_save
from django.dispatch import receiver
from googleapiclient.errors import HttpError
from mailinglists.gsuite import (mailinglist_to_group, sync_mailinglists,
update_group, create_group)
from mailinglists.gsuite import GSuiteSyncService
from mailinglists.models import MailingList
@receiver(pre_save, sender='mailinglists.MailingList')
def pre_mailinglist_save(instance, **kwargs):
group = mailinglist_to_group(instance)
sync_service = GSuiteSyncService()
group = sync_service.mailinglist_to_group(instance)
old_list = MailingList.objects.filter(pk=instance.pk).first()
try:
if old_list is None:
create_group(group)
sync_service.create_group(group)
else:
update_group(old_list.name, group)
sync_service.update_group(old_list.name, group)
except HttpError:
# Cannot do direct create or update, do full sync for list
sync_mailinglists([group])
@receiver(pre_delete, sender='mailinglists.MailingList')
def pre_mailinglist_delete(instance, **kwargs):
sync_mailinglists([mailinglist_to_group(instance)])
sync_service.sync_mailinglists([group])
"""Test for the GSuite sync in the mailinglists package"""
from unittest import mock
from unittest.mock import MagicMock
import factory
from django.db.models import signals
from django.test import TestCase
from googleapiclient.errors import HttpError
from googleapiclient.http import HttpMockSequence
from httplib2 import Response
from mailinglists.gsuite import GSuiteSyncService
from mailinglists.models import MailingList, ListAlias, VerbatimAddress
class CatchRequestHttpMockSequence(HttpMockSequence):
requests = []
class Request:
def __init__(self, uri, method, body):
super().__init__()
self.uri = uri
self.method = method
self.body = body
def __eq__(self, other: object) -> bool:
if isinstance(other, self.__class__):
return self.__dict__ == other.__dict__
return False
def request(self, uri, method='GET', body=None, headers=None,
redirections=1, connection_type=None):
self.requests.append(self.Request(uri, method, body))
return super().request(uri, method, body, headers, redirections,
connection_type)
# http = CatchRequestHttpMockSequence([
# ({'status': '200'}, 'data'),
# ({'status': '200'}, 'data')
# ])
class GSuiteSyncTestCase(TestCase):
@classmethod
@factory.django.mute_signals(signals.pre_save)
def setUpTestData(cls):
cls.settings_api = MagicMock()
cls.directory_api = MagicMock()
cls.sync_service = GSuiteSyncService(
cls.settings_api, cls.directory_api)
cls.mailinglist = MailingList.objects.create(
name='new_group',
description='some description',
moderated=False
)
ListAlias.objects.create(
mailinglist=cls.mailinglist, alias='alias2')
VerbatimAddress.objects.create(
mailinglist=cls.mailinglist, address='test2@thalia.localhost')
def setUp(self):
self.settings_api.reset_mock()
self.directory_api.reset_mock()
def test_default_lists(self):
self.assertEqual(len(self.sync_service._get_default_lists()),
18)
def test_automatic_to_group(self):
group = GSuiteSyncService._automatic_to_group({
'moderated': False,
'name': 'new_group',
'description': 'some description',
'aliases': ['alias1'],
'addresses': ['test1@thalia.localhost']
})
self.assertEqual(group, GSuiteSyncService.GroupData(
'new_group', 'some description', False,
['alias1'], ['test1@thalia.localhost']
))
def test_mailinglist_to_group(self):
group = GSuiteSyncService.mailinglist_to_group(self.mailinglist)
self.assertEqual(group, GSuiteSyncService.GroupData(
'new_group', 'some description', False,
['alias2'], ['test2@thalia.localhost']
))
def test_group_settings(self):
self.assertEqual(self.sync_service._group_settings(False), {
'allowExternalMembers': 'true',
'allowWebPosting': 'false',
'archiveOnly': 'false',
'isArchived': 'true',
'membersCanPostAsTheGroup': 'false',
'messageModerationLevel': 'MODERATE_NONE',
'replyTo': 'REPLY_TO_SENDER',
'whoCanAssistContent': 'NONE',
'whoCanContactOwner': 'ALL_MANAGERS_CAN_CONTACT',
'whoCanDiscoverGroup': 'ALL_MEMBERS_CAN_DISCOVER',
'whoCanJoin': 'INVITED_CAN_JOIN',
'whoCanLeaveGroup': 'NONE_CAN_LEAVE',
'whoCanModerateContent': 'OWNERS_AND_MANAGERS',
'whoCanModerateMembers': 'NONE',
'whoCanPostMessage': 'ANYONE_CAN_POST',
'whoCanViewGroup': 'ALL_MANAGERS_CAN_VIEW',
'whoCanViewMembership': 'ALL_MANAGERS_CAN_VIEW'
})
self.assertEqual(self.sync_service._group_settings(True), {
'allowExternalMembers': 'true',
'allowWebPosting': 'false',
'archiveOnly': 'false',
'isArchived': 'true',
'membersCanPostAsTheGroup': 'false',
'messageModerationLevel': 'MODERATE_ALL_MESSAGES',
'replyTo': 'REPLY_TO_SENDER',
'whoCanAssistContent': 'NONE',
'whoCanContactOwner': 'ALL_MANAGERS_CAN_CONTACT',
'whoCanDiscoverGroup': 'ALL_MEMBERS_CAN_DISCOVER',
'whoCanJoin': 'INVITED_CAN_JOIN',
'whoCanLeaveGroup': 'NONE_CAN_LEAVE',
'whoCanModerateContent': 'OWNERS_AND_MANAGERS',
'whoCanModerateMembers': 'NONE',
'whoCanPostMessage': 'ANYONE_CAN_POST',
'whoCanViewGroup': 'ALL_MANAGERS_CAN_VIEW',
'whoCanViewMembership': 'ALL_MANAGERS_CAN_VIEW'
})
@mock.patch('mailinglists.gsuite.logger')
def test_create_group(self, logger_mock):
with self.subTest('Successful'):
self.sync_service.create_group(GSuiteSyncService.GroupData(
'new_group', 'some description', False,
['alias2'], ['test2@thalia.localhost']
))
self.directory_api.groups().insert.assert_called_once_with(body={
'email': 'new_group@thalia.localhost',
'name': 'new_group',
'description': 'some description',
})
self.settings_api.groups().update.assert_called_once_with(
groupUniqueId='new_group@thalia.localhost',
body=self.sync_service._group_settings(False)
)
self.directory_api.members().list.assert_called()
self.directory_api.groups().aliases().list.assert_called()
self.settings_api.reset_mock()
self.directory_api.reset_mock()
with self.subTest('Failure'):
self.directory_api.groups().insert(body={
'email': 'new_group@thalia.localhost',
'name': 'new_group',
'description': 'some description',
}).execute.side_effect = HttpError(Response({'status': 500}),
bytes())
self.sync_service.create_group(GSuiteSyncService.GroupData(
'new_group', 'some description', False,
['alias2'], ['test2@thalia.localhost']
))
self.directory_api.members().list.assert_not_called()
self.directory_api.groups().aliases().list.assert_not_called()
logger_mock.error.assert_called_once_with(
'Could not successfully finish '
'creating the list new_group', bytes()
)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment