oltbase.py

import oftest.base_tests as base_tests
from oftest import config
import ofp
from oftest.testutils import *
from oltconstants import *
import copy
import logging


class OltBaseTest(base_tests.SimpleDataPlane):

    def testPacketIn(self, match, parsed_pkt):
        delete_all_flows(self.controller)

        pkt = str(parsed_pkt)

        for of_port in config["port_map"]:
            m = copy.deepcopy(match)
            m.oxm_list.append(ofp.oxm.in_port(of_port))
            request = ofp.message.flow_add(
                table_id=test_param_get("table", 0),
                cookie=42,
                match=m,
                instructions=[
                    ofp.instruction.apply_actions(
                        actions=[
                            ofp.action.output(
                                port=ofp.OFPP_CONTROLLER,
                                max_len=ofp.OFPCML_NO_BUFFER)])],
                buffer_id=ofp.OFP_NO_BUFFER,
                priority=1000)
            logging.info("Inserting flow sending matching packets to controller")
            self.controller.message_send(request)
            do_barrier(self.controller)

    for of_port in config["port_map"]:
        logging.info("PacketInExact test, port %d", of_port)
        self.dataplane.send(of_port, pkt)
        verify_packet_in(self, pkt, of_port, ofp.OFPR_ACTION)
        verify_packets(self, pkt, [])

    def processEapolRule(self, in_port, install = True):
        match = ofp.match()
        match.oxm_list.append(ofp.oxm.eth_type(0x888e))
        match.oxm_list.append(ofp.oxm.in_port(in_port))
        if install:
            request = ofp.message.flow_add(
                    table_id=test_param_get("table", 0),
                    cookie=42,
                    match=match,
                    instructions=[
                        ofp.instruction.apply_actions(
                            actions=[
                                ofp.action.output(
                                    port=ofp.OFPP_CONTROLLER,
                                    max_len=ofp.OFPCML_NO_BUFFER)])],
                    buffer_id=ofp.OFP_NO_BUFFER,
                    priority=1000)
        else:
            request = ofp.message.flow_delete(
                    table_id=test_param_get("table", 0),
                    cookie=42,
                    match=match,
                    instructions=[
                        ofp.instruction.apply_actions(
                            actions=[
                                ofp.action.output(
                                    port=ofp.OFPP_CONTROLLER,
                                    max_len=ofp.OFPCML_NO_BUFFER)])],
                    buffer_id=ofp.OFP_NO_BUFFER,
                    priority=1000)
        logging.info("%s flow sending matching packets to controller" % "Install" if install else "Remove")
        self.controller.message_send(request)
        do_barrier(self.controller)

    def testPacketFlow(self, c_vlan_id, s_vlan_id):

        incorrectTagPkt = simple_udp_packet(pktlen=100, dl_vlan_enable=True, vlan_vid=100, vlan_pcp=1)
        zeroTaggedPkt = simple_udp_packet(pktlen=100, dl_vlan_enable=True, vlan_vid=0, vlan_pcp=0)
        untaggedPkt = simple_udp_packet(pktlen=96)

        upstreamDoubleTaggedPkt = double_vlan_udp_packet(pktlen=104, dl_vlan_enable=True,
                                                         c_vlan_vid=c_vlan_id,
                                                         s_vlan_vid=s_vlan_id,
                                                         c_vlan_pcp=0, s_vlan_pcp=0)

        logging.info("Testing s-tag %d, c-tag %d" % (s_vlan_id, c_vlan_id))

        # test upstream untagged packet got double tag at OLT
        self.dataplane.send(onu_port, str(zeroTaggedPkt))
        verify_packet(self, upstreamDoubleTaggedPkt, olt_port)

        # test downstream doubletagged packet got untagged at ONU
        self.dataplane.send(olt_port, str(upstreamDoubleTaggedPkt))
        if device_type == "pmc":
            verify_packet(self, zeroTaggedPkt, onu_port)
        else:
            verify_packet(self, untaggedPkt, onu_port)

        # test upstream doubletagged packet got dropped
        self.dataplane.send(onu_port, str(upstreamDoubleTaggedPkt))
        verify_no_packet(self, upstreamDoubleTaggedPkt, olt_port)

        # test downstream untagged packet got dropped at ONU
        self.dataplane.send(olt_port, str(untaggedPkt))
        verify_no_packet(self, untaggedPkt, onu_port)

        # test upstream icorrectly tagged packet; should get dropped
        self.dataplane.send(onu_port, str(incorrectTagPkt))
        verify_no_packet(self, upstreamDoubleTaggedPkt, olt_port)

    def installDoubleTaggingRules(self, s_vlan_id, c_vlan_id, cookie=42):

        # upstream flow rule
        match = ofp.match()
        match.oxm_list.append(ofp.oxm.in_port(onu_port))
        if device_type == "cpqd":
            match.oxm_list.append(ofp.oxm.vlan_vid(value=ofp.OFPVID_NONE))
            actions = [
                ofp.action.push_vlan(ethertype=0x8100),
                ofp.action.set_field(ofp.oxm.vlan_vid(ofp.OFPVID_PRESENT | c_vlan_id)),
                ofp.action.set_field(ofp.oxm.vlan_pcp(0))
            ]
        else:  # "pmc", "normal"
            match.oxm_list.append(ofp.oxm.vlan_vid(value=ofp.OFPVID_PRESENT))
            match.oxm_list.append(ofp.oxm.vlan_pcp(value=0))
            actions = [
                ofp.action.set_field(ofp.oxm.vlan_vid(ofp.OFPVID_PRESENT | c_vlan_id))
            ]
        cookie += 1

        # push inner vlan (c-vlan) for upstream
        request = ofp.message.flow_add(
            table_id=test_param_get("table", 0),
            cookie=cookie,
            match=match,
            instructions=[
                ofp.instruction.apply_actions(actions=actions),
                ofp.instruction.goto_table(1)],
            buffer_id=ofp.OFP_NO_BUFFER,
            priority=1000)

        self.controller.message_send(request)
        do_barrier(self.controller)
        verify_no_errors(self.controller)

        # push outer vlan (s-vlan) for upstream
        match = ofp.match()
        match.oxm_list.append(ofp.oxm.in_port(onu_port))
        match.oxm_list.append(ofp.oxm.vlan_vid(ofp.OFPVID_PRESENT | c_vlan_id))
        match.oxm_list.append(ofp.oxm.vlan_pcp(0))
        cookie += 1

        request = ofp.message.flow_add(
            table_id=test_param_get("table", 1),
            cookie=cookie,
            match=match,
            instructions=[
                ofp.instruction.apply_actions(
                    actions=[
                        ofp.action.push_vlan(ethertype=0x8100),
                        ofp.action.set_field(ofp.oxm.vlan_vid(ofp.OFPVID_PRESENT | s_vlan_id)),
                        ofp.action.set_field(ofp.oxm.vlan_pcp(0)),
                        ofp.action.output(port=olt_port)]),
            ],
            buffer_id=ofp.OFP_NO_BUFFER,
            priority=1000)

        self.controller.message_send(request)
        do_barrier(self.controller)
        verify_no_errors(self.controller)
        cookie += 1

        # strip outer vlan (s-vlan) for downstream
        match = ofp.match()
        match.oxm_list.append(ofp.oxm.in_port(olt_port))
        match.oxm_list.append(ofp.oxm.vlan_vid(ofp.OFPVID_PRESENT | s_vlan_id))
        match.oxm_list.append(ofp.oxm.vlan_pcp(0))
        request = ofp.message.flow_add(
            table_id=test_param_get("table", 0),
            cookie=cookie,
            match=match,
            instructions=[
                ofp.instruction.apply_actions(
                    actions=[ofp.action.pop_vlan()]),
                ofp.instruction.goto_table(1)],
            buffer_id=ofp.OFP_NO_BUFFER,
            priority=1000)

        self.controller.message_send(request)
        do_barrier(self.controller)
        verify_no_errors(self.controller)

        # rewrite inner vlan (c-vlan) to default (0) for downstream
        match = ofp.match()
        match.oxm_list.append(ofp.oxm.in_port(olt_port))
        match.oxm_list.append(ofp.oxm.vlan_vid(ofp.OFPVID_PRESENT | c_vlan_id))
        match.oxm_list.append(ofp.oxm.vlan_pcp(0))
        cookie += 1

        request = ofp.message.flow_add(
            table_id=test_param_get("table", 1),
            cookie=cookie,
            match=match,
            instructions=[
                ofp.instruction.apply_actions(
                    actions=[
                        ofp.action.pop_vlan(),
                        ofp.action.output(port=onu_port)])
            ],
            buffer_id=ofp.OFP_NO_BUFFER,
            priority=1000)

        self.controller.message_send(request)
        do_barrier(self.controller)
        verify_no_errors(self.controller)