Commit 04bc73da authored by Francois Blackburn's avatar Francois Blackburn

manage outcalltrunk via relationship

Instead of create an outcalltrunk resource, we use relationship to
manage the association table.
parent f305f7d7
......@@ -17,7 +17,9 @@
import itertools
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.ext.orderinglist import ordering_list
from sqlalchemy.orm import relationship
from sqlalchemy.schema import Column, PrimaryKeyConstraint, UniqueConstraint
from sqlalchemy.sql import func, cast, not_
......@@ -25,6 +27,7 @@ from xivo_dao.alchemy.dialpattern import DialPattern
from sqlalchemy.types import Integer, String, Text, Boolean
from xivo_dao.helpers.db_manager import Base
from xivo_dao.alchemy.outcalltrunk import OutcallTrunk
class Outcall(Base):
......@@ -51,10 +54,14 @@ class Outcall(Base):
foreign_keys='DialPattern.typeid',
cascade='all, delete-orphan')
trunks = relationship('TrunkFeatures',
secondary='outcalltrunk',
order_by='OutcallTrunk.priority',
back_populates='outcalls')
outcall_trunks = relationship('OutcallTrunk',
order_by='OutcallTrunk.priority',
collection_class=ordering_list('priority'),
cascade='all, delete-orphan',
back_populates='outcall')
trunks = association_proxy('outcall_trunks', 'trunk',
creator=lambda _trunk: OutcallTrunk(trunk=_trunk))
@hybrid_property
def internal_caller_id(self):
......
......@@ -16,6 +16,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import relationship
from sqlalchemy.schema import Column, ForeignKey, PrimaryKeyConstraint, Index
from sqlalchemy.types import Integer
......@@ -34,6 +35,12 @@ class OutcallTrunk(Base):
trunkfeaturesid = Column(Integer, ForeignKey('trunkfeatures.id'), nullable=False)
priority = Column(Integer, nullable=False, server_default='0')
trunk = relationship('TrunkFeatures',
back_populates='outcall_trunks')
outcall = relationship('Outcall',
back_populates='outcall_trunks')
@hybrid_property
def outcall_id(self):
return self.outcallid
......
......@@ -15,6 +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 sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import relationship, backref
from sqlalchemy.schema import Column, PrimaryKeyConstraint, UniqueConstraint, Index
......@@ -25,6 +26,7 @@ from xivo_dao.alchemy import enum
from xivo_dao.alchemy.usersip import UserSIP
from xivo_dao.alchemy.useriax import UserIAX
from xivo_dao.alchemy.usercustom import UserCustom
from xivo_dao.alchemy.outcalltrunk import OutcallTrunk
class TrunkFeatures(Base):
......@@ -69,9 +71,12 @@ class TrunkFeatures(Base):
foreign_keys=[protocolid],
backref=backref('trunk', uselist=False))
outcalls = relationship('Outcall',
secondary='outcalltrunk',
back_populates='trunks')
outcall_trunks = relationship('OutcallTrunk',
cascade='all, delete-orphan',
back_populates='trunk')
outcalls = association_proxy('outcall_trunks', 'outcall',
creator=lambda _outcall: OutcallTrunk(outcall=_outcall))
@hybrid_property
def endpoint(self):
......
......@@ -17,17 +17,22 @@
from hamcrest import (assert_that,
contains,
contains_inanyorder,
equal_to,
empty,
has_items,
has_properties,
has_property,
is_not,
none)
none,
not_)
from xivo_dao.alchemy.dialpattern import DialPattern
from xivo_dao.alchemy.extension import Extension
from xivo_dao.alchemy.outcall import Outcall
from xivo_dao.alchemy.outcalltrunk import OutcallTrunk
from xivo_dao.alchemy.rightcallmember import RightCallMember
from xivo_dao.alchemy.trunkfeatures import TrunkFeatures
from xivo_dao.alchemy.schedulepath import SchedulePath
from xivo_dao.tests.test_dao import DAOTestCase
from xivo_dao.resources.outcall import dao as outcall_dao
......@@ -62,20 +67,6 @@ class TestGet(DAOTestCase):
assert_that(outcall, equal_to(outcall_row))
def test_trunks_relationship(self):
outcall_row = self.add_outcall()
trunk1_row = self.add_trunk()
trunk2_row = self.add_trunk()
trunk3_row = self.add_trunk()
self.add_outcall_trunk(outcall_id=outcall_row.id, trunk_id=trunk1_row.id, priority=2)
self.add_outcall_trunk(outcall_id=outcall_row.id, trunk_id=trunk2_row.id, priority=0)
self.add_outcall_trunk(outcall_id=outcall_row.id, trunk_id=trunk3_row.id, priority=1)
outcall = outcall_dao.get(outcall_row.id)
assert_that(outcall, equal_to(outcall_row))
assert_that(outcall.trunks, contains(trunk2_row, trunk3_row, trunk1_row))
class TestFindBy(DAOTestCase):
......@@ -468,3 +459,51 @@ class TestDelete(DAOTestCase):
extension = self.session.query(Extension).first()
assert_that(extension, none())
class TestRelations(DAOTestCase):
def test_outcalls_create(self):
trunk = TrunkFeatures()
outcall_row = self.add_outcall()
outcall_row.trunks = [trunk]
outcall = outcall_dao.get(outcall_row.id)
assert_that(outcall, equal_to(outcall_row))
assert_that(outcall.trunks, contains_inanyorder(trunk))
def test_outcalls_association(self):
outcall_row = self.add_outcall()
trunk1_row = self.add_trunk()
trunk2_row = self.add_trunk()
trunk3_row = self.add_trunk()
outcall_row.trunks = [trunk2_row, trunk3_row, trunk1_row]
outcall = outcall_dao.get(outcall_row.id)
assert_that(outcall, equal_to(outcall_row))
assert_that(outcall.trunks, contains(trunk2_row, trunk3_row, trunk1_row))
def test_outcalls_dissociation(self):
outcall_row = self.add_outcall()
trunk1_row = self.add_trunk()
trunk2_row = self.add_trunk()
trunk3_row = self.add_trunk()
outcall_row.trunks = [trunk2_row, trunk3_row, trunk1_row]
outcall = outcall_dao.get(outcall_row.id)
assert_that(outcall, equal_to(outcall_row))
assert_that(outcall.trunks, not_(empty()))
outcall_row.trunks = []
outcall = outcall_dao.get(outcall_row.id)
assert_that(outcall, equal_to(outcall_row))
assert_that(outcall.trunks, empty())
row = self.session.query(TrunkFeatures).first()
assert_that(row, not_(none()))
row = self.session.query(OutcallTrunk).first()
assert_that(row, none())
......@@ -21,13 +21,17 @@ from hamcrest import (assert_that,
contains,
contains_inanyorder,
equal_to,
empty,
has_items,
has_properties,
has_property,
is_not,
none)
none,
not_)
from xivo_dao.alchemy.outcall import Outcall
from xivo_dao.alchemy.outcalltrunk import OutcallTrunk
from xivo_dao.alchemy.trunkfeatures import TrunkFeatures as Trunk
from xivo_dao.alchemy.staticiax import StaticIAX
from xivo_dao.alchemy.staticsip import StaticSIP
......@@ -396,3 +400,51 @@ class TestDelete(DAOTestCase):
deleted_register = self.session.query(StaticIAX).filter(StaticIAX.id == trunk.registerid).first()
assert_that(deleted_register, none())
class TestRelations(DAOTestCase):
def test_outcalls_create(self):
outcall = Outcall(name='test', context='to-extern')
trunk_row = self.add_trunk()
trunk_row.outcalls = [outcall]
trunk = trunk_dao.get(trunk_row.id)
assert_that(trunk, equal_to(trunk_row))
assert_that(trunk.outcalls, contains_inanyorder(outcall))
def test_outcalls_association(self):
outcall1_row = self.add_outcall()
outcall2_row = self.add_outcall()
outcall3_row = self.add_outcall()
trunk_row = self.add_trunk()
trunk_row.outcalls = [outcall1_row, outcall2_row, outcall3_row]
trunk = trunk_dao.get(trunk_row.id)
assert_that(trunk, equal_to(trunk_row))
assert_that(trunk.outcalls, contains_inanyorder(outcall1_row, outcall2_row, outcall3_row))
def test_outcalls_dissociation(self):
outcall1_row = self.add_outcall()
outcall2_row = self.add_outcall()
outcall3_row = self.add_outcall()
trunk_row = self.add_trunk()
trunk_row.outcalls = [outcall1_row, outcall2_row, outcall3_row]
trunk = trunk_dao.get(trunk_row.id)
assert_that(trunk, equal_to(trunk_row))
assert_that(trunk.outcalls, not_(empty()))
trunk_row.outcalls = []
trunk = trunk_dao.get(trunk_row.id)
assert_that(trunk, equal_to(trunk_row))
assert_that(trunk.outcalls, empty())
row = self.session.query(Outcall).first()
assert_that(row, not_(none()))
row = self.session.query(OutcallTrunk).first()
assert_that(row, none())
......@@ -65,7 +65,6 @@ from xivo_dao.alchemy.queuemember import QueueMember
from xivo_dao.alchemy.queueskill import QueueSkill
from xivo_dao.alchemy.queueskillrule import QueueSkillRule
from xivo_dao.alchemy.outcall import Outcall
from xivo_dao.alchemy.outcalltrunk import OutcallTrunk
from xivo_dao.alchemy.rightcall import RightCall as CallPermission
from xivo_dao.alchemy.rightcallmember import RightCallMember as CallPermissionAssociation
from xivo_dao.alchemy.sccpdevice import SCCPDevice as SCCPDeviceSchema
......@@ -443,11 +442,6 @@ class DAOTestCase(unittest.TestCase):
self.add_me(outcall)
return outcall
def add_outcall_trunk(self, **kwargs):
outcall_trunk = OutcallTrunk(**kwargs)
self.add_me(outcall_trunk)
return outcall_trunk
def add_outcall_call_permission(self, **kwargs):
kwargs.setdefault('type', 'outcall')
outcall_call_permission = CallPermissionAssociation(**kwargs)
......
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