Merge "Handle IPv6 addresses for LB IP port mappings" into stable/2024.1

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)

ovsdbapp/tests/functional/schema/ovn_northbound/test_impl_idl.py

@@ -1917,10 +1917,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):
+    def _test_lb_add_del_ip_port_mapping(self, col, input, expected):
-        endpoint_ip = '172.31.0.4'
+        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, {})
@@ -1929,18 +1930,30 @@ class TestLoadBalancerOps(OvnNorthboundTest):
                                         endpoint_ip,
                                         port_name,
                                         source_ip).execute(check_error=True)
-        self.assertEqual(lb.ip_port_mappings[endpoint_ip],
+        self.assertEqual(lb.ip_port_mappings[expected_endpoint_ip],
-                         '%s:%s' % (port_name, source_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 = {