Commit 6a1dac86 authored by Etienne Allovon's avatar Etienne Allovon

Merge branch '2352-fix-agent-api-funckey' into '2018.16'

2351 - Add api to create func key destination for agent

See merge request !35
parents 7c188ffd 1ac783e7
xivo-dao (2018.16.06) xivo-borealis; urgency=medium
* 2352 Add api to create func key destination for agent
-- Jean-Pierre Thomasset <jpthomasset@gmail.com> Fri, 15 Feb 2019 16:16:07 +0000
xivo-dao (2018.16.04) xivo-borealis; urgency=medium
* 2266 Add agent_group_id in user api
......
......@@ -39,7 +39,15 @@ class FuncKey(Base):
destination_type = relationship(FuncKeyDestinationType, foreign_keys=destination_type_id)
@classmethod
def new_for_user(cls):
def new_for_type(cls, dest_type):
type_id = FuncKeyType.query_id('speeddial').as_scalar()
destination_type_id = FuncKeyDestinationType.query_id('user').as_scalar()
destination_type_id = FuncKeyDestinationType.query_id(dest_type).as_scalar()
return cls(type_id=type_id, destination_type_id=destination_type_id)
@classmethod
def new_for_user(cls):
return FuncKey().new_for_type('user')
@classmethod
def new_for_agent(cls):
return FuncKey().new_for_type('agent')
......@@ -19,10 +19,12 @@
from sqlalchemy import and_
from xivo_dao.alchemy.callfiltermember import Callfiltermember
from xivo_dao.alchemy.extension import Extension
from xivo_dao.alchemy.func_key import FuncKey
from xivo_dao.alchemy.func_key_mapping import FuncKeyMapping
from xivo_dao.alchemy.func_key_dest_bsfilter import FuncKeyDestBSFilter
from xivo_dao.alchemy.func_key_dest_user import FuncKeyDestUser
from xivo_dao.alchemy.func_key_dest_agent import FuncKeyDestAgent
class DestinationPersistor(object):
......@@ -37,6 +39,16 @@ class DestinationPersistor(object):
self.session.add(destination)
self.session.flush()
def create_agent_destination(self, agent):
typevals = ['agentstaticlogin', 'agentstaticlogoff', 'agentstaticlogtoggle']
extensions = self.session.query(Extension).filter(Extension.typeval.in_(typevals))
for extension in extensions:
func_key = FuncKey.new_for_agent()
destination = FuncKeyDestAgent(func_key=func_key, agent=agent, extension=extension)
self.session.add(func_key)
self.session.add(destination)
self.session.flush()
def delete_user_destination(self, user):
self.delete_user_func_key(user)
self.delete_bsfilter_func_key(user)
......
......@@ -15,7 +15,7 @@
# 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, none, contains, is_not, equal_to
from hamcrest import assert_that, none, contains, is_not, equal_to, has_length
from xivo_dao.tests.test_dao import DAOTestCase
from xivo_dao.resources.func_key.tests.test_helpers import FuncKeyHelper
......@@ -53,6 +53,25 @@ class TestCreateUserDestination(TestFuncKeyDao):
.first())
assert_that(row, is_not(none()))
class TestCreateAgentDestination(TestFuncKeyDao):
def prepare_exten(self):
self.add_extenfeatures('_*31.', 'agentstaticlogin')
self.add_extenfeatures('_*32.', 'agentstaticlogoff')
self.add_extenfeatures('_*30.', 'agentstaticlogtoggle')
def test_given_agent_destination_then_func_key_created(self):
self.prepare_exten()
agent_row = self.add_agent()
self.persistor.create_agent_destination(agent_row)
agent_destination_rows = self.find_all_destinations('agent', agent_row.id).all()
self.assert_func_key_rows_created(agent_destination_rows)
def assert_func_key_rows_created(self, destination_rows):
assert_that(destination_rows, is_not(none()))
assert_that(destination_rows, has_length(3))
class TestDeleteUserDestination(TestFuncKeyDao):
......
......@@ -61,6 +61,7 @@ class FuncKeyHelper(object):
'features': (FuncKeyDestFeaturesSchema, 'features_id', 8),
'paging': (FuncKeyDestPagingSchema, 'paging_id', 9),
'custom': (FuncKeyDestCustomSchema, 'exten', 10),
'agent': (FuncKeyDestAgentSchema, 'agent_id', 11),
'bsfilter': (FuncKeyDestBSFilterSchema, 'filtermember_id', 12),
}
......@@ -227,15 +228,17 @@ class FuncKeyHelper(object):
position=position)
self.add_me(mapping_row)
def find_destination(self, destination, destination_id):
def find_all_destinations(self, destination, destination_id):
schema, column_name, _ = self.destinations[destination]
column = getattr(schema, column_name)
row = (self.session.query(schema)
.filter(column == destination_id)
.first())
rows = (self.session.query(schema)
.filter(column == destination_id))
return rows
return row
def find_destination(self, destination, destination_id):
return self.find_all_destinations(destination, destination_id).first()
def assert_destination_deleted(self, destination, destination_id):
row = self.find_destination(destination, destination_id)
......
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