Network deletion and subnet creation bug fixes

bug 1017395

1. When a subnet is created the network id is validated
2. when a network is deleted - if a subnet or port is defined then
the opertaion fails

Change-Id: I746d0f4994c539a908df0220465d753bc3eb8e04

diff --git a/quantum/db/db_base_plugin_v2.py b/quantum/db/db_base_plugin_v2.py
index 8bc51bf9be7bb1e791d5b991b065c41b7961b0bd..a2072dcc74d25695c1d3806b3a1ff8397c350610 100644 (file)
--- a/quantum/db/db_base_plugin_v2.py
+++ b/quantum/db/db_base_plugin_v2.py
@@ -216,13 +216,13 @@ class QuantumDbPluginV2(quantum_plugin_base_v2.QuantumPluginBaseV2):
         with context.session.begin():
             network = self._get_network(context, id)
 
-            # TODO(anyone) Delegation?
-            ports_qry = context.session.query(models_v2.Port)
-            ports_qry.filter_by(network_id=id).delete()
+            filter = {'network_id': [id]}
+            ports = self.get_ports(context, filters=filter)
+            if ports:
+                raise q_exc.NetworkInUse(net_id=id)
 
             subnets_qry = context.session.query(models_v2.Subnet)
             subnets_qry.filter_by(network_id=id).delete()
-
             context.session.delete(network)
 
     def get_network(self, context, id, fields=None, verbose=None):
@@ -243,6 +243,7 @@ class QuantumDbPluginV2(quantum_plugin_base_v2.QuantumPluginBaseV2):
             s['gateway_ip'] = str(netaddr.IPAddress(net.first + 1))
 
         with context.session.begin():
+            network = self._get_network(context, s["network_id"])
             subnet = models_v2.Subnet(network_id=s['network_id'],
                                       ip_version=s['ip_version'],
                                       cidr=s['cidr'],

diff --git a/quantum/tests/unit/test_db_plugin.py b/quantum/tests/unit/test_db_plugin.py
index 4351ba05fc2c8d5fd8debe866424cbda1a2f6bf4..621d95e93db64b6c22491f3d000548ad91be65d1 100644 (file)
--- a/quantum/tests/unit/test_db_plugin.py
+++ b/quantum/tests/unit/test_db_plugin.py
@@ -253,6 +253,15 @@ class TestPortsV2(QuantumDbPluginV2TestCase):
             self.assertEqual(res['port']['admin_state_up'],
                              data['port']['admin_state_up'])
 
+    def test_delete_network_if_port_exists(self):
+        fmt = 'json'
+        with self.port() as port:
+            net_id = port['port']['network_id']
+            req = self.new_delete_request('networks',
+                                          port['port']['network_id'])
+            res = req.get_response(self.api)
+            self.assertEquals(res.status_int, 409)
+
     def test_requested_duplicate_mac(self):
         fmt = 'json'
         with self.port() as port: