37
37
config.cfg.CONF.set_override('mechanism_drivers',
38
38
['logger', 'test'],
40
config.cfg.CONF.set_override('network_vlan_ranges',
41
['physnet1:1000:1099'],
42
group='ml2_type_vlan')
40
43
super(PortBindingTestCase, self).setUp(PLUGIN_NAME)
41
44
self.port_create_status = 'DOWN'
42
45
self.plugin = manager.NeutronManager.get_plugin()
55
58
self.assertEqual(port_status, 'DOWN')
57
60
def _test_port_binding(self, host, vif_type, has_port_filter, bound,
61
status=None, network_type='local'):
59
62
mac_address = 'aa:aa:aa:aa:aa:aa'
60
63
host_arg = {portbindings.HOST_ID: host,
61
64
'mac_address': mac_address}
68
71
details = self.plugin.endpoints[0].get_device_details(
69
72
neutron_context, agent_id="theAgentId", device=port_id)
71
self.assertEqual(details['network_type'], 'local')
74
self.assertEqual(details['network_type'], network_type)
72
75
self.assertEqual(mac_address, details['mac_address'])
74
77
self.assertNotIn('network_type', details)
108
111
self.assertIsNone(
109
112
self.plugin.get_bound_port_context(ctx, port['port']['id']))
114
def test_hierarchical_binding(self):
115
self._test_port_binding("host-hierarchical",
116
portbindings.VIF_TYPE_OVS,
117
False, True, network_type='vlan')
119
def test_get_bound_port_context_cache_hit(self):
120
ctx = context.get_admin_context()
121
with self.port(name='name') as port:
122
cached_network_id = port['port']['network_id']
123
some_network = {'id': cached_network_id}
124
cached_networks = {cached_network_id: some_network}
125
self.plugin.get_network = mock.Mock(return_value=some_network)
126
self.plugin.get_bound_port_context(ctx, port['port']['id'],
127
cached_networks=cached_networks)
128
self.assertFalse(self.plugin.get_network.called)
130
def test_get_bound_port_context_cache_miss(self):
131
ctx = context.get_admin_context()
132
with self.port(name='name') as port:
133
some_network = {'id': u'2ac23560-7638-44e2-9875-c1888b02af72'}
134
self.plugin.get_network = mock.Mock(return_value=some_network)
135
self.plugin.get_bound_port_context(ctx, port['port']['id'],
137
self.assertEqual(1, self.plugin.get_network.call_count)
111
139
def _test_update_port_binding(self, host, new_host=None):
112
140
with mock.patch.object(self.plugin,
113
141
'_notify_port_updated') as notify_mock: