pyceo/ceod/transactions/members/AddMemberTransaction.py

112 lines
3.5 KiB
Python

import traceback
from typing import Union, List
from zope import component
from ..AbstractTransaction import AbstractTransaction
from ceo_common.interfaces import IConfig, IMailService
from ceo_common.logger_factory import logger_factory
from ceod.model import User, Group
import ceod.utils as utils
logger = logger_factory(__name__)
class AddMemberTransaction(AbstractTransaction):
"""Transaction to add a new club member."""
operations = [
'add_user_to_ldap',
'add_group_to_ldap',
'add_user_to_kerberos',
'create_home_dir',
'set_forwarding_addresses',
'subscribe_to_mailing_list',
'send_welcome_message',
]
def __init__(
self,
uid: str,
cn: str,
program: Union[str, None],
terms: Union[List[str], None] = None,
non_member_terms: Union[List[str], None] = None,
forwarding_addresses: Union[List[str], None] = None,
):
super().__init__()
cfg = component.getUtility(IConfig)
self.uid = uid
self.cn = cn
self.program = program
self.terms = terms
self.non_member_terms = non_member_terms
self.forwarding_addresses = forwarding_addresses
self.user = None
self.group = None
self.new_member_list = cfg.get('mailman3_new_member_list')
self.mail_srv = component.getUtility(IMailService)
def child_execute_iter(self):
user = User(
uid=self.uid,
cn=self.cn,
program=self.program,
terms=self.terms,
non_member_terms=self.non_member_terms,
)
self.user = user
user.add_to_ldap()
yield 'add_user_to_ldap'
group = Group(
cn=user.uid,
gid_number=user.gid_number,
)
self.group = group
group.add_to_ldap()
yield 'add_group_to_ldap'
password = utils.gen_password()
user.add_to_kerberos(password)
yield 'add_user_to_kerberos'
user.create_home_dir()
yield 'create_home_dir'
if self.forwarding_addresses:
user.set_forwarding_addresses(self.forwarding_addresses)
yield 'set_forwarding_addresses'
# The following operations can't/shouldn't be rolled back because the
# user has already seen the email
try:
self.mail_srv.send_welcome_message_to(user)
yield 'send_welcome_message'
except Exception as err:
logger.warning('send_welcome_message failed:\n' + traceback.format_exc())
yield 'failed_to_send_welcome_message\n' + str(err)
try:
user.subscribe_to_mailing_list(self.new_member_list)
yield 'subscribe_to_mailing_list'
except Exception as err:
logger.warning('subscribe_to_mailing_list failed:\n' + traceback.format_exc())
yield 'failed_to_subscribe_to_mailing_list\n' + str(err)
user_json = user.to_dict(True)
# insert the password into the JSON so that the client can see it
user_json['password'] = password
self.finish(user_json)
def rollback(self):
if 'create_home_dir' in self.finished_operations:
self.user.delete_home_dir()
if 'add_user_to_kerberos' in self.finished_operations:
self.user.remove_from_kerberos()
if 'add_group_to_ldap' in self.finished_operations:
self.group.remove_from_ldap()
if 'add_user_to_ldap' in self.finished_operations:
self.user.remove_from_ldap()