Commit a3683e0c authored by Vojtech Sodoma's avatar Vojtech Sodoma

Merge branch 'master' into 2936-disabled-users-generated

# Conflicts:
#	debian/changelog
parents a3ff509f 74542fa4
xivo-dao (2019.05.13) xivo-callisto; urgency=medium
xivo-dao (2019.12.06) xivo-deneb; urgency=medium
* 2936 Disabled extensions are still generated in the dialplan
-- Vojtech Sodoma <vsodoma@avencall.com> Tue, 05 Nov 2019 12:13:46 +0100
-- Vojtech Sodoma <vsodoma@avencall.com> Tue, 05
xivo-dao (2019.12.03) xivo-deneb; urgency=medium
* Bump version to 2019.12.03
* Add dao layer to create,update,delete and search agent
-- Jean-Pierre Thomasset <jpthomasset@gmail.com> Mon, 28 Oct 2019 14:10:07 +0000
xivo-dao (2019.09.00) xivo-deneb; urgency=medium
* 2733 Add dissuasion_access type to right_type
-- Etienne Allovon <eallovon@avencall.com> Tue, 06 Aug 2019 15:56:37 +0000
xivo-dao (2019.05.06) xivo-callisto; urgency=medium
......
......@@ -34,7 +34,7 @@ class Rights(Base):
id = Column(Integer, nullable=False)
user_id = Column(Integer, nullable=False)
category = Column(Enum('queue', 'agentgroup', 'incall', 'recording_access',
category = Column(Enum('queue', 'agentgroup', 'incall', 'recording_access', 'dissuasion_access',
name='right_type',
metadata=Base.metadata),
nullable=False)
......
......@@ -17,7 +17,13 @@
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
from xivo_dao.alchemy.agentfeatures import AgentFeatures
from xivo_dao.alchemy.agentgroup import AgentGroup
from xivo_dao.alchemy.queuemember import QueueMember
from xivo_dao.alchemy.queuefeatures import QueueFeatures
from xivo_dao.helpers.db_manager import daosession
from xivo_dao.helpers.db_manager import Session
from xivo_dao.resources.agent.search import agent_search
from xivo_dao.resources.utils.search import SearchResult
@daosession
......@@ -27,3 +33,51 @@ def exists(session, agent_id):
)
return query.count() > 0
def search(**parameters):
rows, total = agent_search.search(Session, parameters)
return SearchResult(total, rows)
def get(agent_id):
query = Session.query(AgentFeatures).filter_by(id=agent_id)
return query.first()
def create(agent):
Session.add(agent)
Session.flush()
return agent
def edit(agent):
Session.add(agent)
Session.flush()
return agent
def delete(agent):
Session.delete(agent)
Session.flush()
def to_membership_dict(row):
return {
'agent_id': row[0],
'queue_id': row[1],
'penalty': row[2]
}
def get_membership(agent_id):
query = (Session.query(QueueMember.userid.label('agent_id'),
QueueFeatures.id.label("queue_id"),
QueueMember.penalty)
.filter(QueueFeatures.name == QueueMember.queue_name)
.filter(QueueMember.usertype == 'agent')
.filter(QueueMember.userid == agent_id))
rows = query.all()
return [to_membership_dict(item) for item in rows]
def get_group_by_id(group_id):
query = (Session.query(AgentGroup)
.filter(AgentGroup.id == group_id))
return query.first()
# -*- coding: utf-8 -*-
# Copyright (C) 2019 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 sqlalchemy.sql import and_
from xivo_dao.alchemy.agentfeatures import AgentFeatures
from xivo_dao.resources.utils.search import SearchSystem
from xivo_dao.resources.utils.search import SearchConfig
config = SearchConfig(table=AgentFeatures,
columns={'id': AgentFeatures.id,
'firstname': AgentFeatures.firstname,
'lastname': AgentFeatures.lastname,
'fullname': (AgentFeatures.firstname + " " + AgentFeatures.lastname),
'description': AgentFeatures.description,
'number': AgentFeatures.number,
'numgroup': AgentFeatures.numgroup,
'context': AgentFeatures.context},
search=['fullname',
'description',
'number',
'numgroup',
'context'],
default_sort='lastname')
agent_search = SearchSystem(config)
......@@ -16,11 +16,15 @@
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
from hamcrest import assert_that, equal_to
from hamcrest import assert_that, equal_to, contains_inanyorder
from xivo_dao import queue_member_dao
from xivo_dao.alchemy.agentfeatures import AgentFeatures
from xivo_dao.alchemy.queuemember import QueueMember
from xivo_dao.alchemy.queuefeatures import QueueFeatures
from xivo_dao.tests.test_dao import DAOTestCase
from xivo_dao.resources.agent import dao as agent_dao
from xivo_dao.resources.utils.search import SearchResult
class TestAgentExist(DAOTestCase):
......@@ -35,3 +39,95 @@ class TestAgentExist(DAOTestCase):
result = agent_dao.exists(agent_row.id)
assert_that(result, equal_to(True))
class BaseTestAgentDAO(DAOTestCase):
def add_agent(self, **kwargs):
print(kwargs)
kwargs.setdefault('numgroup', 1)
kwargs.setdefault('passwd', '')
kwargs.setdefault('context', 'default')
kwargs.setdefault('language', '')
kwargs.setdefault('description', '')
print(kwargs)
agent = AgentFeatures(**kwargs)
self.session.add(agent)
self.session.flush()
return agent
class TestSearch(BaseTestAgentDAO):
def assert_search_returns_result(self, search_result, **parameters):
result = agent_dao.search(**parameters)
assert_that(result, equal_to(search_result))
def test_empty_result(self):
expected = SearchResult(0, [])
self.assert_search_returns_result(expected)
def test_return_all(self):
agent1 = self.add_agent(firstname='bob', number='2001')
agent2 = self.add_agent(firstname='sully', number='2002')
expected = SearchResult(2, [agent1, agent2])
self.assert_search_returns_result(expected)
def test_return_match(self):
agent1 = self.add_agent(firstname='bob', number='2001')
agent2 = self.add_agent(firstname='sully', number='2002')
expected = SearchResult(1, [agent2])
self.assert_search_returns_result(expected, firstname='sully')
class TestAgentCRUD(BaseTestAgentDAO):
def test_get_agent(self):
expected = self.add_agent(firstname='James', lastname='Bond', number='2002')
agent = agent_dao.get(expected.id)
assert_that(agent, equal_to(expected))
def test_create_agent(self):
props = {
'firstname': 'Jack',
'lastname': 'Bauer',
'number': '2003',
'numgroup': 1,
'passwd': '',
'context': 'default',
'language': 'en_US',
'description': 'An agent'
}
agent = agent_dao.create(AgentFeatures(**props))
expected = agent_dao.get(agent.id)
assert_that(agent, equal_to(expected))
def test_edit_agent(self):
agent = self.add_agent(firstname='James', lastname='Bond', number='2002')
agent.firstname= 'Secret'
agent.lastname= 'Agent'
agent_dao.edit(agent)
result = agent_dao.get(agent.id)
assert_that(agent, equal_to(result))
class TestAgentMembership(BaseTestAgentDAO):
def test_get_agent_membership(self):
agent_id = 123
agent_number = '123'
queue1 = QueueFeatures(name='queue1', displayname='queue 1')
queue2 = QueueFeatures(name='queue2', displayname='queue 2')
self.session.add(queue1)
self.session.add(queue2)
self.session.flush()
queue_member_dao.add_agent_to_queue(agent_id, agent_number, queue1.name)
queue_member_dao.add_agent_to_queue(agent_id, agent_number, queue2.name)
res = agent_dao.get_membership(123)
member1 = {'agent_id': 123, 'queue_id': queue1.id, 'penalty': 0}
member2 = {'agent_id': 123, 'queue_id': queue2.id, 'penalty': 0}
assert_that(res, contains_inanyorder(member1, member2))
......@@ -49,6 +49,12 @@ class DestinationPersistor(object):
self.session.add(destination)
self.session.flush()
def delete_agent_destination(self, agent):
(self.session
.query(FuncKeyDestAgent)
.filter_by(agent_id=agent.id)
.delete())
def delete_user_destination(self, user):
self.delete_user_func_key(user)
self.delete_bsfilter_func_key(user)
......
......@@ -69,6 +69,14 @@ class TestCreateAgentDestination(TestFuncKeyDao):
agent_destination_rows = self.find_all_destinations('agent', agent_row.id).all()
self.assert_func_key_rows_created(agent_destination_rows)
def test_agent_destination_func_key_deleted(self):
self.prepare_exten()
agent_row = self.add_agent()
self.persistor.create_agent_destination(agent_row)
self.persistor.delete_agent_destination(agent_row)
self.assert_destination_deleted('agent', agent_row.id)
def assert_func_key_rows_created(self, destination_rows):
assert_that(destination_rows, is_not(none()))
assert_that(destination_rows, has_length(3))
......
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