← Back to branch summary

~ubuntu-branches/ubuntu/utopic/neutron/utopic-updates

« back to all changes in this revision

Viewing changes to neutron/tests/unit/ml2/test_security_group.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, Corey Bryant
  • Date: 2014-12-08 13:12:05 UTC
  • mfrom: (1.1.19)
  • Revision ID: package-import@ubuntu.com-20141208131205-ukvn1x0fka514qcx
Tags: 1:2014.2.1-0ubuntu1
https://launchpad.net/bugs/1398952
[ Corey Bryant ]
* Resynchronize with stable/juno (330ca81) (LP: #1398952):
  - [209160b] Fix hostname validation for nameservers
  - [d7ee2ff] Alter execute_alembic_command() to not assume all commands
  - [3e1d3a7] Prevent an iteration through ports on IPv6 slaac
  - [93f2536] Fix handling of CIDR in allowed address pairs
  - [8b5c08b] Fix metadata proxy start problem for v6-v4 network
  - [c00b4b3] Convert all incoming protocol numbers to string
  - [56b0029] linuxbridge-agent: make vxlan unicast check more efficent
  - [63b4337] Subnet delete for IPv6 SLAAC should not require prior port disassoc
  - [d09ba29] Fix context.elevated
  - [467a449] Drop and recreate FK if adding new PK to routerl3bindings
  - [87311b0] BSN: include missing data in floating IP call
  - [ad6fefc] Fix hostname regex pattern
  - [ba647ca] BSN: Set inconsistency record on delete failure
  - [c7954ab] fix event_send for re-assign floating ip
  - [b0d3e74] Fix KeyError in dhcp_rpc when plugin.port_update raise exception
  - [4e7b97e] rootwrap config files reference deleted quantum binaries
  - [d343288] Remove openvswitch core plugin entry point
  - [033e141] Fix L3 HA network creation to allow user to create router
  - [fe7c233] Stop ignoring 400 errors returned by ODL
  - [6efe68b] Allow to add router interface to IPv6 SLAAC network
  - [0877891] Set vif_details to reflect enable_security_group
  - [ee9c51d] Fix sleep function call
  - [96f9e7b] Batch ports from security groups RPC handler
  - [0be0dab] Big Switch: Fix SSL version on get_server_cert
  - [8575fe9] Create DHCP port for IPv6 subnet
  - [e51d647] NSX: allow multiple networks with same vlan on different phy_net
  - [a575cb7] _update_router_db: don't hold open transactions
  - [dd57281] Big Switch: Switch to TLSv1 in server manager
  - [560f8f8] Cisco N1kv: Fix update network profile for add tenants
  - [c5ae9dd] Only fetch port_id from SG binding table
  - [25c8280] Optimize query in _select_dhcp_ips_for_network_ids
  - [c14b58b] Improve performance of security group DB query
  - [f9f694f] Reduce security group db calls to neutron server
  - [72ef7b3] Cisco N1kv: Remove vmnetwork delete REST call on last port delete
  - [11f480e] Big Switch: Don't clear hash before sync
  - [4a84be5] Race for l2pop when ports go up/down on same host
  - [330ca81] Use EUI64 for IPv6 SLAAC when subnet is specified

Show diffs side-by-side

added added

removed removed

Lines of Context:
neutron/tests/unit/ml2/test_security_group.py
14
14
#    License for the specific language governing permissions and limitations
15
15
#    under the License.
16
16
 
 
17
import contextlib
 
18
import math
17
19
import mock
18
20
 
19
21
from neutron.api.v2 import attributes
 
22
from neutron.common import constants as const
20
23
from neutron.extensions import securitygroup as ext_sg
21
24
from neutron import manager
 
25
from neutron.tests.unit import test_api_v2
22
26
from neutron.tests.unit import test_extension_security_group as test_sg
23
27
from neutron.tests.unit import test_security_groups_rpc as test_sg_rpc
24
28
 
55
59
        plugin = manager.NeutronManager.get_plugin()
56
60
        plugin.start_rpc_listeners()
57
61
 
58
 
    def test_security_group_get_port_from_device(self):
 
62
    def _make_port_with_new_sec_group(self, net_id):
 
63
        sg = self._make_security_group(self.fmt, 'name', 'desc')
 
64
        port = self._make_port(
 
65
            self.fmt, net_id, security_groups=[sg['security_group']['id']])
 
66
        return port['port']
 
67
 
 
68
    def test_security_group_get_ports_from_devices(self):
59
69
        with self.network() as n:
60
70
            with self.subnet(n):
61
 
                with self.security_group() as sg:
62
 
                    security_group_id = sg['security_group']['id']
63
 
                    res = self._create_port(self.fmt, n['network']['id'])
64
 
                    port = self.deserialize(self.fmt, res)
65
 
                    fixed_ips = port['port']['fixed_ips']
66
 
                    data = {'port': {'fixed_ips': fixed_ips,
67
 
                                     'name': port['port']['name'],
68
 
                                     ext_sg.SECURITYGROUPS:
69
 
                                     [security_group_id]}}
70
 
 
71
 
                    req = self.new_update_request('ports', data,
72
 
                                                  port['port']['id'])
73
 
                    res = self.deserialize(self.fmt,
74
 
                                           req.get_response(self.api))
75
 
                    port_id = res['port']['id']
76
 
                    plugin = manager.NeutronManager.get_plugin()
77
 
                    port_dict = plugin.get_port_from_device(port_id)
78
 
                    self.assertEqual(port_id, port_dict['id'])
79
 
                    self.assertEqual([security_group_id],
 
71
                port1 = self._make_port_with_new_sec_group(n['network']['id'])
 
72
                port2 = self._make_port_with_new_sec_group(n['network']['id'])
 
73
                plugin = manager.NeutronManager.get_plugin()
 
74
                # should match full ID and starting chars
 
75
                ports = plugin.get_ports_from_devices(
 
76
                    [port1['id'], port2['id'][0:8]])
 
77
                self.assertEqual(2, len(ports))
 
78
                for port_dict in ports:
 
79
                    p = port1 if port1['id'] == port_dict['id'] else port2
 
80
                    self.assertEqual(p['id'], port_dict['id'])
 
81
                    self.assertEqual(p['security_groups'],
80
82
                                     port_dict[ext_sg.SECURITYGROUPS])
81
83
                    self.assertEqual([], port_dict['security_group_rules'])
82
 
                    self.assertEqual([fixed_ips[0]['ip_address']],
 
84
                    self.assertEqual([p['fixed_ips'][0]['ip_address']],
83
85
                                     port_dict['fixed_ips'])
84
 
                    self._delete('ports', port_id)
85
 
 
86
 
    def test_security_group_get_port_from_device_with_no_port(self):
87
 
        plugin = manager.NeutronManager.get_plugin()
88
 
        port_dict = plugin.get_port_from_device('bad_device_id')
89
 
        self.assertIsNone(port_dict)
 
86
                    self._delete('ports', p['id'])
 
87
 
 
88
    def test_security_group_get_ports_from_devices_with_bad_id(self):
 
89
        plugin = manager.NeutronManager.get_plugin()
 
90
        ports = plugin.get_ports_from_devices(['bad_device_id'])
 
91
        self.assertFalse(ports)
 
92
 
 
93
    def test_security_group_no_db_calls_with_no_ports(self):
 
94
        plugin = manager.NeutronManager.get_plugin()
 
95
        with mock.patch(
 
96
            'neutron.plugins.ml2.db.get_sg_ids_grouped_by_port'
 
97
        ) as get_mock:
 
98
            self.assertFalse(plugin.get_ports_from_devices([]))
 
99
            self.assertFalse(get_mock.called)
 
100
 
 
101
    def test_large_port_count_broken_into_parts(self):
 
102
        plugin = manager.NeutronManager.get_plugin()
 
103
        max_ports_per_query = 5
 
104
        ports_to_query = 73
 
105
        for max_ports_per_query in (1, 2, 5, 7, 9, 31):
 
106
            with contextlib.nested(
 
107
                mock.patch('neutron.plugins.ml2.db.MAX_PORTS_PER_QUERY',
 
108
                           new=max_ports_per_query),
 
109
                mock.patch('neutron.plugins.ml2.db.get_sg_ids_grouped_by_port',
 
110
                           return_value={}),
 
111
            ) as (max_mock, get_mock):
 
112
                plugin.get_ports_from_devices(
 
113
                    ['%s%s' % (const.TAP_DEVICE_PREFIX, i)
 
114
                     for i in range(ports_to_query)])
 
115
                all_call_args = map(lambda x: x[1][0], get_mock.mock_calls)
 
116
                last_call_args = all_call_args.pop()
 
117
                # all but last should be getting MAX_PORTS_PER_QUERY ports
 
118
                self.assertTrue(
 
119
                    all(map(lambda x: len(x) == max_ports_per_query,
 
120
                            all_call_args))
 
121
                )
 
122
                remaining = ports_to_query % max_ports_per_query
 
123
                if remaining:
 
124
                    self.assertEqual(remaining, len(last_call_args))
 
125
                # should be broken into ceil(total/MAX_PORTS_PER_QUERY) calls
 
126
                self.assertEqual(
 
127
                    math.ceil(ports_to_query / float(max_ports_per_query)),
 
128
                    get_mock.call_count
 
129
                )
 
130
 
 
131
    def test_full_uuids_skip_port_id_lookup(self):
 
132
        plugin = manager.NeutronManager.get_plugin()
 
133
        # when full UUIDs are provided, the _or statement should only
 
134
        # have one matching 'IN' critiera for all of the IDs
 
135
        with contextlib.nested(
 
136
            mock.patch('neutron.plugins.ml2.db.or_'),
 
137
            mock.patch('neutron.plugins.ml2.db.db_api.get_session')
 
138
        ) as (or_mock, sess_mock):
 
139
            fmock = sess_mock.query.return_value.outerjoin.return_value.filter
 
140
            # return no ports to exit the method early since we are mocking
 
141
            # the query
 
142
            fmock.return_value.all.return_value = []
 
143
            plugin.get_ports_from_devices([test_api_v2._uuid(),
 
144
                                           test_api_v2._uuid()])
 
145
            # the or_ function should only have one argument
 
146
            or_mock.assert_called_once_with(mock.ANY)
90
147
 
91
148
 
92
149
class TestMl2SecurityGroupsXML(TestMl2SecurityGroups):