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

   1 # Copyright (C) 2017 Nippon Telegraph and Telephone Corporation.
   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 from __future__ import print_function
  17
  18 import logging
  19 import os
  20 import sys
  21
  22 import unittest
  23 from nose.tools import eq_
  24 from nose.tools import ok_
  25
  26 from ryu.lib import pcaplib
  27 from ryu.lib.packet import openflow
  28 from ryu.lib.packet import packet
  29 from ryu.utils import binary_str
  30
  31
  32 LOG = logging.getLogger(__name__)
  33
  34 OPENFLOW_DATA_DIR = os.path.join(
  35     os.path.dirname(sys.modules[__name__].__file__),
  36     '../../packet_data/pcap/')
  37
  38
  39 class Test_openflow(unittest.TestCase):
  40     """
  41     Test case for ryu.lib.packet.openflow.
  42     """
  43
  44     def test_pcap(self):
  45         files = [
  46             'openflow_flowmod',
  47             'openflow_flowstats_req',
  48             'openflow_invalid_version',
  49         ]
  50
  51         for f in files:
  52             # print('*** testing %s ...' % f)
  53             for _, buf in pcaplib.Reader(
  54                     open(OPENFLOW_DATA_DIR + f + '.pcap', 'rb')):
  55                 # Checks if message can be parsed as expected.
  56                 pkt = packet.Packet(buf)
  57                 openflow_pkt = pkt.get_protocol(openflow.openflow)
  58                 ok_(isinstance(openflow_pkt, openflow.openflow),
  59                     'Failed to parse OpenFlow message: %s' % pkt)
  60
  61                 # Checks if message can be serialized as expected.
  62                 pkt.serialize()
  63                 eq_(buf, pkt.data,
  64                     "b'%s' != b'%s'" % (binary_str(buf), binary_str(pkt.data)))