ryu/build/lib.linux-armv7l-2.7/ryu/tests/unit/lib/test_ofp_pktinfilter.py

   1 # Copyright (C) 2013 Stratosphere Inc.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #    http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 # vim: tabstop=4 shiftwidth=4 softtabstop=4
  17
  18 import unittest
  19 import logging
  20 import six
  21
  22 from nose.tools import *
  23
  24 from ryu.controller import ofp_event
  25 from ryu.controller.handler import (
  26     set_ev_cls,
  27     MAIN_DISPATCHER,
  28 )
  29 from ryu.lib.packet import vlan, ethernet, ipv4
  30 from ryu.lib.ofp_pktinfilter import packet_in_filter, RequiredTypeFilter
  31 from ryu.lib import mac
  32 from ryu.ofproto import ether, ofproto_v1_3, ofproto_v1_3_parser
  33 from ryu.ofproto.ofproto_protocol import ProtocolDesc
  34
  35
  36 LOG = logging.getLogger('test_pktinfilter')
  37
  38
  39 class _PacketInFilterApp(object):
  40     @set_ev_cls(ofp_event.EventOFPPacketIn, MAIN_DISPATCHER)
  41     @packet_in_filter(RequiredTypeFilter, {'types': [
  42         vlan.vlan,
  43     ]})
  44     def packet_in_handler(self, ev):
  45         return True
  46
  47
  48 class Test_packet_in_filter(unittest.TestCase):
  49
  50     """ Test case for pktinfilter
  51     """
  52
  53     def setUp(self):
  54         self.app = _PacketInFilterApp()
  55
  56     def tearDown(self):
  57         pass
  58
  59     def test_pkt_in_filter_pass(self):
  60         datapath = ProtocolDesc(version=ofproto_v1_3.OFP_VERSION)
  61         e = ethernet.ethernet(mac.BROADCAST_STR,
  62                               mac.BROADCAST_STR,
  63                               ether.ETH_TYPE_8021Q)
  64         v = vlan.vlan()
  65         i = ipv4.ipv4()
  66         pkt = (e / v / i)
  67         pkt.serialize()
  68         pkt_in = ofproto_v1_3_parser.OFPPacketIn(datapath,
  69                                                  data=six.binary_type(pkt.data))
  70         ev = ofp_event.EventOFPPacketIn(pkt_in)
  71         ok_(self.app.packet_in_handler(ev))
  72
  73     def test_pkt_in_filter_discard(self):
  74         datapath = ProtocolDesc(version=ofproto_v1_3.OFP_VERSION)
  75         e = ethernet.ethernet(mac.BROADCAST_STR,
  76                               mac.BROADCAST_STR,
  77                               ether.ETH_TYPE_IP)
  78         i = ipv4.ipv4()
  79         pkt = (e / i)
  80         pkt.serialize()
  81         pkt_in = ofproto_v1_3_parser.OFPPacketIn(datapath,
  82                                                  data=six.binary_type(pkt.data))
  83         ev = ofp_event.EventOFPPacketIn(pkt_in)
  84         ok_(not self.app.packet_in_handler(ev))
  85
  86     def test_pkt_in_filter_truncated(self):
  87         datapath = ProtocolDesc(version=ofproto_v1_3.OFP_VERSION)
  88         truncated_data = ''
  89         pkt_in = ofproto_v1_3_parser.OFPPacketIn(datapath,
  90                                                  data=truncated_data)
  91         ev = ofp_event.EventOFPPacketIn(pkt_in)
  92         ok_(not self.app.packet_in_handler(ev))