x/vmware-nsx
Code Issues Proposed changes

LBaaS: throw proper exception on duplicating monitor association

Browse Source 
Fixes bug 1208728

Change-Id: I6cf27766eedff34a6ed5062a50b049ab0a5bf96a
This commit is contained in:
Oleg Bondarev 2013-08-06 17:44:30 +04:00
parent e70b10ffc7
commit 012348829b
3 changed files with 30 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
neutron/db/loadbalancer/loadbalancer_db.py
View File

@ -565,6 +565,13 @@ class LoadBalancerPluginDb(LoadBalancerPluginBase,
def create_pool_health_monitor(self, context, health_monitor, pool_id):
monitor_id = health_monitor['health_monitor']['id']
with context.session.begin(subtransactions=True):
assoc_qry = context.session.query(PoolMonitorAssociation)
assoc = assoc_qry.filter_by(pool_id=pool_id,
monitor_id=monitor_id).first()
if assoc:
raise loadbalancer.PoolMonitorAssociationExists(
monitor_id=monitor_id, pool_id=pool_id)
pool = self._get_resource(context, Pool, pool_id)
assoc = PoolMonitorAssociation(pool_id=pool_id,

5
neutron/extensions/loadbalancer.py
View File

@ -54,6 +54,11 @@ class PoolMonitorAssociationNotFound(qexception.NotFound):
"with Pool %(pool_id)s")
class PoolMonitorAssociationExists(qexception.Conflict):
message = _('health_monitor %(monitor_id)s is already associated '
'with pool %(pool_id)s')
class StateInvalid(qexception.NeutronException):
message = _("Invalid state %(state)s of Loadbalancer resource %(id)s")

18
neutron/tests/unit/db/loadbalancer/test_db_loadbalancer.py
View File

@ -1210,6 +1210,24 @@ class TestLoadBalancer(LoadBalancerPluginDbTestCase):
driver_call.assert_called_once_with(
mock.ANY, hm['health_monitor'], pool['pool']['id'])
def test_create_pool_health_monitor_already_associated(self):
with contextlib.nested(
self.pool(name="pool"),
self.health_monitor(),
) as (pool, hm):
res = self.plugin.create_pool_health_monitor(
context.get_admin_context(),
hm, pool['pool']['id']
)
self.assertEqual({'health_monitor':
[hm['health_monitor']['id']]},
res)
self.assertRaises(loadbalancer.PoolMonitorAssociationExists,
self.plugin.create_pool_health_monitor,
context.get_admin_context(),
hm,
pool['pool']['id'])
def test_create_pool_healthmon_invalid_pool_id(self):
with self.health_monitor() as healthmon:
self.assertRaises(loadbalancer.PoolNotFound,