olt-complex.py

'''
More complex (composite) test-cases for the OLT
'''

import logging
from oftest.testutils import *
from oltbase import OltBaseTest
from oltconstants import *


class Thing(object):
    """An object we can stash arbitrary attributes for easy reach"""
    def __init__(self, **kws):
        self.__dict__.update(kws)


class ControllerAccess(OltBaseTest):
    """Verify openflow access from OLT device"""

    def runTest(self):
        logging.info("Running ControllerAccess test")

        # implicitly testing access to the openflow device
        # and deleting all residual flows and groups
        delete_all_flows(self.controller)
        delete_all_groups(self.controller)


class TestScenario1SingleOnu(OltBaseTest):
    """
    Run a comprehensive test scenrario on the OLT.

    Plan:

    1.  Reset the OLT into a clean state
    2.  Setup the OLT to pass authentication (L2/Unicast) traffic for the ONU1
    3.  Verify that unicast traffic works in both directions (use a few hundred frames in both ways)
    4.  Setup IGMP forwarding toward the controller
    5.  Send periodic IGMP queries out toward the ONU and verify its arrival
    6.  Send in an IGMP join request for one channel (specific multicast address) and verify that
        the controller receives it.
    7.  Setup flows for forwarding multicast traffic from the OLT port toward the ONU(s)
    8.  Verify that multicast packets can reach the ONU
    9.  Verify that bidirectional unicast traffic still works across the PON
    10. Change channel to a new multicast group and verify both multicast receiption as well as
        unicast traffic works. This step involves:
        - sending a leave message by the ONU to the controller and verify its reception
        - sending a join message and verify its reception
        - removing the existing multicast flow
        - adding a new multicast flow
        - verify that original flow no longer flows
        - verify new flow
        - verify that unicast still works
    11. Add a second channel while keeping the existing one. Verify all what needs to be verified
        (similar as above)
    12. Add two more multicast channels, and verify everything
    13. Flip a channel for one of the multicast channels, and verify everything
    14. Tear down the test.
    """

    def runTest(self):
        logging.info("Running %s" % self.__class__.__name__)

        # Some constants
        c_vlan_id = 111
        mcast_vlan_id = 140
        mcast_groups = Thing(
            ch1="230.10.10.10",
            ch2="231.11.11.11",
            ch3="232.12.12.12",
            ch4="233.13.13.13",
            ch5="234.14.14.14"
        )
        onu1 = Thing(
            port=onu_port,
            ip="13.14.14.13",
            mac="b6:b8:3e:fb:1a:3f",
            s_vlan_id=13
        )

        # 1.  Reset the OLT into a clean state
        self.resetOlt()

        # 2.  Setup the OLT to pass authentication (L2/Unicast) traffic for the ONU1
        self.installDoubleTaggingRules(onu1.s_vlan_id, c_vlan_id, self.getCookieBlock())

        # 3.  Verify that unicast traffic works in both directions (use a few hundred frames in both ways)
        self.testPacketFlow(onu1.s_vlan_id, c_vlan_id)  # this tests just one packet in each way
        self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 100)

        # 4.  Setup IGMP forwarding toward the controller
        self.setupIgmpCaptureFlowRules(self.getCookieBlock())
        self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows

        # 5.  Send periodic IGMP queries out toward the ONU and verify its arrival
        self.testIgmpQueryOut()
        self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows

        # 6.  Send in an IGMP join request for one channel (specific multicast address) and verify that
        #     the controller receives it.
        self.sendIgmpReport(join=[mcast_groups.ch1])

        # 7.  Setup flows for forwarding multicast traffic from the OLT port toward the ONU(s)
        ch1_cookie = self.getCookieBlock()
        ch1_group_id = 1
        self.setupMcastChannel(mcast_groups.ch1, mcast_vlan_id, [onu1.port], ch1_group_id, ch1_cookie)

        # 8.  Verify that multicast packets can reach the ONU
        self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port], 10)

        # 9.  Verify that bidirectional unicast traffic still works across the PON
        self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows

        # 10. Change channel to a new multicast group and verify both multicast receiption as well as
        #     unicast traffic works. This step involves:
        #     - sending a leave message by the ONU to the controller and verify its reception
        #     - sending a join message and verify its reception
        #     - removing the existing multicast flow
        #     - adding a new multicast flow
        #     - verify that original flow no longer flows
        #     - verify new flow
        #     - verify that unicast still works
        self.sendIgmpReport(leave=[mcast_groups.ch1], join=[mcast_groups.ch2])
        self.removeMcastChannel(mcast_groups.ch1, mcast_vlan_id, [onu1.port], ch1_group_id, ch1_cookie)
        ch2_cookie = self.getCookieBlock()
        ch2_group_id = 2
        self.setupMcastChannel(mcast_groups.ch2, mcast_vlan_id, [onu1.port], ch2_group_id, ch2_cookie)
        self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port], expect_to_be_blocked=True)
        self.testMcastFlow(mcast_groups.ch2, mcast_vlan_id, [onu1.port], 10)
        self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows

        # 11. Add a second channel while keeping the existing one. Verify all what needs to be verified
        #     (similar as above)
        self.sendIgmpReport(join=[mcast_groups.ch2, mcast_groups.ch3])
        ch3_cookie = self.getCookieBlock()
        ch3_group_id = 3
        self.setupMcastChannel(mcast_groups.ch3, mcast_vlan_id, [onu1.port], ch3_group_id, ch3_cookie)
        self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port], expect_to_be_blocked=True)
        self.testMcastFlow(mcast_groups.ch2, mcast_vlan_id, [onu1.port], 10)
        self.testMcastFlow(mcast_groups.ch3, mcast_vlan_id, [onu1.port], 10)
        self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows

        # 12. Add two more multicast channels, and verify everything again
        self.sendIgmpReport(join=[mcast_groups.ch2, mcast_groups.ch3, mcast_groups.ch4, mcast_groups.ch5])
        ch4_cookie = self.getCookieBlock()
        ch4_group_id = 4
        self.setupMcastChannel(mcast_groups.ch4, mcast_vlan_id, [onu1.port], ch4_group_id, ch4_cookie)
        ch5_cookie = self.getCookieBlock()
        ch5_group_id = 5
        self.setupMcastChannel(mcast_groups.ch5, mcast_vlan_id, [onu1.port], ch5_group_id, ch5_cookie)
        self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port], expect_to_be_blocked=True)
        self.testMcastFlow(mcast_groups.ch2, mcast_vlan_id, [onu1.port], 10)
        self.testMcastFlow(mcast_groups.ch3, mcast_vlan_id, [onu1.port], 10)
        self.testMcastFlow(mcast_groups.ch4, mcast_vlan_id, [onu1.port], 10)
        self.testMcastFlow(mcast_groups.ch5, mcast_vlan_id, [onu1.port], 10)
        self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows

        # 13. Flip a channel for one of the multicast channels, and verify everything (ch3 -> ch1)
        self.sendIgmpReport(join=[mcast_groups.ch1, mcast_groups.ch2, mcast_groups.ch4, mcast_groups.ch5],
                            leave=[mcast_groups.ch3])
        self.removeMcastChannel(mcast_groups.ch3, mcast_vlan_id, [onu1.port], ch3_group_id, ch3_cookie)
        ## reusing the same group id and cookie
        self.setupMcastChannel(mcast_groups.ch1, mcast_vlan_id, [onu1.port], ch1_group_id, ch1_cookie)
        self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port], 10)
        self.testMcastFlow(mcast_groups.ch2, mcast_vlan_id, [onu1.port], 10)
        self.testMcastFlow(mcast_groups.ch3, mcast_vlan_id, [onu1.port], expect_to_be_blocked=True)
        self.testMcastFlow(mcast_groups.ch4, mcast_vlan_id, [onu1.port], 10)
        self.testMcastFlow(mcast_groups.ch5, mcast_vlan_id, [onu1.port], 10)
        self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows

        # 14. Tear down the test.
        self.resetOlt()