tests/nicira_dec_ttl.py

import logging
import struct

from oftest import config
import of10 as ofp
import oftest.base_tests as base_tests

from oftest.testutils import *

class action_nx_dec_ttl(ofp.action.action_vendor):
    def __init__(self):
        ofp.action.action_vendor.__init__(self)
        self.vendor = 0x00002320

    def pack(self):
        return ofp.action.action_vendor.pack(self) + struct.pack("!HHL", 18, 0x0, 0x0)

    def __len__(self):
        return 16

    def show(self, prefix=''):
        return prefix + 'dec_ttl: ' + '\n' + ofp.action.action_vendor.show(self)

@nonstandard
class TtlDecrement(base_tests.SimpleDataPlane):
    def runTest(self):
        of_ports = config["port_map"].keys()
        of_ports.sort()
        self.assertTrue(len(of_ports) >= 3, "Not enough ports for test")
        portA = of_ports[0]
        portB = of_ports[1]
        portC = of_ports[2]

        outpkt = simple_tcp_packet(pktlen=100, ip_ttl=3)
        msg = ofp.message.packet_out(in_port=ofp.OFPP_NONE,
                                     data=str(outpkt),
                                     actions=[
                                         action_nx_dec_ttl(),
                                         ofp.action.output(port=portA),
                                         action_nx_dec_ttl(),
                                         ofp.action.output(port=portB),
                                         action_nx_dec_ttl(),
                                         ofp.action.output(port=portC)])
        self.controller.message_send(msg)

        receive_pkt_check(self.dataplane, simple_tcp_packet(ip_ttl=2), [portA], [], self)
        receive_pkt_check(self.dataplane, simple_tcp_packet(ip_ttl=1), [portB], [], self)
        receive_pkt_check(self.dataplane, simple_tcp_packet(ip_ttl=0), [], [portC], self)