openstack/os-ken
Code Issues Proposed changes

test_controller: Add unit test for _recv_loop()

Browse Source 
Signed-off-by: IWASE Yusuke <iwase.yusuke0@gmail.com>
Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>
This commit is contained in:
IWASE Yusuke 2015-12-04 14:56:55 +09:00 committed by FUJITA Tomonori
parent 0a223481dd
commit 106e88f77a
1 changed files with 72 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
ryu/tests/unit/controller/test_controller.py
View File

@ -1,3 +1,4 @@
# Copyright (C) 2015 Nippon Telegraph and Telephone Corporation.
# Copyright (C) 2015 Stratosphere Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
@ -19,16 +20,17 @@ try:
except ImportError:
from unittest import mock # Python 3
import json
import os
import sys
import warnings
import unittest
import logging
import nose
from nose.tools import assert_equal
from nose.tools import assert_true
import random
from ryu.base import app_manager # To suppress cyclic import
from ryu.controller import controller
from ryu.controller import handler
from ryu.ofproto import ofproto_v1_3_parser
from ryu.ofproto import ofproto_v1_2_parser
from ryu.ofproto import ofproto_v1_0_parser
@ -38,8 +40,8 @@ LOG = logging.getLogger('test_controller')
class Test_Datapath(unittest.TestCase):
""" Test case for Datapath
"""
Test cases for controller.Datapath
"""
def _test_ports_accessibility(self, ofproto_parser, msgs_len):
@ -65,9 +67,9 @@ class Test_Datapath(unittest.TestCase):
# Read & Delete
del dp.ports[0]
assert_equal(len(msgs), msgs_len)
self.assertEqual(len(msgs), msgs_len)
for msg in msgs:
assert_true(issubclass(msg.category, UserWarning))
self.assertTrue(issubclass(msg.category, UserWarning))
def test_ports_accessibility_v13(self):
self._test_ports_accessibility(ofproto_v1_3_parser, 2)
@ -78,6 +80,66 @@ class Test_Datapath(unittest.TestCase):
def test_ports_accessibility_v10(self):
self._test_ports_accessibility(ofproto_v1_0_parser, 0)
@mock.patch("ryu.base.app_manager", spec=app_manager)
def test_recv_loop(self, app_manager_mock):
# Prepare test data
test_messages = [
"4-6-ofp_features_reply.packet",
"4-14-ofp_echo_reply.packet",
"4-14-ofp_echo_reply.packet",
"4-4-ofp_packet_in.packet",
"4-14-ofp_echo_reply.packet",
"4-14-ofp_echo_reply.packet",
]
this_dir = os.path.dirname(sys.modules[__name__].__file__)
packet_data_dir = os.path.join(this_dir, '../../packet_data/of13')
json_dir = os.path.join(this_dir, '../ofproto/json/of13')
if __name__ == '__main__':
nose.main(argv=['nosetests', '-s', '-v'], defaultTest=__file__)
packet_buf = bytearray()
expected_json = list()
for msg in test_messages:
# Construct the received packet buffer as one packet data in order
# to test the case of the OpenFlow messages composed in one packet.
packet_data_file = os.path.join(packet_data_dir, msg)
packet_buf += open(packet_data_file, 'rb').read()
json_data_file = os.path.join(json_dir, msg + '.json')
expected_json.append(json.load(open(json_data_file)))
# Prepare mock for socket
class SocketMock(mock.MagicMock):
buf = bytearray()
random = None
def recv(self, bufsize):
size = self.random.randint(1, bufsize)
out = self.buf[:size]
self.buf = self.buf[size:]
return out
# Prepare mock
ofp_brick_mock = mock.MagicMock(spec=app_manager.RyuApp)
app_manager_mock.lookup_service_brick.return_value = ofp_brick_mock
sock_mock = SocketMock()
sock_mock.buf = packet_buf
sock_mock.random = random.Random('Ryu SDN Framework')
addr_mock = mock.MagicMock()
# Prepare test target
dp = controller.Datapath(sock_mock, addr_mock)
dp.set_state(handler.MAIN_DISPATCHER)
ofp_brick_mock.reset_mock()
# Test
dp._recv_loop()
# Assert calls
output_json = list()
for call in ofp_brick_mock.send_event_to_observers.call_args_list:
args, kwargs = call
ev, state = args
if not hasattr(ev, 'msg'):
continue
output_json.append(ev.msg.to_jsondict())
self.assertEqual(state, handler.MAIN_DISPATCHER)
self.assertEqual(kwargs, {})
self.assertEqual(expected_json, output_json)