Commit 7813fff7 authored by Cedric Abunar's avatar Cedric Abunar

[evo]add cti config

parent 3d43b210
......@@ -21,7 +21,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from sqlalchemy.schema import Column, ForeignKey, Sequence
from sqlalchemy.types import Integer, String, Boolean
from sqlalchemy.types import Integer, String
from xivo_dao.alchemy.base import Base
......@@ -31,6 +31,5 @@ class CtiProfile(Base):
id = Column(Integer, Sequence('cti_profile_id_seq'), primary_key=True)
name = Column(String(255), nullable=False)
presence_id = Column(Integer, ForeignKey("ctiphonehintsgroup.id"))
phonehints_id = Column(Integer, ForeignKey("ctipresences.id"))
deletable = Column(Boolean)
presence_id = Column(Integer, ForeignKey("ctipresences.id"))
phonehints_id = Column(Integer, ForeignKey("ctiphonehintsgroup.id"))
......@@ -31,5 +31,5 @@ class CtiPhoneHintsGroup(Base):
id = Column(Integer, Sequence('ctiphonehintsgroup_id_seq'), primary_key=True)
name = Column(String(255), nullable=False)
description = Column(String(255), nullable=False)
description = Column(String(255))
deletable = Column(Integer)
......@@ -31,5 +31,5 @@ class CtiPresences(Base):
id = Column(Integer, Sequence('ctipresences_id_seq'), primary_key=True)
name = Column(String(255), nullable=False)
description = Column(String(255), nullable=False)
deletable = Column(Integer)
description = Column(String(255))
deletable = Column(Integer, default=1)
# -*- coding: utf-8 -*-
#
# Copyright (C) 2012 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.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Avencall. See the LICENSE file at top of the souce tree
# or delivered in the installable package in which XiVO CTI Server is
# distributed for more details.
#
# 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 sqlalchemy.schema import Column, Sequence
from sqlalchemy.types import Integer, String
from xivo_dao.alchemy.base import Base
class CtiStatus(Base):
__tablename__ = 'ctistatus'
id = Column(Integer, Sequence('ctistatus_id_seq'), primary_key=True)
presence_id = Column(Integer)
name = Column(String(255), nullable=False)
display_name = Column(String(255), nullable=False)
actions = Column(String(255), nullable=False)
color = Column(String(20), nullable=False)
access_status = Column(String(255), nullable=False)
deletable = Column(Integer, default=1)
# -*- coding: utf-8 -*-
#
# Copyright (C) 2012 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.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Avencall. See the LICENSE file at top of the source tree
# or delivered in the installable package in which XiVO CTI Server is
# distributed for more details.
#
# 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 import dbconnection
from xivo_dao.alchemy.cti_preference import CtiPreference
from xivo_dao.alchemy.cti_profile import CtiProfile
from xivo_dao.alchemy.cti_profile_preference import CtiProfilePreference
_DB_NAME = 'asterisk'
def _session():
connection = dbconnection.get_connection(_DB_NAME)
return connection.get_session()
def _get(preference_id):
return _session().query(CtiPreference).filter(CtiPreference.id == preference_id).first()
def get_name(preference_id):
return _get(preference_id).name
def get_preferences():
res = {}
rows = (_session().query(CtiProfile).all())
for row in rows:
res[row.name] = {}
rows = (_session().query(CtiProfile, CtiProfilePreference, CtiPreference)
.join((CtiProfilePreference, CtiProfilePreference.profile_id == CtiProfile.id),
(CtiPreference, CtiProfilePreference.preference_id == CtiPreference.id))
.all())
for row in rows:
cti_profile, cti_profile_preference, cti_preference = row
profile = cti_profile.name
option_dict = {cti_preference.option: cti_profile_preference.value}
res[profile].update(option_dict)
return res
......@@ -22,6 +22,11 @@
from xivo_dao.alchemy import dbconnection
from xivo_dao.alchemy.cti_profile import CtiProfile
from xivo_dao.alchemy.ctipresences import CtiPresences
from xivo_dao.alchemy.ctiphonehintsgroup import CtiPhoneHintsGroup
from xivo_dao.alchemy.cti_profile_xlet import CtiProfileXlet
from xivo_dao.alchemy.cti_xlet import CtiXlet
from xivo_dao.alchemy.cti_xlet_layout import CtiXletLayout
_DB_NAME = 'asterisk'
......@@ -37,3 +42,39 @@ def _get(profile_id):
def get_name(profile_id):
return _get(profile_id).name
def get_profiles():
rows = (_session().query(CtiProfile, CtiPresences, CtiPhoneHintsGroup, CtiProfileXlet, CtiXlet, CtiXletLayout)
.join((CtiPresences, CtiProfile.presence_id == CtiPresences.id),
(CtiPhoneHintsGroup, CtiProfile.phonehints_id == CtiPhoneHintsGroup.id),
(CtiProfileXlet, CtiProfile.id == CtiProfileXlet.profile_id),
(CtiXlet, CtiProfileXlet.xlet_id == CtiXlet.id),
(CtiXletLayout, CtiProfileXlet.layout_id == CtiXletLayout.id))
.all())
res = {}
for row in rows:
cti_profiles, cti_presences, cti_phonehints_group, cti_profile_xlet, cti_xlet, cti_xlet_layout = row
if cti_profiles.name not in res:
res[cti_profiles.name] = {}
new_profile = res[cti_profiles.name]
new_profile['name'] = cti_profiles.name
new_profile['phonestatus'] = cti_phonehints_group.name
new_profile['userstatus'] = cti_presences.name
new_profile['preferences'] = 'itm_preferences_%s' % cti_profiles.name
new_profile['services'] = 'itm_services_%s' % cti_profiles.name
new_profile['xlets'] = []
xlet = {}
xlet['name'] = cti_xlet.plugin_name
xlet['layout'] = cti_xlet_layout.name
xlet['floating'] = cti_profile_xlet.floating
xlet['closable'] = cti_profile_xlet.closable
xlet['movable'] = cti_profile_xlet.movable
xlet['scrollable'] = cti_profile_xlet.scrollable
xlet['order'] = cti_profile_xlet.order
res[cti_profiles.name]['xlets'].append(xlet)
return res
# -*- coding: utf-8 -*-
#
# Copyright (C) 2012 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.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Avencall. See the LICENSE file at top of the source tree
# or delivered in the installable package in which XiVO CTI Server is
# distributed for more details.
#
# 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 import dbconnection
from xivo_dao.alchemy.cti_service import CtiService
from xivo_dao.alchemy.cti_profile_service import CtiProfileService
from xivo_dao.alchemy.cti_profile import CtiProfile
_DB_NAME = 'asterisk'
def _session():
connection = dbconnection.get_connection(_DB_NAME)
return connection.get_session()
def _get(profile_id):
return _session().query(CtiService).filter(CtiService.id == profile_id).first()
def get_name(profile_id):
return _get(profile_id).name
def get_services():
res = {}
rows = (_session().query(CtiProfile).all())
for row in rows:
res[row.name] = []
rows = (_session().query(CtiProfile, CtiService)
.join((CtiProfileService, CtiProfileService.profile_id == CtiProfile.id),
(CtiService, CtiProfileService.service_id == CtiService.id))
.all())
for row in rows:
cti_profile, cti_service = row
profile = cti_profile.name
res[profile].append(cti_service.key)
return res
# -*- coding: utf-8 -*-
# Copyright (C) 2012 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.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Avencall. See the LICENSE file at top of the source tree
# or delivered in the installable package in which XiVO CTI Server is
# distributed for more details.
#
# 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 unittest
from xivo_dao.alchemy import dbconnection
from xivo_dao.alchemy.cti_profile import CtiProfile
from xivo_dao import cti_profile_dao, cti_preference_dao
from xivo_dao.alchemy.base import Base
from sqlalchemy.schema import MetaData
from xivo_dao.alchemy.ctiphonehintsgroup import CtiPhoneHintsGroup
from xivo_dao.alchemy.ctipresences import CtiPresences
from xivo_dao.alchemy.cti_profile_preference import CtiProfilePreference
from xivo_dao.alchemy.cti_preference import CtiPreference
class TestCtiPreferenceDAO(unittest.TestCase):
tables = [CtiProfile, CtiPhoneHintsGroup,
CtiPresences, CtiProfilePreference, CtiPreference]
@classmethod
def setUpClass(cls):
db_connection_pool = dbconnection.DBConnectionPool(dbconnection.DBConnection)
dbconnection.register_db_connection_pool(db_connection_pool)
uri = 'postgresql://asterisk:asterisk@localhost/asterisktest'
dbconnection.add_connection_as(uri, 'asterisk')
cls.connection = dbconnection.get_connection('asterisk')
cls.cleanTables()
cls.session = cls.connection.get_session()
@classmethod
def tearDownClass(cls):
dbconnection.unregister_db_connection_pool()
@classmethod
def cleanTables(cls):
if len(cls.tables):
engine = cls.connection.get_engine()
meta = MetaData(engine)
meta.reflect()
meta.drop_all()
table_list = [table.__table__ for table in cls.tables]
Base.metadata.create_all(engine, table_list)
engine.dispose()
def empty_tables(self):
for table in self.tables:
self.session.execute("TRUNCATE %s CASCADE;" % table.__tablename__)
def setUp(self):
self.empty_tables()
def test_get_name(self):
cti_profile = CtiProfile()
cti_profile.name = 'test_name'
self.session.add(cti_profile)
self.session.commit()
result = cti_profile_dao.get_name(cti_profile.id)
self.assertEqual(result, cti_profile.name)
def _add_presence(self, name):
cti_presence = CtiPresences()
cti_presence.name = name
self.session.add(cti_presence)
self.session.commit()
return cti_presence.id
def _add_preference(self, name):
cti_preference = CtiPreference()
cti_preference.option = name
self.session.add(cti_preference)
self.session.commit()
return cti_preference.id
def _add_phone_hints_group(self, name):
cti_phone_hints_group = CtiPhoneHintsGroup()
cti_phone_hints_group.name = name
self.session.add(cti_phone_hints_group)
self.session.commit()
return cti_phone_hints_group.id
def _add_profile(self, name):
cti_profile = CtiProfile()
cti_profile.name = name
cti_profile.presence_id = self._add_presence('test_presence')
cti_profile.phonehints_id = self._add_phone_hints_group('test_add_phone_hints_group')
self.session.add(cti_profile)
self.session.commit()
return cti_profile.id
def _add_preference_to_profile(self,
preference_id,
profile_id,
value):
cti_profile_preference = CtiProfilePreference()
cti_profile_preference.preference_id = preference_id
cti_profile_preference.profile_id = profile_id
cti_profile_preference.value = value
self.session.add(cti_profile_preference)
self.session.commit()
def test_get_profiles(self):
expected_result = {
"agent": {},
"client": {
"xlet.identity.logagent": "1",
"xlet.identity.pauseagent": "1"
}
}
preference1_id = self._add_preference('xlet.identity.logagent')
preference2_id = self._add_preference('xlet.identity.pauseagent')
agent_profile_id = self._add_profile('agent')
client_profile_id = self._add_profile('client')
self._add_preference_to_profile(preference1_id, client_profile_id, "1")
self._add_preference_to_profile(preference2_id, client_profile_id, "1")
result = cti_preference_dao.get_preferences()
self.assertEquals(result, expected_result)
......@@ -35,13 +35,15 @@ from xivo_dao.alchemy.cti_profile_xlet import CtiProfileXlet
from xivo_dao.alchemy.cti_service import CtiService
from xivo_dao.alchemy.cti_xlet import CtiXlet
from xivo_dao.alchemy.cti_xlet_layout import CtiXletLayout
from xivo_dao.alchemy.ctistatus import CtiStatus
class TestCtiProfileDAO(unittest.TestCase):
tables = [CtiProfile, CtiPreference, CtiPhoneHintsGroup,
CtiPresences, CtiProfilePreference, CtiProfileService,
CtiProfileXlet, CtiService, CtiXlet, CtiXletLayout]
CtiProfileXlet, CtiService, CtiXlet, CtiXletLayout,
CtiStatus]
@classmethod
def setUpClass(cls):
......@@ -81,12 +83,123 @@ class TestCtiProfileDAO(unittest.TestCase):
self.empty_tables()
def test_get_name(self):
session = dbconnection.get_connection('asterisk').get_session()
cti_profile = CtiProfile()
cti_profile.name = 'test_name'
session.add(cti_profile)
session.commit()
self.session.add(cti_profile)
self.session.commit()
result = cti_profile_dao.get_name(cti_profile.id)
self.assertEqual(result, cti_profile.name)
def _add_presence(self, name):
cti_presence = CtiPresences()
cti_presence.name = name
self.session.add(cti_presence)
self.session.commit()
return cti_presence.id
def _add_phone_hints_group(self, name):
cti_phone_hints_group = CtiPhoneHintsGroup()
cti_phone_hints_group.name = name
self.session.add(cti_phone_hints_group)
self.session.commit()
return cti_phone_hints_group.id
def _add_profile(self, name):
cti_profile = CtiProfile()
cti_profile.name = name
cti_profile.presence_id = self._add_presence('test_presence')
cti_profile.phonehints_id = self._add_phone_hints_group('test_add_phone_hints_group')
self.session.add(cti_profile)
self.session.commit()
return cti_profile.id
def _add_xlet_layout(self, name):
cti_xlet_layout = CtiXletLayout()
cti_xlet_layout.name = name
self.session.add(cti_xlet_layout)
self.session.commit()
return cti_xlet_layout.id
def _add_xlet(self, name):
cti_xlet = CtiXlet()
cti_xlet.plugin_name = name
self.session.add(cti_xlet)
self.session.commit()
return cti_xlet.id
def _add_xlet_to_profile(self,
xlet_id,
profile_id,
layout_id,
floating,
closable,
movable,
scrollable,
order):
cti_profile_xlet = CtiProfileXlet()
cti_profile_xlet.xlet_id = xlet_id
cti_profile_xlet.profile_id = profile_id
cti_profile_xlet.layout_id = layout_id
cti_profile_xlet.floating = floating
cti_profile_xlet.closable = closable
cti_profile_xlet.movable = movable
cti_profile_xlet.scrollable = scrollable
cti_profile_xlet.order = order
self.session.add(cti_profile_xlet)
self.session.commit()
def test_get_profiles(self):
expected_result = {
"test_profile": {
"name": "test_profile",
"phonestatus": "test_add_phone_hints_group",
"userstatus": "test_presence",
"preferences": "itm_preferences_test_profile",
"services": "itm_services_test_profile",
"xlets": [
{'name': 'tabber',
'layout': 'grid',
'floating': True,
'closable': True,
'movable': True,
'scrollable': True,
'order': 1},
{'name': 'agentdetails',
'layout': 'dock',
'floating': False,
'closable': True,
'movable': True,
'scrollable': True,
'order': 0}
]
}
}
profile_id = self._add_profile('test_profile')
xlet_layout_grid_id = self._add_xlet_layout('grid')
xlet_tabber_id = self._add_xlet('tabber')
self._add_xlet_to_profile(xlet_tabber_id,
profile_id,
xlet_layout_grid_id,
floating=True,
closable=True,
movable=True,
scrollable=True,
order=1)
xlet_agentdetails_id = self._add_xlet('agentdetails')
xlet_layout_dock_id = self._add_xlet_layout('dock')
self._add_xlet_to_profile(xlet_agentdetails_id,
profile_id,
xlet_layout_dock_id,
floating=False,
closable=True,
movable=True,
scrollable=True,
order=0)
result = cti_profile_dao.get_profiles()
self.assertEquals(result, expected_result)
# -*- coding: utf-8 -*-
# Copyright (C) 2012 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.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Avencall. See the LICENSE file at top of the source tree
# or delivered in the installable package in which XiVO CTI Server is
# distributed for more details.
#
# 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 unittest
from xivo_dao.alchemy import dbconnection
from xivo_dao.alchemy.cti_profile import CtiProfile
from xivo_dao import cti_profile_dao, cti_service_dao
from xivo_dao.alchemy.base import Base
from sqlalchemy.schema import MetaData
from xivo_dao.alchemy.ctiphonehintsgroup import CtiPhoneHintsGroup
from xivo_dao.alchemy.ctipresences import CtiPresences
from xivo_dao.alchemy.cti_profile_service import CtiProfileService
from xivo_dao.alchemy.cti_service import CtiService
class TestCtiServiceDAO(unittest.TestCase):
tables = [CtiProfile, CtiPhoneHintsGroup,
CtiPresences, CtiProfileService, CtiService]
@classmethod
def setUpClass(cls):
db_connection_pool = dbconnection.DBConnectionPool(dbconnection.DBConnection)
dbconnection.register_db_connection_pool(db_connection_pool)
uri = 'postgresql://asterisk:asterisk@localhost/asterisktest'
dbconnection.add_connection_as(uri, 'asterisk')
cls.connection = dbconnection.get_connection('asterisk')
cls.cleanTables()
cls.session = cls.connection.get_session()
@classmethod
def tearDownClass(cls):
dbconnection.unregister_db_connection_pool()
@classmethod
def cleanTables(cls):
if len(cls.tables):
engine = cls.connection.get_engine()
meta = MetaData(engine)
meta.reflect()
meta.drop_all()
table_list = [table.__table__ for table in cls.tables]
Base.metadata.create_all(engine, table_list)
engine.dispose()
def empty_tables(self):
for table in self.tables:
self.session.execute("TRUNCATE %s CASCADE;" % table.__tablename__)
def setUp(self):
self.empty_tables()
def test_get_name(self):
cti_profile = CtiProfile()
cti_profile.name = 'test_name'
self.session.add(cti_profile)
self.session.commit()
result = cti_profile_dao.get_name(cti_profile.id)
self.assertEqual(result, cti_profile.name)
def _add_presence(self, name):
cti_presence = CtiPresences()
cti_presence.name = name
self.session.add(cti_presence)
self.session.commit()
return cti_presence.id
def _add_service(self, key):
cti_service = CtiService()
cti_service.key = key
self.session.add(cti_service)
self.session.commit()
return cti_service.id
def _add_phone_hints_group(self, name):
cti_phone_hints_group = CtiPhoneHintsGroup()
cti_phone_hints_group.name = name
self.session.add(cti_phone_hints_group)
self.session.commit()
return cti_phone_hints_group.id
def _add_profile(self, name):
cti_profile = CtiProfile()
cti_profile.name = name
cti_profile.presence_id = self._add_presence('test_presence')
cti_profile.phonehints_id = self._add_phone_hints_group('test_add_phone_hints_group')
self.session.add(cti_profile)
self.session.commit()
return cti_profile.id
def _add_service_to_profile(self,
service_id,
profile_id):
cti_profile_service = CtiProfileService()
cti_profile_service.service_id = service_id
cti_profile_service.profile_id = profile_id
self.session.add(cti_profile_service)
self.session.commit()
def test_get_profiles(self):
expected_result = {
"agent": [],
"test_profile": ["enablednd",
"fwdunc",
"fwdbusy",
"fwdrna"]
}
service1_id = self._add_service('enablednd')
service2_id = self._add_service('fwdunc')
service3_id = self._add_service('fwdbusy')
service4_id = self._add_service('fwdrna')
agent_profile_id = self._add_profile('agent')
test_profile_id = self._add_profile('test_profile')
self._add_service_to_profile(service1_id, test_profile_id)
self._add_service_to_profile(service2_id, test_profile_id)
self._add_service_to_profile(service3_id, test_profile_id)
self._add_service_to_profile(service4_id, test_profile_id)
result = cti_service_dao.get_services()
self.assertEquals(result, expected_result)
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