Refactor _populate_ports_for_subnets for testability

I want to add to this function but I found it very difficult to test
the additions that I was making (because this code isn't well covered
by UTs).  This patch adds some unit testing so that I can change it
more confidently but does not change any functionality.

Change-Id: Id2cf57968f2dfb8bcf18946c922ce2bb51a27766
Partially-implements:  bp/address-scopes

diff --git a/neutron/db/l3_db.py b/neutron/db/l3_db.py
index b465b5042e43d81adb7149a320e2cca62c3f6c28..b5e6191ffea896fb5e65770c7eeb5447ffa89055 100644 (file)
--- a/neutron/db/l3_db.py
+++ b/neutron/db/l3_db.py
@@ -1186,27 +1186,22 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase):
                       for rp in qry]
         return interfaces
 
-    def _populate_subnets_for_ports(self, context, ports):
-        """Populate ports with subnets.
-
-        These ports already have fixed_ips populated.
-        """
-        def each_port_having_fixed_ips():
-            for port in ports or []:
-                fixed_ips = port.get('fixed_ips', [])
-                if not fixed_ips:
-                    # Skip ports without IPs, which can occur if a subnet
-                    # attached to a router is deleted
-                    LOG.info(_LI("Skipping port %s as no IP is configure on "
-                                 "it"),
-                             port['id'])
-                    continue
-                yield port
-
-        network_ids = set(p['network_id']
-                          for p in each_port_having_fixed_ips())
+    @staticmethod
+    def _each_port_having_fixed_ips(ports):
+        for port in ports or []:
+            fixed_ips = port.get('fixed_ips', [])
+            if not fixed_ips:
+                # Skip ports without IPs, which can occur if a subnet
+                # attached to a router is deleted
+                LOG.info(_LI("Skipping port %s as no IP is configure on "
+                             "it"),
+                         port['id'])
+                continue
+            yield port
+
+    def _get_subnets_by_network_list(self, context, network_ids):
         if not network_ids:
-            return
+            return {}
 
         filters = {'network_id': [id for id in network_ids]}
         fields = ['id', 'cidr', 'gateway_ip',
@@ -1215,8 +1210,20 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase):
         subnets_by_network = dict((id, []) for id in network_ids)
         for subnet in self._core_plugin.get_subnets(context, filters, fields):
             subnets_by_network[subnet['network_id']].append(subnet)
+        return subnets_by_network
+
+    def _populate_subnets_for_ports(self, context, ports):
+        """Populate ports with subnets.
+
+        These ports already have fixed_ips populated.
+        """
+        network_ids = [p['network_id']
+                       for p in self._each_port_having_fixed_ips(ports)]
+
+        subnets_by_network = self._get_subnets_by_network_list(
+            context, network_ids)
 
-        for port in each_port_having_fixed_ips():
+        for port in self._each_port_having_fixed_ips(ports):
 
             port['subnets'] = []
             port['extra_subnets'] = []

diff --git a/neutron/tests/unit/db/test_l3_db.py b/neutron/tests/unit/db/test_l3_db.py
new file mode 100644 (file)
index 0000000..19171c8
--- /dev/null
+++ b/neutron/tests/unit/db/test_l3_db.py
@@ -0,0 +1,84 @@
+# Copyright 2015 Hewlett-Packard Development Company, L.P.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import mock
+
+from neutron.db import l3_db
+from neutron import manager
+from neutron.tests import base
+
+
+class TestL3_NAT_dbonly_mixin(base.BaseTestCase):
+    def setUp(self):
+        super(TestL3_NAT_dbonly_mixin, self).setUp()
+        self.db = l3_db.L3_NAT_dbonly_mixin()
+
+    def test__each_port_having_fixed_ips_none(self):
+        """Be sure the method returns an empty list when None is passed"""
+        filtered = l3_db.L3_NAT_dbonly_mixin._each_port_having_fixed_ips(None)
+        self.assertEqual([], list(filtered))
+
+    def test__each_port_having_fixed_ips(self):
+        """Basic test that ports without fixed ips are filtered out"""
+        ports = [{'id': 'a', 'fixed_ips': [mock.sentinel.fixedip]},
+                 {'id': 'b'}]
+        filtered = l3_db.L3_NAT_dbonly_mixin._each_port_having_fixed_ips(ports)
+        ids = [p['id'] for p in filtered]
+        self.assertEqual(['a'], ids)
+
+    def test__get_subnets_by_network_no_query(self):
+        """Basic test that no query is performed if no Ports are passed"""
+        with mock.patch.object(manager.NeutronManager, 'get_plugin') as get_p:
+            self.db._get_subnets_by_network_list(mock.sentinel.context, [])
+        self.assertFalse(get_p().get_subnets.called)
+
+    def test__get_subnets_by_network(self):
+        """Basic test that the right query is called"""
+        network_ids = ['a', 'b']
+        with mock.patch.object(manager.NeutronManager, 'get_plugin') as get_p:
+            self.db._get_subnets_by_network_list(
+                mock.sentinel.context, network_ids)
+        get_p().get_subnets.assert_called_once_with(
+            mock.sentinel.context,
+            {'network_id': network_ids},
+            mock.ANY)
+
+    def test__populate_ports_for_subnets_none(self):
+        """Basic test that the method runs correctly with no ports"""
+        ports = []
+        self.db._populate_subnets_for_ports(mock.sentinel.context, ports)
+        self.assertEqual([], ports)
+
+    @mock.patch.object(l3_db.L3_NAT_dbonly_mixin,
+                       '_get_subnets_by_network_list')
+    def test__populate_ports_for_subnets(self, get_subnets_by_network):
+        cidr = "2001:db8::/64"
+        subnet = {'id': mock.sentinel.subnet_id,
+                  'cidr': cidr,
+                  'gateway_ip': mock.sentinel.gateway_ip,
+                  'ipv6_ra_mode': mock.sentinel.ipv6_ra_mode,
+                  'subnetpool_id': mock.sentinel.subnetpool_id}
+        get_subnets_by_network.return_value = {'net_id': [subnet]}
+
+        ports = [{'network_id': 'net_id',
+                  'id': 'port_id',
+                  'fixed_ips': [{'subnet_id': mock.sentinel.subnet_id}]}]
+        self.db._populate_subnets_for_ports(mock.sentinel.context, ports)
+        self.assertEqual([{'extra_subnets': [],
+                           'fixed_ips': [{'subnet_id': mock.sentinel.subnet_id,
+                                          'prefixlen': 64}],
+                           'id': 'port_id',
+                           'network_id': 'net_id',
+                           'subnets': [subnet]}], ports)