oltconstants.py

import logging
from oftest.testutils import test_param_get, MINSIZE
import oftest.packet as scapy
from oftest import config

import ofp

# These parameters can be altered from the command line using the -t or --test-params= options.
# Example: -t 'onu_port=129;olt_port=288;device_type=pmc'
#
onu_port = test_param_get("onu_port", 130)
onu_port2 = test_param_get("onu_port2", 131)
olt_port = test_param_get("olt_port", 258)
device_type = test_param_get("device_type", "normal")  # options: "normal", "pmc", "cpqd"
fake_dataplane = test_param_get("fake_dataplane", False)

def createAllGroupAdd(group_id, ports=[]):
    buckets = []

    for portNum in ports:
        buckets.append(ofp.common.bucket(watch_port=ofp.OFPP_ANY, watch_group=ofp.OFPG_ANY,
                                         actions=[ofp.action.output(port=portNum)]))

    group_add = ofp.message.group_add(group_type=ofp.OFPGT_ALL, group_id=group_id, buckets=buckets)

    return group_add


def createAllGroupMod(group_id, ports=[]):
    buckets = []

    for portNum in ports:
        buckets.append(ofp.common.bucket(watch_port=ofp.OFPP_ANY, watch_group=ofp.OFPG_ANY,
                                         actions=[ofp.action.output(port=portNum)]))

    group_mod = ofp.message.group_mod(command=ofp.OFPGC_MODIFY, group_type=ofp.OFPGT_ALL, group_id=group_id,
                                      buckets=buckets)

    return group_mod

def double_vlan_udp_packet(pktlen=100,
                           eth_dst='00:01:02:03:04:05',
                           eth_src='00:06:07:08:09:0a',
                           dl_vlan_enable=False,
                           c_vlan_vid=0,
                           c_vlan_pcp=0,
                           s_vlan_vid=0,
                           s_vlan_pcp=0,
                           ip_src='192.168.0.1',
                           ip_dst='192.168.0.2',
                           ip_tos=0,
                           ip_ttl=64,
                           udp_sport=1234,
                           udp_dport=80,
                           ip_ihl=None,
                           ip_options=False,
                           eth_type=0x8100
                           ):
    """
    Return a double vlan tagged dataplane UDP packet
    Supports a few parameters:
    @param len Length of packet in bytes w/o CRC
    @param eth_dst Destination MAC
    @param eth_src Source MAC
    @param dl_vlan_enable True if the packet is with vlan, False otherwise
    @param c_vlan_vid CVLAN ID
    @param c_vlan_pcp CVLAN priority
    @param s_vlan_vid SVLAN ID
    @param s_vlan_pcp SVLAN priority
    @param ip_src IP source
    @param ip_dst IP destination
    @param ip_tos IP ToS
    @param ip_ttl IP TTL
    @param udp_dport UDP destination port
    @param udp_sport UDP source port

    Generates a simple UDP packet. Users shouldn't assume anything about
    this packet other than that it is a valid ethernet/IP/UDP frame.
    """

    if MINSIZE > pktlen:
        pktlen = MINSIZE

    # Note Dot1Q.id is really CFI
    if (dl_vlan_enable):
        pkt = scapy.Ether(dst=eth_dst, src=eth_src, type=eth_type) / \
              scapy.Dot1Q(prio=s_vlan_pcp, vlan=s_vlan_vid) / \
              scapy.Dot1Q(prio=c_vlan_pcp, vlan=c_vlan_vid) / \
              scapy.IP(src=ip_src, dst=ip_dst, tos=ip_tos, ttl=ip_ttl, ihl=ip_ihl) / \
              scapy.UDP(sport=udp_sport, dport=udp_dport)
    else:
        if not ip_options:
            pkt = scapy.Ether(dst=eth_dst, src=eth_src)/ \
                scapy.IP(src=ip_src, dst=ip_dst, tos=ip_tos, ttl=ip_ttl, ihl=ip_ihl)/ \
                scapy.UDP(sport=udp_sport, dport=udp_dport)

        else:
            pkt = scapy.Ether(dst=eth_dst, src=eth_src) / \
                  scapy.IP(src=ip_src, dst=ip_dst, tos=ip_tos, ttl=ip_ttl, ihl=ip_ihl, options=ip_options) / \
                  scapy.UDP(sport=udp_sport, dport=udp_dport)

    pkt = pkt / ("D" * (pktlen - len(pkt)))

    return pkt

# Monkey-patch the oftest library to allow faking the data-plane
# This mode allows oftest to glide through the OF programming steps while
# passing all packet receiving tests. This is useful when you want to
# verify that your agent's OF interface handles all OF protocol messages
# from the controller, but in the absence of actual data-plane tests.
# In order to achieve this, in this mode the following chnages are made:
# - All verify_packet_in() tests pass
if fake_dataplane:
    logging.info('Monkey-patching for fake-dataplane operations')

    from oftest import testutils

    def patched_verify_packet_in(test, data, in_port, reason, controller=None):
        logging.debug('Faking verify_packet_in')
        msg = ofp.message.packet_in(reason=reason, data=data)
        return msg

    testutils.verify_packet_in = patched_verify_packet_in

    def patched_verify_packet(test, pkt, of_port):
        logging.debug('Faking verify_packet')
        return

    testutils.verify_packet = patched_verify_packet