[tests][twamporch] TWAMP Light orchagent implementation. (#2927)

* add TWAMP Light vstest cases

Signed-off-by: Xiaodong Hu <[email protected]>

tests/conftest.py

@@ -28,6 +28,7 @@
 from dvslib import dvs_policer
 from dvslib import dvs_hash
 from dvslib import dvs_switch
+from dvslib import dvs_twamp
 
 from buffer_model import enable_dynamic_buffer
 
@@ -1945,6 +1946,14 @@ def dvs_hash_manager(request, dvs):
 def dvs_switch_manager(request, dvs):
     request.cls.dvs_switch = dvs_switch.DVSSwitch(dvs.get_asic_db())
 
+@pytest.fixture(scope="class")
+def dvs_twamp_manager(request, dvs):
+    request.cls.dvs_twamp = dvs_twamp.DVSTwamp(dvs.get_asic_db(),
+                                               dvs.get_config_db(),
+                                               dvs.get_state_db(),
+                                               dvs.get_counters_db(),
+                                               dvs.get_app_db())
+
 ##################### DPB fixtures ###########################################
 def create_dpb_config_file(dvs):
     cmd = "sonic-cfggen -j /etc/sonic/init_cfg.json -j /tmp/ports.json --print-data > /tmp/dpb_config_db.json"

tests/dvslib/dvs_twamp.py

@@ -0,0 +1,98 @@
+"""Utilities for interacting with TWAMP Light objects when writing VS tests."""
+
+class DVSTwamp(object):
+    def __init__(self, adb, cdb, sdb, cntrdb, appdb):
+        self.asic_db = adb
+        self.config_db = cdb
+        self.state_db = sdb
+        self.counters_db = cntrdb
+        self.app_db = appdb
+
+    def create_twamp_light_session_sender_packet_count(self, name, sip, sport, dip, dport, packet_count=100, tx_interval=100, timeout=5, stats_interval=None):
+        twamp_light_entry = {"mode": "LIGHT",
+                             "role": "SENDER",
+                             "src_ip": sip,
+                             "src_udp_port": sport,
+                             "dst_ip": dip,
+                             "dst_udp_port": dport,
+                             "packet_count": packet_count,
+                             "tx_interval": tx_interval,
+                             "timeout": timeout
+                            }
+        if stats_interval:
+            twamp_light_entry["statistics_interval"] = str(stats_interval)
+        else:
+            twamp_light_entry["statistics_interval"] = str(int(packet_count) * int(tx_interval) + int(timeout)*1000)
+        self.config_db.create_entry("TWAMP_SESSION", name, twamp_light_entry)
+
+    def create_twamp_light_session_sender_continuous(self, name, sip, sport, dip, dport, monitor_time=0, tx_interval=100, timeout=5, stats_interval=None):
+        twamp_light_entry = {"mode": "LIGHT",
+                             "role": "SENDER",
+                             "src_ip": sip,
+                             "src_udp_port": sport,
+                             "dst_ip": dip,
+                             "dst_udp_port": dport,
+                             "monitor_time": monitor_time,
+                             "tx_interval": tx_interval,
+                             "timeout": timeout
+                            }
+        if stats_interval:
+            twamp_light_entry["statistics_interval"] = str(stats_interval)
+        else:
+            twamp_light_entry["statistics_interval"] = str(int(monitor_time)*1000)
+        self.config_db.create_entry("TWAMP_SESSION", name, twamp_light_entry)
+
+    def create_twamp_light_session_reflector(self, name, sip, sport, dip, dport):
+        twamp_light_entry = {"mode": "LIGHT",
+                             "role": "REFLECTOR",
+                             "src_ip": sip,
+                             "src_udp_port": sport,
+                             "dst_ip": dip,
+                             "dst_udp_port": dport
+                            }
+        self.config_db.create_entry("TWAMP_SESSION", name, twamp_light_entry)
+
+    def start_twamp_light_sender(self, name):
+        twamp_light_entry = {"admin_state": "enabled"}
+        self.config_db.create_entry("TWAMP_SESSION", name, twamp_light_entry)
+
+    def stop_twamp_light_sender(self, name):
+        twamp_light_entry = {"admin_state": "disabled"}
+        self.config_db.create_entry("TWAMP_SESSION", name, twamp_light_entry)
+
+    def remove_twamp_light_session(self, name):
+        self.config_db.delete_entry("TWAMP_SESSION", name)
+
+    def get_twamp_light_session_status(self, name):
+        return self.get_twamp_light_session_state(name)["status"]
+
+    def get_twamp_light_session_state(self, name):
+        tbl = swsscommon.Table(self.sdb, "TWAMP_SESSION_TABLE")
+        (status, fvs) = tbl.get(name)
+        assert status == True
+        assert len(fvs) > 0
+        return { fv[0]: fv[1] for fv in fvs }
+
+    def verify_session_status(self, name, status="active", expected=1):
+        self.state_db.wait_for_n_keys("TWAMP_SESSION_TABLE", expected)
+        if expected:
+            self.state_db.wait_for_field_match("TWAMP_SESSION_TABLE", name, {"status": status})
+
+    def verify_no_session(self):
+        self.config_db.wait_for_n_keys("TWAMP_SESSION", 0)
+        self.state_db.wait_for_n_keys("TWAMP_SESSION_TABLE", 0)
+
+    def verify_session_asic_db(self, dvs, name, asic_table=None, expected=1):
+        session_oids = self.asic_db.wait_for_n_keys("ASIC_STATE:SAI_OBJECT_TYPE_TWAMP_SESSION", expected)
+        session_oid = session_oids[0]
+        dvs.asic_db.wait_for_field_match("ASIC_STATE:SAI_OBJECT_TYPE_TWAMP_SESSION", session_oid, asic_table)
+
+    def verify_session_counter_db(self, dvs, name, counter_table=None, expected=1, expected_item=1):
+        fvs = dvs.counters_db.get_entry("COUNTERS_TWAMP_SESSION_NAME_MAP", "")
+        fvs = dict(fvs)
+        total_key = self.counters_db.db_connection.keys("COUNTERS:{}".format(fvs[name]))
+        assert len(total_key) == expected, "TWAMP Light counter entries are not available in counter db"
+        dvs.counters_db.wait_for_field_match("COUNTERS", fvs[name], counter_table)
+        item_keys = self.counters_db.db_connection.keys("COUNTERS:{}:INDEX:*".format(fvs[name]))
+        assert len(item_keys) == expected_item, "TWAMP Light counter entries are not available in counter db"
+

tests/test_twamp.py

@@ -0,0 +1,182 @@
+# This test suite covers the functionality of twamp light feature in SwSS
+import pytest
+import time
+
+@pytest.mark.usefixtures("testlog")
+@pytest.mark.usefixtures('dvs_twamp_manager')
+class TestTwampLight(object):
+
+    def check_syslog(self, dvs, marker, log, expected_cnt):
+        (ec, out) = dvs.runcmd(['sh', '-c', "awk \'/%s/,ENDFILE {print;}\' /var/log/syslog | grep \'%s\' | wc -l" % (marker, log)])
+        assert out.strip() == str(expected_cnt)
+
+    def test_SenderPacketCountSingle(self, dvs, testlog):
+        """
+        This test covers the TWAMP Light session creation and removal operations
+        Operation flow:
+        1. Create twamp-light session-sender using once packet-count
+           The session remains inactive
+        2. Start twamp-light session
+           The session becomes active
+        3. Remove twamp-light session
+        """
+
+        session = "TEST_SENDER1"
+        src_ip = "1.1.1.1"
+        src_udp_port = "862"
+        dst_ip = "2.2.2.2"
+        dst_udp_port = "863"
+        packet_count = "1000"
+        tx_interval = "10"
+        timeout = "10"
+        stats_interval = "20000"
+
+        marker = dvs.add_log_marker()
+
+        # create twamp-light session
+        self.dvs_twamp.create_twamp_light_session_sender_packet_count(session, src_ip, src_udp_port, dst_ip, dst_udp_port, packet_count, tx_interval, timeout)
+
+        # start twamp-light session
+        self.dvs_twamp.start_twamp_light_sender(session)
+
+        # wait for sending TWAMP-test done
+        time.sleep(12)
+
+        # remove twamp-light session
+        self.dvs_twamp.remove_twamp_light_session(session)
+        self.dvs_twamp.verify_no_session()
+
+    def test_SenderPacketCountMulti(self, dvs, testlog):
+        """
+        This test covers the TWAMP Light Sender session creation and removal operations
+        Operation flow:
+        1. Create twamp-light session-sender using multi packet-count
+           The session remains inactive
+        2. Start twamp-light session
+           The session becomes active
+        3. Remove twamp-light session
+        """
+
+        session = "TEST_SENDER1"
+        src_ip = "1.2.3.4"
+        src_udp_port = "862"
+        dst_ip = "5.6.7.8"
+        dst_udp_port = "863"
+        packet_count = "1000"
+        tx_interval = "10"
+        timeout = "10"
+        stats_interval = "11000"
+
+        marker = dvs.add_log_marker()
+
+        # create twamp-light session
+        self.dvs_twamp.create_twamp_light_session_sender_packet_count(session, src_ip, src_udp_port, dst_ip, dst_udp_port, packet_count, tx_interval, timeout, stats_interval)
+
+        # start twamp-light session
+        self.dvs_twamp.start_twamp_light_sender(session)
+
+        # wait for sending TWAMP-test done
+        time.sleep(120)
+
+        # remove twamp-light session
+        self.dvs_twamp.remove_twamp_light_session(session)
+        self.dvs_twamp.verify_no_session()
+
+    def test_SenderContinuousSingle(self, dvs, testlog):
+        """
+        This test covers the TWAMP Light Sender session creation and removal operations
+        Operation flow:
+        1. Create twamp-light session-sender using once continuous
+           The session remains inactive
+        2. Start twamp-light session
+           The session becomes active
+        3. Remove twamp-light session
+        """
+
+        session = "TEST_SENDER2"
+        src_ip = "11.11.11.11"
+        src_udp_port = "862"
+        dst_ip = "12.12.12.12"
+        dst_udp_port = "863"
+        monitor_time = "60"
+        tx_interval = "100"
+        timeout = "10"
+        stats_interval = "60000"
+
+        marker = dvs.add_log_marker()
+
+        # create twamp-light session
+        self.dvs_twamp.create_twamp_light_session_sender_continuous(session, src_ip, src_udp_port, dst_ip, dst_udp_port, monitor_time, tx_interval, timeout)
+
+        # start twamp-light session
+        self.dvs_twamp.start_twamp_light_sender(session)
+        # wait for sending TWAMP-test done
+        time.sleep(60)
+
+        # remove twamp-light session
+        self.dvs_twamp.remove_twamp_light_session(session)
+        self.dvs_twamp.verify_no_session()
+
+    def test_SenderContinuousMulti(self, dvs, testlog):
+        """
+        This test covers the continuous TWAMP Light Sender session creation and removal operations
+        Operation flow:
+        1. Create twamp-light session-sender using multi continuous
+           The session remains inactive
+        2. Start twamp-light session
+           The session becomes active
+        3. Remove twamp-light session
+        """
+
+        session = "TEST_SENDER2"
+        src_ip = "11.12.13.14"
+        src_udp_port = "862"
+        dst_ip = "15.16.17.18"
+        dst_udp_port = "863"
+        monitor_time = "60"
+        tx_interval = "100"
+        timeout = "10"
+        stats_interval = "20000"
+
+        marker = dvs.add_log_marker()
+
+        # create twamp-light session
+        self.dvs_twamp.create_twamp_light_session_sender_continuous(session, src_ip, src_udp_port, dst_ip, dst_udp_port, monitor_time, tx_interval, timeout, stats_interval)
+
+        # start twamp-light session
+        self.dvs_twamp.start_twamp_light_sender(session)
+
+        # wait for sending TWAMP-test done
+        time.sleep(60)
+
+        # remove twamp-light session
+        self.dvs_twamp.remove_twamp_light_session(session)
+        self.dvs_twamp.verify_no_session()
+
+    def test_Reflector(self, dvs, testlog):
+        """
+        This test covers the TWAMP Light Reflector session creation and removal operations
+        Operation flow:
+        1. Create twamp-light session-reflector
+        2. Remove twamp-light session
+        """
+
+        session = "TEST_REFLECTOR1"
+        src_ip = "22.1.1.1"
+        src_udp_port = "862"
+        dst_ip = "22.1.1.2"
+        dst_udp_port = "863"
+
+        marker = dvs.add_log_marker()
+
+        # create twamp-light session
+        self.dvs_twamp.create_twamp_light_session_reflector(session, src_ip, src_udp_port, dst_ip, dst_udp_port)
+
+        # remove twamp-light session
+        self.dvs_twamp.remove_twamp_light_session(session)
+        self.dvs_twamp.verify_no_session()
+
+# Add Dummy always-pass test at end as workaroud
+# for issue when Flaky fail on final test it invokes module tear-down before retrying
+def test_nonflaky_dummy():
+    pass