diff --git a/tests/test_nat.py b/tests/test_nat.py index 1223ceaec2..1c06f6a8bc 100644 --- a/tests/test_nat.py +++ b/tests/test_nat.py @@ -1,59 +1,43 @@ import time -import re -import json -import pytest -import pdb -import os - -from swsscommon import swsscommon class TestNat(object): - def setup_db(self, dvs): - self.appdb = swsscommon.DBConnector(0, dvs.redis_sock, 0) - self.asicdb = swsscommon.DBConnector(1, dvs.redis_sock, 0) - self.configdb = swsscommon.DBConnector(4, dvs.redis_sock, 0) + def setup_db(self, app_db, asic_db, config_db): + self.app_db = app_db + self.asic_db = asic_db + self.config_db = config_db def set_interfaces(self, dvs): - intf_tbl = swsscommon.Table(self.configdb, "INTERFACE") - fvs = swsscommon.FieldValuePairs([("NULL","NULL")]) - intf_tbl.set("Ethernet0|67.66.65.1/24", fvs) - intf_tbl.set("Ethernet4|18.18.18.1/24", fvs) - intf_tbl.set("Ethernet0", fvs) - intf_tbl.set("Ethernet4", fvs) + fvs = {"NULL": "NULL"} + self.config_db.create_entry("INTERFACE", "Ethernet0|67.66.65.1/24", fvs) + self.config_db.create_entry("INTERFACE", "Ethernet4|18.18.18.1/24", fvs) + self.config_db.create_entry("INTERFACE", "Ethernet0", fvs) + self.config_db.create_entry("INTERFACE", "Ethernet4", fvs) dvs.runcmd("ifconfig Ethernet0 up") dvs.runcmd("ifconfig Ethernet4 up") - dvs.servers[0].runcmd("ip link set down dev eth0") == 0 - dvs.servers[0].runcmd("ip link set up dev eth0") == 0 + dvs.servers[0].runcmd("ip link set down dev eth0") + dvs.servers[0].runcmd("ip link set up dev eth0") dvs.servers[0].runcmd("ifconfig eth0 67.66.65.2/24") dvs.servers[0].runcmd("ip route add default via 67.66.65.1") - dvs.servers[1].runcmd("ip link set down dev eth0") == 0 - dvs.servers[1].runcmd("ip link set up dev eth0") == 0 + dvs.servers[1].runcmd("ip link set down dev eth0") + dvs.servers[1].runcmd("ip link set up dev eth0") dvs.servers[1].runcmd("ifconfig eth0 18.18.18.2/24") dvs.servers[1].runcmd("ip route add default via 18.18.18.1") - ps = swsscommon.ProducerStateTable(self.appdb, "ROUTE_TABLE") - fvs = swsscommon.FieldValuePairs([("nexthop","18.18.18.2"), \ - ("ifname", "Ethernet0")]) - - pubsub = dvs.SubscribeAsicDbObject("SAI_OBJECT_TYPE_ROUTE_ENTRY") - dvs.runcmd("config nat add interface Ethernet0 -nat_zone 1") time.sleep(1) def clear_interfaces(self, dvs): dvs.servers[0].runcmd("ifconfig eth0 0.0.0.0") - dvs.servers[1].runcmd("ifconfig eth0 0.0.0.0") - # dvs.servers[1].runcmd("ip route del default") time.sleep(1) - def test_NatGlobalTable(self, dvs, testlog): + def test_NatGlobalTable(self, app_db, asic_db, config_db, dvs, testlog): # initialize - self.setup_db(dvs) + self.setup_db(app_db, asic_db, config_db) # enable NAT feature dvs.runcmd("config nat feature enable") @@ -62,37 +46,29 @@ def test_NatGlobalTable(self, dvs, testlog): dvs.runcmd("config nat set tcp-timeout 900") # check NAT global values in appdb - tbl = swsscommon.Table(self.appdb, "NAT_GLOBAL_TABLE") - values = tbl.getKeys() + self.app_db.wait_for_n_keys("NAT_GLOBAL_TABLE", 1) - assert len(values) == 1 + fvs = self.app_db.wait_for_entry("NAT_GLOBAL_TABLE", "Values") - (status, fvs) = tbl.get("Values") + assert fvs == {"admin_mode": "enabled", "nat_timeout": "450", "nat_udp_timeout": "360", "nat_tcp_timeout": "900"} - assert fvs==(('admin_mode', 'enabled'), ('nat_timeout', '450'), ('nat_udp_timeout', '360'), ('nat_tcp_timeout', '900')) - - def test_NatInterfaceZone(self, dvs, testlog): + def test_NatInterfaceZone(self, app_db, asic_db, config_db, dvs, testlog): # initialize - self.setup_db(dvs) + self.setup_db(app_db, asic_db, config_db) self.set_interfaces(dvs) # check NAT zone is set for interface in app db - tbl = swsscommon.Table(self.appdb, "INTF_TABLE") - keys = tbl.getKeys() - - (status, fvs) = tbl.get("Ethernet0") - assert status == True - assert len(fvs) > 0 + fvs = self.app_db.wait_for_entry("INTF_TABLE", "Ethernet0") zone = False - for fv in fvs: - if fv[0] == 'nat_zone' and fv[1] == '1': - zone = True - break - assert zone == True + for f, v in fvs.items(): + if f == "nat_zone" and v == '1': + zone = True + break + assert zone - def test_AddNatStaticEntry(self, dvs, testlog): + def test_AddNatStaticEntry(self, app_db, asic_db, config_db, dvs, testlog): # initialize - self.setup_db(dvs) + self.setup_db(app_db, asic_db, config_db) # get neighbor and arp entry dvs.servers[0].runcmd("ping -c 1 18.18.18.2") @@ -101,60 +77,48 @@ def test_AddNatStaticEntry(self, dvs, testlog): dvs.runcmd("config nat add static basic 67.66.65.1 18.18.18.2") # check the entry in the config db - tbl = swsscommon.Table(self.configdb, "STATIC_NAT") - entry = tbl.getKeys() - assert len(entry) == 1 + self.config_db.wait_for_n_keys("STATIC_NAT", 1) - (status, fvs) = tbl.get("67.66.65.1") + fvs = self.config_db.wait_for_entry("STATIC_NAT", "67.66.65.1") - assert fvs==(('local_ip', '18.18.18.2'),) + assert fvs == {"local_ip": "18.18.18.2"} # check the entry in app db - tbl = swsscommon.Table(self.appdb, "NAT_TABLE") - entry = tbl.getKeys() - assert len(entry) == 2 + self.app_db.wait_for_n_keys("NAT_TABLE", 2) - (status, fvs) = tbl.get("67.66.65.1") + fvs = self.app_db.wait_for_entry("NAT_TABLE", "67.66.65.1") - assert fvs== (('translated_ip', '18.18.18.2'), ('nat_type', 'dnat'), ('entry_type', 'static')) + assert fvs == { + "translated_ip": "18.18.18.2", + "nat_type": "dnat", + "entry_type": "static" + } #check the entry in asic db - tbl = swsscommon.Table(self.asicdb, "ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY") - keys = tbl.getKeys() - assert len(keys) == 2 + keys = self.asic_db.wait_for_n_keys("ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY", 2) for key in keys: - if (key.find("dst_ip:67.66.65.1")) or (key.find("src_ip:18.18.18.2")): - assert True - else: - assert False + assert "\"dst_ip\":\"67.66.65.1\"" in key or "\"src_ip\":\"18.18.18.2\"" in key - def test_DelNatStaticEntry(self, dvs, testlog): + def test_DelNatStaticEntry(self, app_db, asic_db, config_db, dvs, testlog): # initialize - self.setup_db(dvs) + self.setup_db(app_db, asic_db, config_db) # delete a static nat entry dvs.runcmd("config nat remove static basic 67.66.65.1 18.18.18.2") # check the entry is no there in the config db - tbl = swsscommon.Table(self.configdb, "STATIC_NAT") - entry = tbl.getKeys() - assert entry == () + self.config_db.wait_for_n_keys("STATIC_NAT", 0) # check the entry is not there in app db - tbl = swsscommon.Table(self.appdb, "NAT_TABLE") - entry = tbl.getKeys() - assert entry == () + self.app_db.wait_for_n_keys("NAT_TABLE", 0) #check the entry is not there in asic db - tbl = swsscommon.Table(self.asicdb, "ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY") - key = tbl.getKeys() - assert key == () + keys = self.asic_db.wait_for_n_keys("ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY", 0) - @pytest.mark.xfail(reason="vs image issue: Azure/sonic-sairedis#575") - def test_AddNaPtStaticEntry(self, dvs, testlog): + def test_AddNaPtStaticEntry(self, app_db, asic_db, config_db, dvs, testlog): # initialize - self.setup_db(dvs) + self.setup_db(app_db, asic_db, config_db) # get neighbor and arp entry dvs.servers[0].runcmd("ping -c 1 18.18.18.2") @@ -163,63 +127,48 @@ def test_AddNaPtStaticEntry(self, dvs, testlog): dvs.runcmd("config nat add static udp 67.66.65.1 670 18.18.18.2 180") # check the entry in the config db - tbl = swsscommon.Table(self.configdb, "STATIC_NAPT") - entry = tbl.getKeys() - assert len(entry) == 1 + self.config_db.wait_for_n_keys("STATIC_NAPT", 1) - (status, fvs) = tbl.get("67.66.65.1|UDP|670") + fvs = self.config_db.wait_for_entry("STATIC_NAPT", "67.66.65.1|UDP|670") - assert fvs==(('local_ip', '18.18.18.2'),('local_port', '180')) + assert fvs == {"local_ip": "18.18.18.2", "local_port": "180"} # check the entry in app db - tbl = swsscommon.Table(self.appdb, "NAPT_TABLE:UDP") - entry = tbl.getKeys() - assert len(entry) == 2 + self.app_db.wait_for_n_keys("NAPT_TABLE:UDP", 2) - (status, fvs) = tbl.get("67.66.65.1:670") + fvs = self.app_db.wait_for_entry("NAPT_TABLE:UDP", "67.66.65.1:670") - assert fvs== (('translated_ip', '18.18.18.2'), ('translated_l4_port', '180'), ('nat_type', 'dnat'), ('entry_type', 'static')) + assert fvs == {"translated_ip": "18.18.18.2", "translated_l4_port": "180", "nat_type": "dnat", "entry_type": "static"} #check the entry in asic db - tbl = swsscommon.Table(self.asicdb, "ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY") - keys = tbl.getKeys() - assert len(keys) == 2 - + keys = self.asic_db.wait_for_n_keys("ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY", 2) for key in keys: - if (key.find("dst_ip:67.66.65.1")) and (key.find("key.l4_dst_port:670")): - assert True - if (key.find("src_ip:18.18.18.2")) or (key.find("key.l4_src_port:180")): - assert True + if "\"dst_ip\":\"67.66.65.1\"" in key and "\"l4_dst_port\":\"670\"" in key: + assert True + elif "\"src_ip\":\"18.18.18.2\"" in key or "\"l4_src_port\":\"180\"" in key: + assert True else: - assert False + assert False - @pytest.mark.xfail(reason="vs image issue: Azure/sonic-sairedis#575") - def test_DelNaPtStaticEntry(self, dvs, testlog): + def test_DelNaPtStaticEntry(self, app_db, asic_db, config_db, dvs, testlog): # initialize - self.setup_db(dvs) + self.setup_db(app_db, asic_db, config_db) # delete a static nat entry dvs.runcmd("config nat remove static udp 67.66.65.1 670 18.18.18.2 180") # check the entry is no there in the config db - tbl = swsscommon.Table(self.configdb, "STATIC_NAPT") - entry = tbl.getKeys() - assert entry == () + self.config_db.wait_for_n_keys("STATIC_NAPT", 0) # check the entry is not there in app db - tbl = swsscommon.Table(self.appdb, "NAPT_TABLE") - entry = tbl.getKeys() - assert entry == () + self.app_db.wait_for_n_keys("NAPT_TABLE:UDP", 0) #check the entry is not there in asic db - tbl = swsscommon.Table(self.asicdb, "ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY") - key = tbl.getKeys() - assert key == () + self.asic_db.wait_for_n_keys("ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY", 0) - @pytest.mark.xfail(reason="vs image issue: Azure/sonic-sairedis#575") - def test_AddTwiceNatEntry(self, dvs, testlog): + def test_AddTwiceNatEntry(self, app_db, asic_db, config_db, dvs, testlog): # initialize - self.setup_db(dvs) + self.setup_db(app_db, asic_db, config_db) # get neighbor and arp entry dvs.servers[0].runcmd("ping -c 1 18.18.18.2") @@ -230,69 +179,48 @@ def test_AddTwiceNatEntry(self, dvs, testlog): dvs.runcmd("config nat add static basic 67.66.65.1 18.18.18.2 -nat_type dnat -twice_nat_id 9") # check the entry in the config db - tbl = swsscommon.Table(self.configdb, "STATIC_NAT") - entry = tbl.getKeys() - assert len(entry) == 2 - - (status, fvs) = tbl.get("67.66.65.1") + self.config_db.wait_for_n_keys("STATIC_NAT", 2) - assert fvs== (('nat_type', 'dnat'), ('twice_nat_id', '9'), ('local_ip', '18.18.18.2')) + fvs = self.config_db.wait_for_entry("STATIC_NAT", "67.66.65.1") + assert fvs == {"nat_type": "dnat", "twice_nat_id": "9", "local_ip": "18.18.18.2"} - (status, fvs) = tbl.get("67.66.65.2") - - assert fvs== (('nat_type', 'snat'), ('twice_nat_id', '9'), ('local_ip', '18.18.18.1')) + fvs = self.config_db.wait_for_entry("STATIC_NAT", "67.66.65.2") + assert fvs == {"nat_type": "snat", "twice_nat_id": "9", "local_ip": "18.18.18.1"} # check the entry in app db - tbl = swsscommon.Table(self.appdb, "NAT_TWICE_TABLE") - entry = tbl.getKeys() - assert len(entry) == 2 - - (status, fvs) = tbl.get("67.66.65.2:67.66.65.1") + self.app_db.wait_for_n_keys("NAT_TWICE_TABLE", 2) - assert fvs== (('translated_src_ip', '18.18.18.1'), ('translated_dst_ip', '18.18.18.2'), ('entry_type', 'static')) + fvs = self.app_db.wait_for_entry("NAT_TWICE_TABLE", "67.66.65.2:67.66.65.1") + assert fvs == {"translated_src_ip": "18.18.18.1", "translated_dst_ip": "18.18.18.2", "entry_type": "static"} - (status, fvs) = tbl.get("18.18.18.2:18.18.18.1") - - assert fvs== (('translated_src_ip', '67.66.65.1'), ('translated_dst_ip', '67.66.65.2'), ('entry_type', 'static')) + fvs = self.app_db.wait_for_entry("NAT_TWICE_TABLE", "18.18.18.2:18.18.18.1") + assert fvs == {"translated_src_ip": "67.66.65.1", "translated_dst_ip": "67.66.65.2", "entry_type": "static"} #check the entry in asic db - tbl = swsscommon.Table(self.asicdb, "ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY") - keys = tbl.getKeys() - assert len(keys) == 2 + keys = self.asic_db.wait_for_n_keys("ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY", 2) for key in keys: - if (key.find("dst_ip:18.18.18.1")) or (key.find("src_ip:18.18.18.2")): - assert True - else: - assert False + assert "\"dst_ip\":\"67.66.65.1\"" in key or "\"src_ip\":\"18.18.18.2\"" in key - @pytest.mark.xfail(reason="vs image issue: Azure/sonic-sairedis#575") - def test_DelTwiceNatStaticEntry(self, dvs, testlog): + def test_DelTwiceNatStaticEntry(self, app_db, asic_db, config_db, dvs, testlog): # initialize - self.setup_db(dvs) + self.setup_db(app_db, asic_db, config_db) # delete a static nat entry dvs.runcmd("config nat remove static basic 67.66.65.2 18.18.18.1") dvs.runcmd("config nat remove static basic 67.66.65.1 18.18.18.2") # check the entry is no there in the config db - tbl = swsscommon.Table(self.configdb, "STATIC_NAT") - entry = tbl.getKeys() - assert entry == () + self.config_db.wait_for_n_keys("STATIC_NAT", 0) # check the entry is not there in app db - tbl = swsscommon.Table(self.appdb, "NAT_TWICE_TABLE") - entry = tbl.getKeys() - assert entry == () + self.app_db.wait_for_n_keys("NAT_TWICE_TABLE", 0) #check the entry is not there in asic db - tbl = swsscommon.Table(self.asicdb, "ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY") - key = tbl.getKeys() - assert key == () + self.asic_db.wait_for_n_keys("ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY", 0) - @pytest.mark.xfail(reason="vs image issue: Azure/sonic-sairedis#575") - def test_AddTwiceNaPtEntry(self, dvs, testlog): + def test_AddTwiceNaPtEntry(self, app_db, asic_db, config_db, dvs, testlog): # initialize - self.setup_db(dvs) + self.setup_db(app_db, asic_db, config_db) # get neighbor and arp entry dvs.servers[0].runcmd("ping -c 1 18.18.18.2") @@ -303,65 +231,49 @@ def test_AddTwiceNaPtEntry(self, dvs, testlog): dvs.runcmd("config nat add static udp 67.66.65.1 660 18.18.18.2 182 -nat_type dnat -twice_nat_id 7") # check the entry in the config db - tbl = swsscommon.Table(self.configdb, "STATIC_NAPT") - entry = tbl.getKeys() - assert len(entry) == 2 - - (status, fvs) = tbl.get("67.66.65.1|UDP|660") + self.config_db.wait_for_n_keys("STATIC_NAPT", 2) - assert fvs== (('nat_type', 'dnat'), ('local_ip', '18.18.18.2'), ('twice_nat_id', '7'), ('local_port', '182')) - (status, fvs) = tbl.get("67.66.65.2|UDP|670") + fvs = self.config_db.wait_for_entry("STATIC_NAPT", "67.66.65.1|UDP|660") + assert fvs == {"nat_type": "dnat", "local_ip": "18.18.18.2", "twice_nat_id": "7", "local_port": "182"} - assert fvs== (('nat_type', 'snat'), ('local_ip', '18.18.18.1'),('twice_nat_id', '7'), ('local_port', '181')) + fvs = self.config_db.wait_for_entry("STATIC_NAPT", "67.66.65.2|UDP|670") + assert fvs == {"nat_type": "snat", "local_ip": "18.18.18.1", "twice_nat_id": "7", "local_port": "181"} # check the entry in app db - tbl = swsscommon.Table(self.appdb, "NAPT_TWICE_TABLE") - entry = tbl.getKeys() - assert len(entry) == 2 + self.app_db.wait_for_n_keys("NAPT_TWICE_TABLE", 2) - (status, fvs) = tbl.get("UDP:67.66.65.2:670:67.66.65.1:660") + fvs = self.app_db.wait_for_entry("NAPT_TWICE_TABLE", "UDP:67.66.65.2:670:67.66.65.1:660") + assert fvs == {"translated_src_ip": "18.18.18.1", "translated_src_l4_port": "181", "translated_dst_ip": "18.18.18.2", "translated_dst_l4_port": "182", "entry_type": "static"} - assert fvs== (('translated_src_ip', '18.18.18.1'), ('translated_src_l4_port', '181'), ('translated_dst_ip', '18.18.18.2'), ('translated_dst_l4_port', '182'), ('entry_type', 'static')) - - (status, fvs) = tbl.get("UDP:18.18.18.2:182:18.18.18.1:181") - - assert fvs== (('translated_src_ip', '67.66.65.1'), ('translated_src_l4_port', '660'),('translated_dst_ip', '67.66.65.2'),('translated_dst_l4_port', '670'), ('entry_type', 'static')) + fvs = self.app_db.wait_for_entry("NAPT_TWICE_TABLE", "UDP:18.18.18.2:182:18.18.18.1:181") + assert fvs == {"translated_src_ip": "67.66.65.1", "translated_src_l4_port": "660", "translated_dst_ip": "67.66.65.2", "translated_dst_l4_port": "670", "entry_type": "static"} #check the entry in asic db - tbl = swsscommon.Table(self.asicdb, "ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY") - keys = tbl.getKeys() - assert len(keys) == 2 + keys = self.asic_db.wait_for_n_keys("ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY", 2) for key in keys: - if (key.find("src_ip:18.18.18.2")) or (key.find("l4_src_port:182")): - assert True - if (key.find("dst_ip:18.18.18.1")) or (key.find("l4_dst_port:181")): - assert True - else: - assert False - - @pytest.mark.xfail(reason="vs image issue: Azure/sonic-sairedis#575") - def test_DelTwiceNaPtStaticEntry(self, dvs, testlog): + if "\"src_ip\":\"18.18.18.2\"" in key or "\"l4_src_port\":\"182\"" in key: + assert True + elif "\"dst_ip\":\"67.66.65.1\"" in key or "\"l4_dst_port\":\"660\"" in key: + assert True + else: + assert False + + def test_DelTwiceNaPtStaticEntry(self, app_db, asic_db, config_db, dvs, testlog): # initialize - self.setup_db(dvs) + self.setup_db(app_db, asic_db, config_db) # delete a static nat entry dvs.runcmd("config nat remove static udp 67.66.65.2 670 18.18.18.1 181") dvs.runcmd("config nat remove static udp 67.66.65.1 660 18.18.18.2 182") # check the entry is not there in the config db - tbl = swsscommon.Table(self.configdb, "STATIC_NAPT") - entry = tbl.getKeys() - assert entry == () + self.config_db.wait_for_n_keys("STATIC_NAPT", 0) # check the entry is not there in app db - tbl = swsscommon.Table(self.appdb, "NAPT_TWICE_TABLE") - entry = tbl.getKeys() - assert entry == () + self.app_db.wait_for_n_keys("NAPT_TWICE_TABLE", 0) #check the entry is not there in asic db - tbl = swsscommon.Table(self.asicdb, "ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY") - key = tbl.getKeys() - assert key == () + self.asic_db.wait_for_n_keys("ASIC_STATE:SAI_OBJECT_TYPE_NAT_ENTRY", 0) # clear interfaces self.clear_interfaces(dvs)