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

   1 # Copyright (C) 2016 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 # vim: tabstop=4 shiftwidth=4 softtabstop=4
  17
  18 import functools
  19 import json
  20 import logging
  21 import os
  22 import sys
  23 import unittest
  24 try:
  25     import mock  # Python 2
  26 except ImportError:
  27     from unittest import mock  # Python 3
  28 from nose.tools import eq_
  29
  30 from ryu.app import ofctl_rest
  31 from ryu.app.wsgi import Request
  32 from ryu.app.wsgi import WSGIApplication
  33 from ryu.controller.dpset import DPSet
  34 from ryu.ofproto import ofproto_protocol
  35 from ryu.ofproto import ofproto_v1_0
  36 from ryu.ofproto import ofproto_v1_2
  37 from ryu.ofproto import ofproto_v1_3
  38 from ryu.ofproto import ofproto_v1_4
  39 from ryu.ofproto import ofproto_v1_5
  40 from ryu.tests import test_lib
  41
  42
  43 LOG = logging.getLogger(__name__)
  44
  45
  46 class DummyDatapath(ofproto_protocol.ProtocolDesc):
  47
  48     def __init__(self, version):
  49         super(DummyDatapath, self).__init__(version)
  50         self.id = 1
  51         _kw = {'port_no': 1, 'hw_addr': 'aa:bb:cc:dd:ee:ff',
  52                'name': 's1-eth1', 'config': 1, 'state': 1}
  53         # for OpenFlow 1.0
  54         if version in [ofproto_v1_0.OFP_VERSION]:
  55             _kw.update(
  56                 {'curr': 2112, 'advertised': 0, 'supported': 0, 'peer': 0})
  57             port_info = self.ofproto_parser.OFPPhyPort(**_kw)
  58         # for OpenFlow 1.2 or 1.3
  59         elif version in [ofproto_v1_2.OFP_VERSION, ofproto_v1_3.OFP_VERSION]:
  60             _kw.update(
  61                 {'curr': 2112, 'advertised': 0, 'supported': 0, 'peer': 0,
  62                  'curr_speed': 10000000, 'max_speed': 0})
  63             port_info = self.ofproto_parser.OFPPort(**_kw)
  64         # for OpenFlow 1.4+
  65         else:
  66             _kw.update({'properties': []})
  67             port_info = self.ofproto_parser.OFPPort(**_kw)
  68         self.ports = {1: port_info}
  69
  70
  71 class Test_ofctl_rest(unittest.TestCase):
  72
  73     def _test(self, name, dp, method, path, body):
  74         # print('processing %s ...' % name)
  75
  76         dpset = DPSet()
  77         dpset._register(dp)
  78         wsgi = WSGIApplication()
  79         contexts = {
  80             'dpset': dpset,
  81             'wsgi': wsgi,
  82         }
  83         ofctl_rest.RestStatsApi(**contexts)
  84
  85         req = Request.blank(path)
  86         req.body = json.dumps(body).encode('utf-8')
  87         req.method = method
  88
  89         with mock.patch('ryu.lib.ofctl_utils.send_stats_request'),\
  90                 mock.patch('ryu.lib.ofctl_utils.send_msg'):
  91             res = req.get_response(wsgi)
  92         eq_(res.status, '200 OK')
  93
  94
  95 def _add_tests():
  96     _ofp_vers = {
  97         'of10': ofproto_v1_0.OFP_VERSION,
  98         'of12': ofproto_v1_2.OFP_VERSION,
  99         'of13': ofproto_v1_3.OFP_VERSION,
 100         'of14': ofproto_v1_4.OFP_VERSION,
 101         'of15': ofproto_v1_5.OFP_VERSION,
 102     }
 103
 104     this_dir = os.path.dirname(sys.modules[__name__].__file__)
 105     ofctl_rest_json_dir = os.path.join(this_dir, 'ofctl_rest_json/')
 106
 107     for ofp_ver in _ofp_vers:
 108         # read a json file
 109         json_path = os.path.join(ofctl_rest_json_dir, ofp_ver + '.json')
 110         if os.path.exists(json_path):
 111             _test_cases = json.load(open(json_path))
 112         else:
 113             # print("Skip to load test cases for %s" % ofp_ver)
 114             continue
 115
 116         # add test
 117         for test in _test_cases:
 118             method = test['method']
 119             path = test['path']
 120             body = test.get('body', {})
 121
 122             name = 'test_ofctl_rest_' + method + '_' + ofp_ver + '_' + path
 123             # print('adding %s ...' % name)
 124             f = functools.partial(
 125                 Test_ofctl_rest._test,
 126                 name=name,
 127                 dp=DummyDatapath(_ofp_vers[ofp_ver]),
 128                 method=test['method'],
 129                 path=test['path'],
 130                 body=body
 131             )
 132             test_lib.add_method(Test_ofctl_rest, name, f)
 133
 134
 135 _add_tests()
 136
 137 if __name__ == "__main__":
 138     unittest.main()