TestON/tests/USECASE/SegmentRouting/INT/INT.py

# SPDX-FileCopyrightText: Copyright 2021-present Open Networking Foundation.
# SPDX-License-Identifier: GPL-2.0-or-later

class INT:

    def __init__(self):
        self.default = ""

    def CASE1 (self, main):
        """
        Send ping packets from one host to another host and check flows from DeepInsight.
        """
        import time
        import socket
        from core import utilities
        from tests.USECASE.SegmentRouting.INT.dependencies.IntTest import IntTest
        main.cfgName = "CASE1"

        main.step("Setting up the test")
        intTest = IntTest(scapy=True)
        intTest.setUpTest(main)

        main.step("Setting up hosts and variables")
        srcIfaceName = main.h1.interfaces[0]["name"]
        dstIfaceName = main.h2.interfaces[0]["name"]
        srcMac = main.h1.getMac(srcIfaceName)
        dstMac = main.h2.getMac(dstIfaceName)
        srcIp = main.h1.getIp(srcIfaceName)
        dstIp = main.h2.getIp(dstIfaceName)
        srcPort = 2000
        dstPort = 8888

        main.step("Send ping packets from h1 to h2")
        startTimeMs = (time.time() - 5) * 1000
        pkt = """(
            Ether(src="{}", dst="{}") /
            IP(src="{}", dst="{}") /
            UDP(sport={}, dport={}) /
            ("A"*30)
        )""".format(srcMac, dstMac, srcIp, dstIp, srcPort, dstPort)
        main.h1.sendPacket(iface=srcIfaceName, packet=pkt)
        endTimeMs = (time.time() + 5) * 1000

        main.step("Checking total number of flow reports from DeepInsight")
        def getFiveTupleCount(*args, **kwargs):
            flows = main.DeepInsight.getFlows(
                startTimeMs=startTimeMs,
                endTimeMs=endTimeMs,
                srcIp=srcIp,
                dstIp=dstIp,
                ipProto=socket.IPPROTO_UDP
            )
            if "FiveTupleCount" in flows:
                return flows["FiveTupleCount"]
            else:
                return 0
        # Need to wait few seconds until DeepInsight database updated.
        fiveTupleCount = utilities.retry(
            f=getFiveTupleCount,
            retValue=0,
            attempts=60,
        )

        utilities.assert_equals(
            expect=1, actual=fiveTupleCount,
            onpass="Got 1 flow report from DeepInsight as expected.",
            onfail="Got %d flow reports from DeepInsight (expect 1)" % (fiveTupleCount)
        )

        main.step("Clean up the test")
        intTest.cleanUp(main)

    def CASE2 (self, main):
        """
        Send a packet with invalid VLAN from one host to another host and check
        if DeepInsight receives drop reports.
        """
        import time
        import socket
        from core import utilities
        from tests.USECASE.SegmentRouting.INT.dependencies.IntTest import IntTest
        main.cfgName = "CASE2"

        main.step("Setting up the test")
        intTest = IntTest(scapy=True)
        intTest.setUpTest(main)

        main.step("Setting up hosts and variables")
        srcIfaceName = main.h1.interfaces[0]["name"]
        dstIfaceName = main.h2.interfaces[0]["name"]
        srcMac = main.h1.getMac(srcIfaceName)
        dstMac = main.h2.getMac(dstIfaceName)
        srcIp = main.h1.getIp(srcIfaceName)
        dstIp = main.h2.getIp(dstIfaceName)
        srcPort = 2000
        dstPort = 8888

        main.step("Sending a packet with invalid VLAN ID from h1")
        startTimeMs = (time.time() - 5) * 1000
        pkt = """(
            Ether(src="{}", dst="{}") /
            Dot1Q(vlan=4093) /
            IP(src="{}", dst="{}") /
            UDP(sport={}, dport={}) /
            ("A"*30)
        )""".format(srcMac, dstMac, srcIp, dstIp, srcPort, dstPort)
        main.h1.sendPacket(iface=srcIfaceName, packet=pkt)
        endTimeMs = (time.time() + 5) * 1000

        main.step("Checking drop report from DeepInsight")
        def getDropAnomalies(*args, **kwargs):
            return main.DeepInsight.getAnomalyRecords(
                startTime=startTimeMs,
                endTime=endTimeMs,
                srcIp=srcIp,
                dstIp=dstIp,
                srcPort=srcPort,
                dstPort=dstPort,
                ipProto=socket.IPPROTO_UDP,
                anomalyType="packet_drop",
           )

        # Need to wait few seconds until DeepInsight database updated.
        dropAnomalies = utilities.retry(
            f=getDropAnomalies,
            retValue=[[]],
            attempts=60,
        )

        utilities.assert_equals(
            expect=1, actual=len(dropAnomalies),
            onpass="Got 1 drop anomaly from DeepInsight as expected.",
            onfail="Got %d drop anomaly from DeepInsight, expect 1" % (len(dropAnomalies))
        )

        main.step("Checking drop reason from the report")
        dropAnomaly = dropAnomalies[0]
        dropReason = dropAnomaly["DropReason"]

        # DROP_REASON_PORT_VLAN_MAPPING_MISS = 55
        utilities.assert_equals(
            expect=55, actual=dropReason,
            onpass="Got drop reason '55' as expected.",
            onfail="Got drop reason '%d', expect '55'." % (dropReason)
        )

        main.step("Clean up the test")
        intTest.cleanUp(main)

    def CASE3 (self, main):
        """
        Send a packet with IP TTL value 1 from one host to another host and check
        if DeepInsight receives drop reports.
        """
        import time
        import socket
        from core import utilities
        from tests.USECASE.SegmentRouting.INT.dependencies.IntTest import IntTest
        main.cfgName = "CASE3"

        main.step("Setting up the test")
        intTest = IntTest(scapy=True)
        intTest.setUpTest(main)

        main.step("Setting up hosts and variables")
        srcIfaceName = main.h1.interfaces[0]["name"]
        dstIfaceName = main.h3.interfaces[0]["name"]
        srcMac = main.h1.getMac(srcIfaceName)
        dstMac = main.params.get("routerMac", "00:00:00:00:00:00")
        srcIp = main.h1.getIp(srcIfaceName)
        dstIp = main.h3.getIp(dstIfaceName)
        srcPort = 3000
        dstPort = 8888

        main.step("Sending a packet with IP TTL value 1 from h1")
        startTimeMs = (time.time() - 5) * 1000
        pkt = """(
            Ether(src="{}", dst="{}") /
            IP(src="{}", dst="{}", ttl=1) /
            UDP(sport={}, dport={}) /
            ("A"*30)
        )""".format(srcMac, dstMac, srcIp, dstIp, srcPort, dstPort)
        main.h1.sendPacket(iface=srcIfaceName, packet=pkt)
        endTimeMs = (time.time() + 5) * 1000

        main.step("Checking drop report from DeepInsight")
        def getDropAnomalies(*args, **kwargs):
            return main.DeepInsight.getAnomalyRecords(
                startTime=startTimeMs,
                endTime=endTimeMs,
                srcIp=srcIp,
                dstIp=dstIp,
                srcPort=srcPort,
                dstPort=dstPort,
                ipProto=socket.IPPROTO_UDP,
                anomalyType="packet_drop",
            )

        # Need to wait few seconds until DeepInsight database updated.
        dropAnomalies = utilities.retry(
            f=getDropAnomalies,
            retValue=[[]],
            attempts=60,
        )

        utilities.assert_equals(
            expect=1, actual=len(dropAnomalies),
            onpass="Got 1 drop anomaly from DeepInsight as expected.",
            onfail="Got %d drop anomaly from DeepInsight, expect '1'." % (len(dropAnomalies))
        )

        main.step("Checking drop reason from report")
        dropAnomaly = dropAnomalies[0]
        dropReason = dropAnomaly["DropReason"]
        # DROP_REASON_IP_TTL_ZERO = 26
        utilities.assert_equals(
            expect=26, actual=dropReason,
            onpass="Got drop reason '26' as expected.",
            onfail="Got drop reason '%d', expect '26'." % (dropReason)
        )

        main.step("Clean up the test")
        intTest.cleanUp(main)