Commit 425639d5 authored by Gregory Eric Sanderson's avatar Gregory Eric Sanderson Committed by Cedric Abunar

delete user_line_extension service

parent ba0fc655
......@@ -26,7 +26,6 @@ setup(
'xivo_dao.data_handler.line_extension',
'xivo_dao.data_handler.user',
'xivo_dao.data_handler.user_line',
'xivo_dao.data_handler.user_line_extension',
'xivo_dao.data_handler.user_voicemail',
'xivo_dao.data_handler.voicemail',
'xivo_dao.data_handler.cti_profile',
......
# -*- coding: utf-8 -*-
# Copyright (C) 2013-2014 Avencall
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>
from xivo_dao.alchemy.user_line import UserLine as ULESchema
from xivo_dao.helpers.db_manager import daosession
from xivo_dao.data_handler.exception import ElementNotExistsError, \
ElementCreationError, ElementDeletionError, ElementEditionError
from sqlalchemy.exc import SQLAlchemyError
from model import UserLineExtension
from xivo_dao.data_handler.user import dao as user_dao
@daosession
def get(session, ule_id):
res = _new_query(session).filter(ULESchema.id == ule_id).first()
if not res:
raise ElementNotExistsError('UserLineExtension', id=ule_id)
return UserLineExtension.from_data_source(res)
@daosession
def find_all(session):
res = session.query(ULESchema).all()
if not res:
return []
tmp = []
for ule in res:
tmp.append(UserLineExtension.from_data_source(ule))
return tmp
@daosession
def find_all_by_user_id(session, user_id):
ules = session.query(ULESchema).filter(ULESchema.user_id == user_id).all()
return [UserLineExtension.from_data_source(ule) for ule in ules]
@daosession
def find_all_by_extension_id(session, extension_id):
ules = session.query(ULESchema).filter(ULESchema.extension_id == extension_id).all()
return [UserLineExtension.from_data_source(ule) for ule in ules]
@daosession
def find_all_by_line_id(session, line_id):
ules = session.query(ULESchema).filter(ULESchema.line_id == line_id).all()
return [UserLineExtension.from_data_source(ule) for ule in ules]
@daosession
def find_main_user(session, ule):
row = (session.query(ULESchema.user_id)
.filter(ULESchema.main_user == True)
.filter(ULESchema.line_id == ule.line_id)
.first())
if not row:
return user_dao.get(ule.user_id)
return user_dao.get(row[0])
@daosession
def create(session, user_line_extension):
user_line_extension_row = user_line_extension.to_data_source(ULESchema)
session.begin()
session.add(user_line_extension_row)
try:
session.commit()
except SQLAlchemyError as e:
session.rollback()
raise ElementCreationError('UserLineExtension', e)
user_line_extension.id = user_line_extension_row.id
return user_line_extension
@daosession
def edit(session, user_line_extension):
session.begin()
nb_row_affected = (session.query(ULESchema)
.filter(ULESchema.id == user_line_extension.id)
.update(user_line_extension.to_data_dict()))
try:
session.commit()
except SQLAlchemyError as e:
session.rollback()
raise ElementEditionError('UserLineExtension', e)
if nb_row_affected == 0:
raise ElementEditionError('UserLineExtension', 'user_line_extension_id %s not exsit' % user_line_extension.id)
return nb_row_affected
@daosession
def delete(session, user_line_extension):
session.begin()
try:
nb_row_affected = session.query(ULESchema).filter(ULESchema.id == user_line_extension.id).delete()
session.commit()
except SQLAlchemyError, e:
session.rollback()
raise ElementDeletionError('UserLineExtension', e)
if nb_row_affected == 0:
raise ElementDeletionError('UserLineExtension', 'user_line_extension_id %s not exsit' % user_line_extension.id)
return nb_row_affected
@daosession
def already_linked(session, user_id, line_id):
count = (session.query(ULESchema)
.filter(ULESchema.user_id == user_id)
.filter(ULESchema.line_id == line_id)
.count())
return count > 0
@daosession
def main_user_is_allowed_to_delete(session, main_line_id):
count = (session.query(ULESchema)
.filter(ULESchema.line_id == main_line_id)
.count())
return count == 1
def _new_query(session):
return session.query(ULESchema)
# -*- coding: utf-8 -*-
# Copyright (C) 2013-2014 Avencall
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>
from xivo_dao.helpers.abstract_model import AbstractModels
class UserLineExtension(AbstractModels):
MANDATORY = [
'user_id',
'line_id',
'extension_id',
]
# mapping = {db_field: model_field}
_MAPPING = {
'id': 'id',
'user_id': 'user_id',
'line_id': 'line_id',
'extension_id': 'extension_id',
'main_user': 'main_user',
'main_line': 'main_line'
}
_RELATION = {}
def __init__(self, *args, **kwargs):
AbstractModels.__init__(self, *args, **kwargs)
# -*- coding: utf-8 -*-
# Copyright (C) 2013-2014 Avencall
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>
from xivo_bus.resources.user_line_extension import event
from xivo_dao import helpers
from xivo_dao.helpers import bus_manager
def created(user_line_extension):
data = _build_edit_user_phone(user_line_extension)
helpers.sysconfd_connector.exec_request_handlers(data)
bus_manager.send_bus_command(event.CreateUserLineExtensionEvent(user_line_extension.id,
user_line_extension.user_id,
user_line_extension.line_id,
user_line_extension.extension_id,
user_line_extension.main_user,
user_line_extension.main_line))
def edited(user_line_extension):
data = _build_edit_user_phone(user_line_extension)
helpers.sysconfd_connector.exec_request_handlers(data)
bus_manager.send_bus_command(event.EditUserLineExtensionEvent(user_line_extension.id,
user_line_extension.user_id,
user_line_extension.line_id,
user_line_extension.extension_id,
user_line_extension.main_user,
user_line_extension.main_line))
def deleted(user_line_extension):
data = _build_edit_user_phone(user_line_extension)
helpers.sysconfd_connector.exec_request_handlers(data)
bus_manager.send_bus_command(event.DeleteUserLineExtensionEvent(user_line_extension.id,
user_line_extension.user_id,
user_line_extension.line_id,
user_line_extension.extension_id,
user_line_extension.main_user,
user_line_extension.main_line))
def _build_edit_user_phone(user_line_extension):
return _new_sysconfd_data([
'xivo[user,edit,%s]' % user_line_extension.user_id,
'xivo[phone,edit,%s]' % user_line_extension.line_id
])
def _new_sysconfd_data(ctibus_commands):
return {
'ctibus': ctibus_commands,
'dird': [],
'ipbx': ['dialplan reload', 'sip reload'],
'agentbus': []
}
# -*- coding: utf-8 -*-
# Copyright (C) 2013-2014 Avencall
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>
import dao
import notifier
import validator
from xivo_dao.data_handler.exception import ElementNotExistsError
from xivo_dao.data_handler.line import services as line_services
from xivo_dao.data_handler.line import dao as line_dao
from xivo_dao.data_handler.user import services as user_services
from xivo_dao.data_handler.extension import services as extension_services
from xivo_dao.data_handler.extension import dao as extension_dao
from xivo import caller_id
def get(ule_id):
return dao.get(ule_id)
def find_all():
return dao.find_all()
def find_all_by_user_id(user_id):
return dao.find_all_by_user_id(user_id)
def find_all_by_extension_id(extension_id):
return dao.find_all_by_extension_id(extension_id)
def find_all_by_line_id(line_id):
return dao.find_all_by_line_id(line_id)
def create(ule):
user, line, extension = validator.validate_create(ule)
main_user = dao.find_main_user(ule)
_adjust_optional_parameters(ule, main_user)
ule = dao.create(ule)
_make_secondary_associations(main_user, line, extension)
notifier.created(ule)
return ule
def edit(ule):
validator.validate(ule)
dao.edit(ule)
notifier.edited(ule)
def delete(ule):
validator.validate(ule)
validator.is_allowed_to_delete(ule)
dao.delete(ule)
notifier.deleted(ule)
def delete_everything(ule):
user, line, extension = validator.validate(ule)
dao.delete(ule)
_remove_user(user)
_remove_line(line)
_remove_exten(extension)
notifier.deleted(ule)
def _adjust_optional_parameters(ule, main_user):
ule.main_line = True
ule.main_user = (ule.user_id == main_user.id)
def _make_secondary_associations(main_user, line, extension):
_associate_extension(main_user, extension)
_associate_line(line, extension, main_user)
def _associate_extension(main_user, extension):
extension_dao.associate_to_user(main_user, extension)
def _associate_line(line, extension, main_user):
line.number = extension.exten
line.context = extension.context
callerid, _, _ = caller_id.build_caller_id('', main_user.fullname, extension.exten)
line.callerid = callerid
line_dao.edit(line)
line_dao.update_xivo_userid(line, main_user)
def _remove_user(user):
try:
user_services.delete(user)
except ElementNotExistsError:
return
def _remove_line(line):
try:
line_services.delete(line)
except ElementNotExistsError:
return
def _remove_exten(extension):
try:
extension_services.delete(extension)
except ElementNotExistsError:
return
# -*- coding: utf-8 -*-
# Copyright (C) 2013-2014 Avencall
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>
from hamcrest import assert_that, contains, equal_to
from mock import patch, Mock
from .. import dao as ule_dao
from ..model import UserLineExtension
from xivo_dao.tests.test_dao import DAOTestCase
from sqlalchemy.exc import SQLAlchemyError
from xivo_dao.data_handler.exception import ElementCreationError, \
ElementDeletionError, ElementEditionError
from xivo_dao.alchemy.userfeatures import UserFeatures as UserSchema
from xivo_dao.alchemy.extension import Extension as ExtensionSchema
from xivo_dao.alchemy.linefeatures import LineFeatures as LineSchema
from xivo_dao.alchemy.user_line import UserLine as ULESchema
from xivo_dao.alchemy.ctiphonehintsgroup import CtiPhoneHintsGroup
from xivo_dao.alchemy.ctipresences import CtiPresences
from xivo_dao.alchemy.cti_profile import CtiProfile
class TestUserLineExtensionDao(DAOTestCase):
tables = [
UserSchema,
ExtensionSchema,
LineSchema,
ULESchema,
CtiPhoneHintsGroup,
CtiPresences,
CtiProfile,
]
def setUp(self):
self.empty_tables()
def test_get_no_exist(self):
self.assertRaises(LookupError, ule_dao.get, 666)
def test_get(self):
expected_ule = self.add_user_line_with_exten()
ule = ule_dao.get(expected_ule.id)
assert_that(ule.id, equal_to(expected_ule.id))
assert_that(ule.user_id, equal_to(expected_ule.user_id))
assert_that(ule.line_id, equal_to(expected_ule.line_id))
assert_that(ule.extension_id, equal_to(expected_ule.extension_id))
assert_that(ule.main_user, equal_to(expected_ule.main_user))
assert_that(ule.main_line, equal_to(expected_ule.main_line))
def test_find_all_by_user_id_not_found(self):
expected_result = []
user_id = 5676
result = ule_dao.find_all_by_user_id(user_id)
assert_that(result, equal_to(expected_result))
def test_find_all_by_user_id_found(self):
ule_row = self.add_user_line_with_exten()
user_id = ule_row.user_id
expected_ule = UserLineExtension.from_data_source(ule_row)
result = ule_dao.find_all_by_user_id(user_id)
assert_that(result, contains(expected_ule))
def test_find_all_by_extension_id_not_found(self):
expected_result = []
extension_id = 5676
result = ule_dao.find_all_by_extension_id(extension_id)
assert_that(result, equal_to(expected_result))
def test_find_all_by_extension_id_found(self):
ule_row = self.add_user_line_with_exten()
extension_id = ule_row.extension_id
expected_ule = UserLineExtension.from_data_source(ule_row)
result = ule_dao.find_all_by_extension_id(extension_id)
assert_that(result, contains(expected_ule))
def test_find_all_by_line_id_not_found(self):
expected_result = []
line_id = 5676
result = ule_dao.find_all_by_line_id(line_id)
assert_that(result, equal_to(expected_result))
def test_find_all_by_line_id_found(self):
ule_row = self.add_user_line_with_exten()
line_id = ule_row.line_id
expected_ule = UserLineExtension.from_data_source(ule_row)
result = ule_dao.find_all_by_line_id(line_id)
assert_that(result, contains(expected_ule))
def test_create(self):
expected_user = self.add_user()
expected_line = self.add_line()
expected_extension = self.add_extension()
ule = UserLineExtension(user_id=expected_user.id,
line_id=expected_line.id,
extension_id=expected_extension.id,
main_user=True,
main_line=False)
created_extension = ule_dao.create(ule)
row = self.session.query(ULESchema).filter(ULESchema.id == created_extension.id).first()
assert_that(row.id, equal_to(created_extension.id))
assert_that(row.user_id, equal_to(expected_user.id))
assert_that(row.line_id, equal_to(expected_line.id))
assert_that(row.extension_id, equal_to(expected_extension.id))
def test_create_same_user_id__line_id(self):
expected_user = self.add_user()
expected_line = self.add_line()
expected_extension_1 = self.add_extension()
expected_extension_2 = self.add_extension(exten='1254')
ule = UserLineExtension(user_id=expected_user.id,
line_id=expected_line.id,
extension_id=expected_extension_1.id,
main_user=True,
main_line=False)
ule_dao.create(ule)
ule = UserLineExtension(user_id=expected_user.id,
line_id=expected_line.id,
extension_id=expected_extension_2.id,
main_user=True,
main_line=False)
self.assertRaises(ElementCreationError, ule_dao.create, ule)
@patch('xivo_dao.helpers.db_manager.AsteriskSession')
def test_create_extension_with_error_from_dao(self, Session):
session = Mock()
session.commit.side_effect = SQLAlchemyError()
Session.return_value = session
expected_user = self.add_user()
expected_line = self.add_line()
expected_extension = self.add_extension()
ule = UserLineExtension(user_id=expected_user.id,
line_id=expected_line.id,
extension_id=expected_extension.id,
main_user=True,
main_line=False)
self.assertRaises(ElementCreationError, ule_dao.create, ule)
session.begin.assert_called_once_with()
session.rollback.assert_called_once_with()
def test_edit(self):
expected_ule = self.add_user_line_with_exten()
expected_user = self.add_user()
ule = UserLineExtension(id=expected_ule.id, user_id=expected_user.id)
ule_dao.edit(ule)
row = (self.session.query(ULESchema)
.filter(ULESchema.user_id == expected_user.id)
.first())
self.assertEquals(row.user_id, expected_user.id)
def test_edit_with_unknown_id(self):
ule = UserLineExtension(id=123, user_id=55)
self.assertRaises(ElementEditionError, ule_dao.edit, ule)
@patch('xivo_dao.helpers.db_manager.AsteriskSession')
def test_edit_with_database_error(self, Session):
session = Mock()
session.commit.side_effect = SQLAlchemyError()
Session.return_value = session
expected_ule = self.add_user_line_with_exten()
ule = UserLineExtension(id=expected_ule.id, user_id=55)
self.assertRaises(ElementEditionError, ule_dao.edit, ule)
session.begin.assert_called_once_with()
session.rollback.assert_called_once_with()
def test_delete(self):
expected_ule = self.add_user_line_with_exten()
extension = ule_dao.get(expected_ule.id)
ule_dao.delete(extension)
row = self.session.query(ULESchema).filter(ULESchema.id == expected_ule.id).first()
self.assertEquals(row, None)
def test_delete_not_exist(self):
extension = UserLineExtension(id=1)
self.assertRaises(ElementDeletionError, ule_dao.delete, extension)
def test_already_linked_no_associations(self):
user_id = 1
line_id = 2
expected = False
result = ule_dao.already_linked(user_id, line_id)
assert_that(result, equal_to(expected))
def test_already_linked(self):
ule = self.add_user_line_with_exten()
expected = True
result = ule_dao.already_linked(ule.user_id, ule.line_id)
assert_that(result, equal_to(expected))
def test_already_linked_different_user(self):
ule = self.add_user_line_with_exten()
user = self.add_user()
expected = False
result = ule_dao.already_linked(user.id, ule.line_id)
assert_that(result, equal_to(expected))
def test_find_main_user_no_user(self):
user = self.add_user()
ule = UserLineExtension(user_id=user.id, line_id=2, extension_id=3)
result = ule_dao.find_main_user(ule)
assert_that(result.id, equal_to(user.id))
def test_find_main_user_one_user(self):
ule = self.add_user_line_with_exten()
result = ule_dao.find_main_user(ule)
assert_that(result.id, equal_to(ule.user_id))
def test_main_user_is_allowed_to_delete(self):
main_user = self.add_user()
line = self.add_line()
extension = self.add_extension()
main_ule = self.add_user_line(user_id=main_user.id,
line_id=line.id,
extension_id=extension.id,
main_user=True,
main_line=False)
result = ule_dao.main_user_is_allowed_to_delete(main_ule.line_id)
assert_that(result, equal_to(True))