- some tests need attached port to switch.
---
ryu/tests/integrated/test_request_reply_v12.py | 409 ++++++++++++++++++++++--
ryu/tests/integrated/tester.py | 12 +-
2 files changed, 397 insertions(+), 24 deletions(-)
diff --git a/ryu/tests/integrated/test_request_reply_v12.py
b/ryu/tests/integrated/test_request_reply_v12.py
index 15d4927..9de06b0 100644
--- a/ryu/tests/integrated/test_request_reply_v12.py
+++ b/ryu/tests/integrated/test_request_reply_v12.py
@@ -15,7 +15,7 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
-import sys
+import time
import logging
from ryu.controller import ofp_event
@@ -29,7 +29,13 @@ LOG = logging.getLogger(__name__)
class RunTest(tester.TestFlowBase):
""" Test case for Request-Reply messages.
+
+ Some tests need attached port to switch.
+ If use the OVS, can do it with the following commands.
+ # ip link add <port> type dummy
+ # ovs-vsctl add-port <bridge> <port>
"""
+
OFP_VERSIONS = [ofproto_v1_2.OFP_VERSION]
def __init__(self, *args, **kwargs):
@@ -61,10 +67,7 @@ class RunTest(tester.TestFlowBase):
self.unclear -= 1
self.start_next_test(dp)
else:
- LOG.info("TEST_RESULTS:")
- for t, r in self.results.items():
- LOG.info(" %s: %s", t, r)
- LOG.info(tester.LOG_TEST_FINISH, self.unclear == 0)
+ self.print_results()
def run_verify(self, ev):
msg = ev.msg
@@ -84,7 +87,6 @@ class RunTest(tester.TestFlowBase):
def verify_default(self, dp, msg):
type_ = self._verify
- self._verify = None
if msg.msg_type == dp.ofproto.OFPT_STATS_REPLY:
return self.verify_stats(dp, msg.body, type_)
@@ -98,7 +100,6 @@ class RunTest(tester.TestFlowBase):
stats_types = dp.ofproto_parser.OFPStatsReply._STATS_TYPES
expect = stats_types.get(type_).__name__
- # LOG.debug(stats)
if isinstance(stats, list):
for s in stats:
if expect == s.__class__.__name__:
@@ -144,6 +145,14 @@ class RunTest(tester.TestFlowBase):
dp.send_msg(m)
+ def get_port(self, dp):
+ p = None
+ for port_no, port in dp.ports.items():
+ if port_no != dp.ofproto.OFPP_LOCAL:
+ p = port
+ break
+ return p
+
# Test for Reply message type
def test_desc_stats_request(self, dp):
self._verify = dp.ofproto.OFPST_DESC
@@ -192,6 +201,17 @@ class RunTest(tester.TestFlowBase):
self._verify = dp.ofproto.OFPT_BARRIER_REPLY
dp.send_barrier()
+ def test_error_reply(self, dp):
+ ports = [0]
+ for p in dp.ports:
+ if p != dp.ofproto.OFPP_LOCAL:
+ ports.append(p)
+
+ port_no = max(ports) + 1
+ self._verify = dp.ofproto.OFPT_ERROR
+ m = dp.ofproto_parser.OFPPortMod(dp, port_no, '\xff' * 6, 0, 0, 0)
+ dp.send_msg(m)
+
# Test for reply value
def test_flow_stats_none(self, dp):
self.send_flow_stats(dp)
@@ -216,18 +236,15 @@ class RunTest(tester.TestFlowBase):
self.send_flow_stats(dp)
def verify_flow_stats_reply_value(self, dp, msg):
- flows = msg.body
- verify = self._verify
- self._verify = None
c = 0
- for f in flows:
+ for f in msg.body:
f_value = (f.table_id, f.cookie, f.idle_timeout,
f.hard_timeout, f.priority, )
- if f_value != verify[c]:
- return 'Value error: send %s, flow %s' \
- % (verify[c], f_value,)
+ if f_value != self._verify[c]:
+ return 'param is mismatched. verify=%s, reply=%s' \
+ % (self._verify[c], f_value,)
c += 1
- return len(flows) == self.n_tables
+ return len(msg.body) == self.n_tables
def test_echo_request_has_data(self, dp):
data = 'test'
@@ -275,6 +292,39 @@ class RunTest(tester.TestFlowBase):
stats = msg.body
return stats.flow_count == 1
+ def test_aggregate_stats_packet_count(self, dp):
+ in_port = 1
+ data = 'test'
+ self._verify = {'packet_count': 1,
+ 'byte_count': len(data)}
+
+ # add flow
+ match = dp.ofproto_parser.OFPMatch()
+ match.set_in_port(in_port)
+ self.mod_flow(dp, table_id=0, match=match)
+
+ # packet out
+ output = dp.ofproto.OFPP_TABLE
+ actions = [dp.ofproto_parser.OFPActionOutput(output, 0)]
+ m = dp.ofproto_parser.OFPPacketOut(dp, 0xffffffff, in_port,
+ actions, data)
+ dp.send_msg(m)
+ dp.send_barrier()
+
+ match = dp.ofproto_parser.OFPMatch()
+ m = dp.ofproto_parser.OFPAggregateStatsRequest(
+ dp, dp.ofproto.OFPTT_ALL, dp.ofproto.OFPP_ANY,
+ dp.ofproto.OFPG_ANY, 0, 0, match)
+ dp.send_msg(m)
+
+ def verify_aggregate_stats_packet_count(self, dp, msg):
+ for name, val in self._verify.items():
+ r_val = getattr(msg.body, name)
+ if val != r_val:
+ return '%s is mismatched. verify=%s, reply=%s' \
+ % (name, val, r_val)
+ return True
+
def test_set_config_nomal(self, dp):
flags = dp.ofproto.OFPC_FRAG_NORMAL
self._verify = flags
@@ -405,7 +455,7 @@ class RunTest(tester.TestFlowBase):
verify = self._verify
if len(verify) != len(stats):
- return 'flow count mismatched. verify=%s stats=%s' \
+ return 'flow_count is mismatched. verify=%s stats=%s' \
% (len(verify), len(stats))
for s in stats:
@@ -417,7 +467,7 @@ class RunTest(tester.TestFlowBase):
s_port = s.instructions[0].actions[0].port
if v_port != s_port:
- return 'port mismatched table_id=%s verify=%s, stats=%s' \
+ return 'port is mismatched. table_id=%s verify=%s, stats=%s' \
% (s.table_id, v_port, s_port)
return True
@@ -654,6 +704,301 @@ class RunTest(tester.TestFlowBase):
def verify_flow_del_strict(self, dp, msg):
return self._verify_flow_value(dp, msg)
+ def _send_port_mod(self, dp, config, mask):
+ p = self.get_port(dp)
+ if not p:
+ err = 'need attached port to switch.'
+ self.results[self.current] = err
+ self.start_next_test(dp)
+ return
+
+ self._verify = [p.port_no, config & mask]
+ m = dp.ofproto_parser.OFPPortMod(dp, p.port_no, p.hw_addr,
+ config, mask, 0)
+ dp.send_msg(m)
+ dp.send_barrier()
+
+ # TODO: waiting to port UP|DOWN.
+ time.sleep(1)
+ m = dp.ofproto_parser.OFPFeaturesRequest(dp)
+ dp.send_msg(m)
+
+ def _verify_port_mod_config(self, dp, msg):
+ port_no = self._verify[0]
+ config = self._verify[1]
+
+ port = msg.ports[port_no]
+ if config != port.config:
+ return "config is mismatched. verify=%s, stats=%s" \
+ % (bin(config), bin(port.config))
+ return True
+
+ def test_port_mod_config_01_all(self, dp):
+ config = 0b1100101
+ mask = 0b1111111
+ self._send_port_mod(dp, config, mask)
+
+ def verify_port_mod_config_01_all(self, dp, msg):
+ return self._verify_port_mod_config(dp, msg)
+
+ def test_port_mod_config_02_none(self, dp):
+ config = 0
+ mask = 0b1111111
+ self._send_port_mod(dp, config, mask)
+
+ def verify_port_mod_config_02_none(self, dp, msg):
+ return self._verify_port_mod_config(dp, msg)
+
+ def test_port_mod_config_03_mask(self, dp):
+ config = 0b1100101
+ mask = 0b1111000
+ self._send_port_mod(dp, config, mask)
+
+ def verify_port_mod_config_03_mask(self, dp, msg):
+ res = self._verify_port_mod_config(dp, msg)
+ # reset port config
+ port_no = self._verify[0]
+ p = msg.ports[port_no]
+ m = dp.ofproto_parser.OFPPortMod(dp, p.port_no, p.hw_addr,
+ 0, 0b1111111, 0)
+ dp.send_msg(m)
+ dp.send_barrier()
+ return res
+
+ def test_port_stats_port_no(self, dp):
+ p = self.get_port(dp)
+ if not p:
+ err = 'need attached port to switch.'
+ self.results[self.current] = err
+ self.start_next_test(dp)
+ return
+
+ self._verify = p.port_no
+ m = dp.ofproto_parser.OFPPortStatsRequest(dp, p.port_no)
+ dp.send_msg(m)
+
+ def verify_port_stats_port_no(self, dp, msg):
+ ports = msg.body
+ if len(ports) > 1:
+ return 'reply some ports.\n%s' % (ports)
+
+ if ports[0].port_no != self._verify:
+ return 'port_no is mismatched. request=%s reply=%s' \
+ % (self._verify, ports[0].port_no)
+
+ return True
+
+ def _add_flow_flow_removed(self, dp, reason, table_id=0,
+ cookie=0xff, priority=100, in_port=1,
+ idle_timeout=0, hard_timeout=0):
+ self._verify = {}
+ self._verify['params'] = {'reason': reason,
+ 'table_id': table_id,
+ 'cookie': cookie,
+ 'priority': priority}
+ self._verify['in_port'] = in_port
+ self._verify['timeout'] = idle_timeout
+ if hard_timeout:
+ if (idle_timeout == 0 or idle_timeout > hard_timeout):
+ self._verify['timeout'] = hard_timeout
+
+ match = dp.ofproto_parser.OFPMatch()
+ match.set_in_port(in_port)
+ self.mod_flow(dp, match=match, cookie=cookie,
+ priority=priority, table_id=table_id,
+ idle_timeout=idle_timeout, hard_timeout=hard_timeout,
+ flags=dp.ofproto.OFPFF_SEND_FLOW_REM)
+
+ def _verify_flow_removed(self, dp, msg):
+ params = self._verify['params']
+ in_port = self._verify['in_port']
+ timeout = self._verify['timeout']
+
+ if timeout:
+ duration_nsec = (msg.duration_sec * 10 ** 9) + msg.duration_nsec
+ timeout_nsec = timeout * 10 ** 9
+
+ # grace of 1.5 second to timeout.
+ l = timeout * 10 ** 9
+ h = (timeout + 1.5) * 10 ** 9
+ if not l < duration_nsec < h:
+ return 'bad duration time. set=%s(nsec), duration=%s(nsec)' \
+ % (timeout_nsec, duration_nsec)
+
+ for name, val in params.items():
+ r_val = getattr(msg, name)
+ if val != r_val:
+ return '%s is mismatched. verify=%s, reply=%s' \
+ % (name, val, r_val)
+
+ for f in msg.match.fields:
+ if f.header == ofproto_v1_2.OXM_OF_IN_PORT:
+ if f.value != in_port:
+ return 'in_port is mismatched. verify=%s, reply=%s' \
+ % (in_port, f.value)
+ return True
+
+ def test_flow_removed_idle_timeout(self, dp):
+ reason = dp.ofproto.OFPRR_IDLE_TIMEOUT
+ idle_timeout = 2
+ self._add_flow_flow_removed(dp, reason,
+ idle_timeout=idle_timeout)
+
+ def verify_flow_removed_idle_timeout(self, dp, msg):
+ return self._verify_flow_removed(dp, msg)
+
+ def test_flow_removed_idle_timeout_hit(self, dp):
+ reason = dp.ofproto.OFPRR_IDLE_TIMEOUT
+ idle_timeout = 5
+ in_port = 1
+ sleep = 2
+
+ # add target flow
+ self._add_flow_flow_removed(dp, reason, in_port=in_port,
+ idle_timeout=idle_timeout)
+ self._verify['timeout'] = idle_timeout + sleep
+
+ # sleep
+ time.sleep(sleep)
+
+ # packet out
+ output = dp.ofproto.OFPP_TABLE
+ actions = [dp.ofproto_parser.OFPActionOutput(output, 0)]
+ m = dp.ofproto_parser.OFPPacketOut(dp, 0xffffffff, in_port,
+ actions, None)
+ dp.send_msg(m)
+
+ def verify_flow_removed_idle_timeout_hit(self, dp, msg):
+ return self._verify_flow_removed(dp, msg)
+
+ def test_flow_removed_hard_timeout(self, dp):
+ reason = dp.ofproto.OFPRR_HARD_TIMEOUT
+ hard_timeout = 2
+ self._add_flow_flow_removed(dp, reason,
+ hard_timeout=hard_timeout)
+
+ def verify_flow_removed_hard_timeout(self, dp, msg):
+ return self._verify_flow_removed(dp, msg)
+
+ def test_flow_removed_hard_timeout_hit(self, dp):
+ reason = dp.ofproto.OFPRR_HARD_TIMEOUT
+ hard_timeout = 5
+ in_port = 1
+ sleep = 2
+
+ self._add_flow_flow_removed(dp, reason, in_port=in_port,
+ hard_timeout=hard_timeout)
+ dp.send_barrier()
+
+ # sleep
+ time.sleep(sleep)
+
+ # packet out
+ output = dp.ofproto.OFPP_TABLE
+ actions = [dp.ofproto_parser.OFPActionOutput(output, 0)]
+ m = dp.ofproto_parser.OFPPacketOut(dp, 0xffffffff, in_port,
+ actions, None)
+ dp.send_msg(m)
+
+ def verify_flow_removed_hard_timeout_hit(self, dp, msg):
+ return self._verify_flow_removed(dp, msg)
+
+ def test_flow_removed_delete(self, dp):
+ reason = dp.ofproto.OFPRR_DELETE
+ self._add_flow_flow_removed(dp, reason)
+ dp.send_barrier()
+ self.delete_all_flows(dp)
+
+ def verify_flow_removed_delete(self, dp, msg):
+ return self._verify_flow_removed(dp, msg)
+
+ def test_flow_removed_table_id(self, dp):
+ reason = dp.ofproto.OFPRR_DELETE
+ table_id = 1
+ self._add_flow_flow_removed(dp, reason, table_id=table_id)
+ dp.send_barrier()
+ self.delete_all_flows(dp)
+
+ def verify_flow_removed_table_id(self, dp, msg):
+ return self._verify_flow_removed(dp, msg)
+
+ def _send_packet_out(self, dp, buffer_id=0xffffffff,
+ in_port=None, output=None, data=''):
+ if in_port is None:
+ in_port = dp.ofproto.OFPP_LOCAL
+
+ if output is None:
+ output = dp.ofproto.OFPP_CONTROLLER
+
+ self._verify['buffer_id'] = buffer_id
+ self._verify['in_port'] = in_port
+ self._verify['data'] = data
+
+ actions = [dp.ofproto_parser.OFPActionOutput(output, len(data))]
+ m = dp.ofproto_parser.OFPPacketOut(dp, buffer_id, in_port,
+ actions, data)
+ dp.send_msg(m)
+
+ def _verify_packet_in(self, dp, msg):
+ for name, val in self._verify.items():
+ if name == 'in_port':
+ for f in msg.match.fields:
+ if f.header == ofproto_v1_2.OXM_OF_IN_PORT:
+ r_val = f.value
+ else:
+ r_val = getattr(msg, name)
+
+ if val != r_val:
+ return '%s is mismatched. verify=%s, reply=%s' \
+ % (name, val, r_val)
+ return True
+
+ def test_packet_in_action(self, dp):
+ self._verify = {}
+ self._verify['reason'] = dp.ofproto.OFPR_ACTION
+ self._send_packet_out(dp)
+
+ def verify_packet_in_action(self, dp, msg):
+ return self._verify_packet_in(dp, msg)
+
+ def test_packet_in_data(self, dp):
+ self._verify = {}
+ self._verify['reason'] = dp.ofproto.OFPR_ACTION
+ data = 'test'
+ self._send_packet_out(dp, data=data)
+
+ def verify_packet_in_data(self, dp, msg):
+ return self._verify_packet_in(dp, msg)
+
+ def test_packet_in_table_id(self, dp):
+ in_port = 1
+ table_id = 2
+ output = dp.ofproto.OFPP_TABLE
+
+ self._verify = {}
+ self._verify['reason'] = dp.ofproto.OFPR_ACTION
+ self._verify['table_id'] = table_id
+
+ # add flow (goto_table)
+ match = dp.ofproto_parser.OFPMatch()
+ match.set_in_port(in_port)
+ inst = [dp.ofproto_parser.OFPInstructionGotoTable(table_id)]
+ self.mod_flow(dp, inst=inst, match=match)
+
+ # add flow (output)
+ match = dp.ofproto_parser.OFPMatch()
+ match.set_in_port(in_port)
+ out = dp.ofproto.OFPP_CONTROLLER
+ actions = [dp.ofproto_parser.OFPActionOutput(out, 0)]
+ self.mod_flow(dp, actions=actions, match=match, table_id=table_id)
+ dp.send_barrier()
+
+ # packet out
+ self._send_packet_out(dp, in_port=in_port, output=output)
+
+ def verify_packet_in_table_id(self, dp, msg):
+ return self._verify_packet_in(dp, msg)
+
# handler
@set_ev_cls(ofp_event.EventOFPEchoReply, MAIN_DISPATCHER)
def echo_replay_handler(self, ev):
@@ -692,6 +1037,25 @@ class RunTest(tester.TestFlowBase):
if self.current == 'test_barrier_request':
self.run_verify(ev)
+ @set_ev_cls(ofp_event.EventOFPPortStatus, MAIN_DISPATCHER)
+ def port_status_handler(self, ev):
+ pass
+
+ @set_ev_cls(ofp_event.EventOFPPacketIn, MAIN_DISPATCHER)
+ def packet_in_handler(self, ev):
+ if self.current.find('packet_in'):
+ self.run_verify(ev)
+
+ @set_ev_cls(ofp_event.EventOFPFlowRemoved, MAIN_DISPATCHER)
+ def flow_removed_handler(self, ev):
+ if self.current.find('flow_removed') > 0:
+ self.run_verify(ev)
+
+ @set_ev_cls(ofp_event.EventOFPErrorMsg, MAIN_DISPATCHER)
+ def error_handler(self, ev):
+ if self.current.find('error') > 0:
+ self.run_verify(ev)
+
def get_supported(self, dp):
if self.capabilities is None:
m = dp.ofproto_parser.OFPFeaturesRequest(dp)
@@ -706,7 +1070,12 @@ class RunTest(tester.TestFlowBase):
self.start_next_test(dp)
def is_supported(self, t):
- # TODO: run only test of supported capabilities.
- if t.find('out_port') > 0:
- return False
+ unsupported = [
+ 'out_port',
+ 'flow_removed_table_id',
+ ]
+ for u in unsupported:
+ if t.find(u) > 0:
+ return False
+
return True
diff --git a/ryu/tests/integrated/tester.py b/ryu/tests/integrated/tester.py
index daf60e1..5264b78 100644
--- a/ryu/tests/integrated/tester.py
+++ b/ryu/tests/integrated/tester.py
@@ -61,6 +61,7 @@ class TestFlowBase(app_manager.RyuApp):
for t in dir(self):
if t.startswith("test_"):
self.pending.append(t)
+ self.pending.sort(reverse=True)
self.unclear = len(self.pending)
def delete_all_flows(self, dp):
@@ -113,10 +114,13 @@ class TestFlowBase(app_manager.RyuApp):
dp.send_barrier()
self.send_flow_stats(dp)
else:
- LOG.info("TEST_RESULTS:")
- for t, r in self.results.items():
- LOG.info(" %s: %s", t, r)
- LOG.info(LOG_TEST_FINISH, self.unclear == 0)
+ self.print_results()
+
+ def print_results(self):
+ LOG.info("TEST_RESULTS:")
+ for t in sorted(self.results.keys()):
+ LOG.info(" %s: %s", t, self.results[t])
+ LOG.info(LOG_TEST_FINISH, self.unclear == 0)
@handler.set_ev_cls(ofp_event.EventOFPFlowStatsReply,
handler.MAIN_DISPATCHER)
--
1.7.9.5
------------------------------------------------------------------------------
Live Security Virtual Conference
Exclusive live event will cover all the ways today's security and
threat landscape has changed and how IT managers can respond. Discussions
will include endpoint security, mobile security and the latest in malware
threats. http://www.accelacomm.com/jaw/sfrnl04242012/114/50122263/
_______________________________________________
Ryu-devel mailing list
[email protected]
https://lists.sourceforge.net/lists/listinfo/ryu-devel
