openstack
/
neutron
Code Issues Proposed changes

Merge "ovn: Add functional tests for get_network_port_bindings_by_ip"

This commit is contained in:
Zuul 2021-04-28 15:52:52 +00:00 committed by Gerrit Code Review
parent 52b14863ef 1a3fa2206d
commit 5a3afdb808
1 changed files with 19 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_impl_idl.py
View File

@ -130,7 +130,7 @@ class TestSbApi(BaseOvnIdlTest):
check_error=True) check_error=True)
self.assertEqual(nets, self.api.get_chassis_metadata_networks(name)) self.assertEqual(nets, self.api.get_chassis_metadata_networks(name))
def test_get_network_port_bindings_by_ip(self): def _create_bound_port_with_ip(self):
chassis, switch, port, binding = self._add_switch_port( chassis, switch, port, binding = self._add_switch_port(
self.data['chassis'][0]['name']) self.data['chassis'][0]['name'])
mac = 'de:ad:be:ef:4d:ad' mac = 'de:ad:be:ef:4d:ad'
@ -143,10 +143,28 @@ class TestSbApi(BaseOvnIdlTest):
port.name, [mac_ip]).execute(check_error=True) port.name, [mac_ip]).execute(check_error=True)
self.assertTrue(pb_update_event.wait()) self.assertTrue(pb_update_event.wait())
self.api.lsp_bind(port.name, chassis.name).execute(check_error=True) self.api.lsp_bind(port.name, chassis.name).execute(check_error=True)
return binding, ipaddr, switch
def test_get_network_port_bindings_by_ip(self):
binding, ipaddr, _ = self._create_bound_port_with_ip()
result = self.api.get_network_port_bindings_by_ip( result = self.api.get_network_port_bindings_by_ip(
str(binding.datapath.uuid), ipaddr) str(binding.datapath.uuid), ipaddr)
self.assertIn(binding, result) self.assertIn(binding, result)
def test_get_network_port_bindings_by_ip_with_unbound_port(self):
binding, ipaddr, switch = self._create_bound_port_with_ip()
unbound_port_name = utils.get_rand_device_name(prefix="port")
mac_ip = "de:ad:be:ef:4d:ab %s" % ipaddr
with self.nbapi.transaction(check_error=True) as txn:
txn.add(
self.nbapi.lsp_add(switch.name, unbound_port_name, type=type))
txn.add(self.nbapi.lsp_set_addresses(unbound_port_name, [mac_ip]))
result = self.api.get_network_port_bindings_by_ip(
str(binding.datapath.uuid), ipaddr)
self.assertIn(binding, result)
self.assertEqual(1, len(result))
def test_get_ports_on_chassis(self): def test_get_ports_on_chassis(self):
chassis, switch, port, binding = self._add_switch_port( chassis, switch, port, binding = self._add_switch_port(
self.data['chassis'][0]['name']) self.data['chassis'][0]['name'])