diff --git a/ovsdbapp/schema/ovn_northbound/commands.py b/ovsdbapp/schema/ovn_northbound/commands.py index a738454c..21cd9402 100644 --- a/ovsdbapp/schema/ovn_northbound/commands.py +++ b/ovsdbapp/schema/ovn_northbound/commands.py @@ -1670,15 +1670,22 @@ class LbDelHealthCheckCommand(cmd.BaseCommand): raise RuntimeError(msg) -class LbAddIpPortMappingCommand(cmd.BaseCommand): +class LbIpPortMappingCommand(cmd.BaseCommand): + @staticmethod + def normalize_ip(ip_str): + ip = netaddr.IPAddress(ip_str) + return f"[{ip}]" if ip.version == 6 else str(ip) + + +class LbAddIpPortMappingCommand(LbIpPortMappingCommand): table = 'Load_Balancer' def __init__(self, api, lb, endpoint_ip, port_name, source_ip): super().__init__(api) self.lb = lb - self.endpoint_ip = str(netaddr.IPAddress(endpoint_ip)) + self.endpoint_ip = self.normalize_ip(endpoint_ip) self.port_name = port_name - self.source_ip = str(netaddr.IPAddress(source_ip)) + self.source_ip = self.normalize_ip(source_ip) def run_idl(self, txn): lb = self.api.lookup(self.table, self.lb) @@ -1686,13 +1693,13 @@ class LbAddIpPortMappingCommand(cmd.BaseCommand): '%s:%s' % (self.port_name, self.source_ip)) -class LbDelIpPortMappingCommand(cmd.BaseCommand): +class LbDelIpPortMappingCommand(LbIpPortMappingCommand): table = 'Load_Balancer' def __init__(self, api, lb, endpoint_ip): super().__init__(api) self.lb = lb - self.endpoint_ip = str(netaddr.IPAddress(endpoint_ip)) + self.endpoint_ip = self.normalize_ip(endpoint_ip) def run_idl(self, txn): lb = self.api.lookup(self.table, self.lb) diff --git a/ovsdbapp/tests/functional/schema/ovn_northbound/test_impl_idl.py b/ovsdbapp/tests/functional/schema/ovn_northbound/test_impl_idl.py index 3699d79e..3e949d32 100644 --- a/ovsdbapp/tests/functional/schema/ovn_northbound/test_impl_idl.py +++ b/ovsdbapp/tests/functional/schema/ovn_northbound/test_impl_idl.py @@ -1907,10 +1907,11 @@ class TestLoadBalancerOps(OvnNorthboundTest): self.api.lb_del_health_check(lb.name, uuid.uuid4(), if_exists=True).execute(check_error=True) - def _test_lb_add_del_ip_port_mapping(self, col): - endpoint_ip = '172.31.0.4' + def _test_lb_add_del_ip_port_mapping(self, col, input, expected): + endpoint_ip, source_ip = input + expected_endpoint_ip, expected_source_ip = expected port_name = 'sw1-p1' - source_ip = '172.31.0.6' + lb = self._lb_add(utils.get_rand_device_name(), '192.0.0.1', ['10.0.0.1']) self.assertEqual(lb.ip_port_mappings, {}) @@ -1919,18 +1920,30 @@ class TestLoadBalancerOps(OvnNorthboundTest): endpoint_ip, port_name, source_ip).execute(check_error=True) - self.assertEqual(lb.ip_port_mappings[endpoint_ip], - '%s:%s' % (port_name, source_ip)) + self.assertEqual(lb.ip_port_mappings[expected_endpoint_ip], + '%s:%s' % (port_name, expected_source_ip)) self.api.lb_del_ip_port_mapping(val, endpoint_ip).execute(check_error=True) self.assertEqual(lb.ip_port_mappings, {}) def test_lb_add_del_ip_port_mapping_uuid(self): - self._test_lb_add_del_ip_port_mapping('uuid') + input = ('172.31.0.3', '172.31.0.6') + self._test_lb_add_del_ip_port_mapping('uuid', input, input) + + def test_lb_add_del_ip_port_mapping_uuid_v6(self): + input = ('2001:db8::1', '2001:db8::2') + expected = (f"[{input[0]}]", f"[{input[1]}]") + self._test_lb_add_del_ip_port_mapping('uuid', input, expected) def test_lb_add_del_ip_port_mapping_name(self): - self._test_lb_add_del_ip_port_mapping('name') + input = ('172.31.0.3', '172.31.0.6') + self._test_lb_add_del_ip_port_mapping('name', input, input) + + def test_lb_add_del_ip_port_mapping_name_v6(self): + input = ('2001:db8::1', '2001:db8::2') + expected = (f"[{input[0]}]", f"[{input[1]}]") + self._test_lb_add_del_ip_port_mapping('name', input, expected) def test_hc_get_set_options(self): hc_options = {