[test]: Add interface IPv6 add/remove test case (#677)

Add one IPv6 address to a port based router interface
and check the router interface is created and the corresponding
routes are also created.

Signed-off-by: Shu0T1an ChenG <[email protected]>

cfgmgr/intfmgr.cpp

@@ -110,11 +110,13 @@ void IntfMgr::doTask(Consumer &consumer)
             }
             setIntfIp(alias, "add", ip_prefix.to_string(), ip_prefix.isV4());
             m_stateIntfTable.hset(keys[0] + state_db_key_delimiter + keys[1], "state", "ok");
+            SWSS_LOG_NOTICE("Add %s to %s", ip_prefix.to_string().c_str(), alias.c_str());
         }
         else if (op == DEL_COMMAND)
         {
             setIntfIp(alias, "del", ip_prefix.to_string(), ip_prefix.isV4());
             m_stateIntfTable.del(keys[0] + state_db_key_delimiter + keys[1]);
+            SWSS_LOG_NOTICE("Remove %s from %s", ip_prefix.to_string().c_str(), alias.c_str());
         }
         else
         {

tests/test_interface.py

@@ -3,17 +3,23 @@
 import time
 import json
 
-class TestRouterInterfaceIpv4(object):
+class TestRouterInterface(object):
     def setup_db(self, dvs):
         self.pdb = swsscommon.DBConnector(0, dvs.redis_sock, 0)
         self.adb = swsscommon.DBConnector(1, dvs.redis_sock, 0)
         self.cdb = swsscommon.DBConnector(4, dvs.redis_sock, 0)
 
+    def set_admin_status(self, interface, status):
+        tbl = swsscommon.Table(self.cdb, "PORT")
+        fvs = swsscommon.FieldValuePairs([("admin_status", status)])
+        tbl.set(interface, fvs)
+        time.sleep(1)
+
     def add_ip_address(self, interface, ip):
         tbl = swsscommon.Table(self.cdb, "INTERFACE")
         fvs = swsscommon.FieldValuePairs([("NULL", "NULL")])
         tbl.set(interface + "|" + ip, fvs)
-        time.sleep(1)
+        time.sleep(2) # IPv6 netlink message needs longer time
 
     def remove_ip_address(self, interface, ip):
         tbl = swsscommon.Table(self.cdb, "INTERFACE")
@@ -26,6 +32,83 @@ def set_mtu(self, interface, mtu):
         tbl.set(interface, fvs)
         time.sleep(1)
 
+    def test_InterfaceAddRemoveIpv6Address(self, dvs, testlog):
+        self.setup_db(dvs)
+
+        # bring up interface
+        # NOTE: For IPv6, only when the interface is up will the netlink message
+        # get generated.
+        self.set_admin_status("Ethernet8", "up")
+
+        # assign IP to interface
+        self.add_ip_address("Ethernet8", "fc00::1/126")
+
+        # check application database
+        tbl = swsscommon.Table(self.pdb, "INTF_TABLE:Ethernet8")
+        intf_entries = tbl.getKeys()
+        assert len(intf_entries) == 1
+        assert intf_entries[0] == "fc00::1/126"
+
+        (status, fvs) = tbl.get(tbl.getKeys()[0])
+        assert status == True
+        assert len(fvs) == 2
+        for fv in fvs:
+            if fv[0] == "scope":
+                assert fv[1] == "global"
+            elif fv[0] == "family":
+                assert fv[1] == "IPv6"
+            else:
+                assert False
+
+        # check ASIC router interface database
+        tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_ROUTER_INTERFACE")
+        intf_entries = tbl.getKeys()
+        # one loopback router interface one port based router interface
+        assert len(intf_entries) == 2
+
+        for key in intf_entries:
+            (status, fvs) = tbl.get(key)
+            assert status == True
+            # a port based router interface has five field/value tuples
+            if len(fvs) == 5:
+                for fv in fvs:
+                    if fv[0] == "SAI_ROUTER_INTERFACE_ATTR_TYPE":
+                        assert fv[1] == "SAI_ROUTER_INTERFACE_TYPE_PORT"
+                    # the default MTU without any configuration is 9100
+                    if fv[0] == "SAI_ROUTER_INTERFACE_ATTR_MTU":
+                        assert fv[1] == "9100"
+
+        # check ASIC route database
+        tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_ROUTE_ENTRY")
+        for key in tbl.getKeys():
+            route = json.loads(key)
+            if route["dest"] == "fc00::/126":
+                subnet_found = True
+            if route["dest"] == "fc00::1/128":
+                ip2me_found = True
+
+        assert subnet_found and ip2me_found
+
+        # remove IP from interface
+        self.remove_ip_address("Ethernet8", "fc00::1/126")
+
+        # bring down interface
+        self.set_admin_status("Ethernet8", "down")
+
+        # check application database
+        tbl = swsscommon.Table(self.pdb, "INTF_TABLE:Ethernet8")
+        intf_entries = tbl.getKeys()
+        assert len(intf_entries) == 0
+
+        # check ASIC database
+        tbl = swsscommon.Table(self.adb, "ASIC_STATE:SAI_OBJECT_TYPE_ROUTE_ENTRY")
+        for key in tbl.getKeys():
+            route = json.loads(key)
+            if route["dest"] == "fc00::/126":
+                assert False
+            if route["dest"] == "fc00::1/128":
+                assert False
+
     def test_InterfaceAddRemoveIpv4Address(self, dvs, testlog):
         self.setup_db(dvs)