Completed 2nd complex testcase with 2 ONUs
For details of the two completed scenarios, please read the Python
file (see class doc strings).
Change-Id: I84edc5c26be3816e9b670a5c97feb072286f0070
Use flow_delete_strict and fake dataplane mode
Two changes in this request:
1. I noticed that the installEapolRule() method in install=False mode
uses the flow_delete message instead of the flow_delete_strict.
However, the way the message is constructed, this should never have
the intended effect, because the out_port is left to 0. Per the
OpenFlow spec, this will match only flows that has the output
action pointing to port 0. The flow_delete would work only if
out_port is set to OFPP_ANY (4294967295). Alternatively, the
flow_delete_strict should be used, in which case the out_port
attribute is ignored. I decided that the latter makes more sense
in this use-case.
2. Introduced a "fake-dataplane" mode in which mode all verify_packet
and verify_packet_in calls automatically pass. This allows the
tests to manage flows and groups on an OF agent, without actually
testing packet forwarding, which is useful if the goal is to
review what flow command the OLT use-case rely on.
Change-Id: I3d39953c39d39a386802c44c5a84aac7560f9c09
Create a pair of isolated Eapol tests
These are extracts from the complex usecase. They differ from the
olt.py Eapol out/in tests in that they really target the ONU port
and not all ports, hence they are more realistic. The main reason
I created them is that on the Tibit OLT Eapol forwarding is not
setup on the OLT port, only on the ONU ports. This is correct behavior
but the original tests would fail with it.
Change-Id: I340c5798c806277ce51b758b6a35f5c9b560a91e
diff --git a/olt-complex.py b/olt-complex.py
index db8f33f..e8ea2e9 100644
--- a/olt-complex.py
+++ b/olt-complex.py
@@ -14,6 +14,19 @@
self.__dict__.update(kws)
+class EapolInOnly(OltBaseTest):
+ def runTest(self):
+ self.resetOlt()
+ self.installEapolRule()
+ self.sendEapolIn()
+
+
+class EapolOutOnly(OltBaseTest):
+ def runTest(self):
+ self.resetOlt()
+ self.sendEapolOut()
+
+
class ControllerAccess(OltBaseTest):
"""Verify openflow access from OLT device"""
@@ -64,7 +77,7 @@
logging.info("Running %s" % self.__class__.__name__)
# Some constants
- c_vlan_id = 111
+ s_vlan_id = 111
mcast_vlan_id = 140
mcast_groups = Thing(
ch1="230.10.10.10",
@@ -77,7 +90,7 @@
port=onu_port,
ip="13.14.14.13",
mac="b6:b8:3e:fb:1a:3f",
- s_vlan_id=13
+ c_vlan_id=13
)
# 0. Reset the OLT into a clean state
@@ -89,19 +102,19 @@
self.sendEapolIn()
# 2. Setup the OLT to pass unicast traffic for the ONU1
- self.installDoubleTaggingRules(onu1.s_vlan_id, c_vlan_id, self.getCookieBlock())
+ self.installDoubleTaggingRules(s_vlan_id, onu1.c_vlan_id, self.getCookieBlock())
# 3. Verify that unicast traffic works in both directions (use a few hundred frames in both ways)
- self.testPacketFlow(onu1.s_vlan_id, c_vlan_id) # this tests just one packet in each way
- self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 100)
+ self.testPacketFlow(s_vlan_id, onu1.c_vlan_id) # this tests just one packet in each way
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 100)
# 4. Setup IGMP forwarding toward the controller
self.installIgmpRule(self.getCookieBlock())
- self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
# 5. Send periodic IGMP queries out toward the ONU and verify its arrival
self.testIgmpQueryOut()
- self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
# 6. Send in an IGMP join request for one channel (specific multicast address) and verify that
# the controller receives it.
@@ -116,7 +129,7 @@
self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port], 10)
# 9. Verify that bidirectional unicast traffic still works across the PON
- self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
# 10. Change channel to a new multicast group and verify both multicast receiption as well as
# unicast traffic works. This step involves:
@@ -134,7 +147,7 @@
self.setupMcastChannel(mcast_groups.ch2, mcast_vlan_id, [onu1.port], ch2_group_id, ch2_cookie)
self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port], expect_to_be_blocked=True)
self.testMcastFlow(mcast_groups.ch2, mcast_vlan_id, [onu1.port], 10)
- self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
# 11. Add a second channel while keeping the existing one. Verify all what needs to be verified
# (similar as above)
@@ -145,7 +158,7 @@
self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port], expect_to_be_blocked=True)
self.testMcastFlow(mcast_groups.ch2, mcast_vlan_id, [onu1.port], 10)
self.testMcastFlow(mcast_groups.ch3, mcast_vlan_id, [onu1.port], 10)
- self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
# 12. Add two more multicast channels, and verify everything again
self.sendIgmpReport(join=[mcast_groups.ch2, mcast_groups.ch3, mcast_groups.ch4, mcast_groups.ch5])
@@ -160,7 +173,7 @@
self.testMcastFlow(mcast_groups.ch3, mcast_vlan_id, [onu1.port], 10)
self.testMcastFlow(mcast_groups.ch4, mcast_vlan_id, [onu1.port], 10)
self.testMcastFlow(mcast_groups.ch5, mcast_vlan_id, [onu1.port], 10)
- self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
# 13. Flip a channel for one of the multicast channels, and verify everything (ch3 -> ch1)
self.sendIgmpReport(join=[mcast_groups.ch1, mcast_groups.ch2, mcast_groups.ch4, mcast_groups.ch5],
@@ -173,7 +186,7 @@
self.testMcastFlow(mcast_groups.ch3, mcast_vlan_id, [onu1.port], expect_to_be_blocked=True)
self.testMcastFlow(mcast_groups.ch4, mcast_vlan_id, [onu1.port], 10)
self.testMcastFlow(mcast_groups.ch5, mcast_vlan_id, [onu1.port], 10)
- self.testSustainedPacketFlow(onu1.s_vlan_id, c_vlan_id, 10) # to check that unicast still flows
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
# 14. Tear down the test.
self.resetOlt()
@@ -243,4 +256,187 @@
"""
- # TODO
\ No newline at end of file
+ def runTest(self):
+ logging.info("Running %s" % self.__class__.__name__)
+
+ # Some constants
+ s_vlan_id = 111
+ mcast_vlan_id = 140
+ mcast_groups = Thing(
+ ch1="230.10.10.10",
+ ch2="231.11.11.11",
+ ch3="232.12.12.12",
+ ch4="233.13.13.13",
+ ch5="234.14.14.14"
+ )
+ onu1 = Thing(
+ port=onu_port,
+ ip="13.14.14.13",
+ mac="b6:b8:3e:fb:1a:3f",
+ c_vlan_id=13
+ )
+ onu2 = Thing(
+ port=onu_port2,
+ ip="113.114.114.113",
+ mac="b6:b8:3e:fb:aa:aa",
+ c_vlan_id=913
+ )
+
+ # 0. Reset the OLT into a clean state
+ self.resetOlt()
+
+ # 1. Setup the OLT to pass authentication (EAPOL) to controller from ONU1
+ # and test it.
+ self.installEapolRule()
+ self.sendEapolIn()
+
+ # 2. Setup the OLT to pass unicast traffic for the ONU1
+ self.installDoubleTaggingRules(s_vlan_id, onu1.c_vlan_id, self.getCookieBlock())
+
+ # 3. Verify that unicast traffic works in both directions (use a few hundred frames in both ways)
+ self.testPacketFlow(s_vlan_id, onu1.c_vlan_id) # this tests just one packet in each way
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 100)
+
+ # 4. Setup IGMP forwarding toward the controller
+ self.installIgmpRule(self.getCookieBlock())
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
+
+ # 5. Send periodic IGMP queries out toward the ONU1 and verify its arrival
+ self.testIgmpQueryOut()
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
+
+ # 6. Send in an IGMP join request for one channel (specific multicast address) and verify that
+ # the controller receives it.
+ self.sendIgmpReport(join=[mcast_groups.ch1])
+
+ # 7. Setup flows for forwarding multicast traffic from the OLT port toward ONU1
+ ch1_cookie = self.getCookieBlock()
+ ch1_group_id = 1
+ self.setupMcastChannel(mcast_groups.ch1, mcast_vlan_id, [onu1.port], ch1_group_id, ch1_cookie)
+
+ # 8. Verify that multicast packets can reach ONU1
+ self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port], 10)
+
+ # 9. Verify that bidirectional unicast traffic still works across the PON
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
+
+ # 10. Change channel to a new multicast group and verify both multicast receiption as well as
+ # unicast traffic works. This step involves:
+ # - sending a leave message by the ONU to the controller and verify its reception
+ # - sending a join message and verify its reception
+ # - removing the existing multicast flow
+ # - adding a new multicast flow
+ # - verify that original flow no longer flows
+ # - verify new flow
+ # - verify that unicast still works
+ self.sendIgmpReport(leave=[mcast_groups.ch1], join=[mcast_groups.ch2])
+ self.removeMcastChannel(mcast_groups.ch1, mcast_vlan_id, [onu1.port], ch1_group_id, ch1_cookie)
+ ch2_cookie = self.getCookieBlock()
+ ch2_group_id = 2
+ self.setupMcastChannel(mcast_groups.ch2, mcast_vlan_id, [onu1.port], ch2_group_id, ch2_cookie)
+ self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port], expect_to_be_blocked=True)
+ self.testMcastFlow(mcast_groups.ch2, mcast_vlan_id, [onu1.port], 10)
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
+
+ # 11. Add a second channel while keeping the existing one. Verify all what needs to be verified
+ # (similar as above) - at this point ONU1 is tuned into ch2 and ch3
+ self.sendIgmpReport(join=[mcast_groups.ch2, mcast_groups.ch3])
+ ch3_cookie = self.getCookieBlock()
+ ch3_group_id = 3
+ self.setupMcastChannel(mcast_groups.ch3, mcast_vlan_id, [onu1.port], ch3_group_id, ch3_cookie)
+ self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port], expect_to_be_blocked=True)
+ self.testMcastFlow(mcast_groups.ch2, mcast_vlan_id, [onu1.port], 10)
+ self.testMcastFlow(mcast_groups.ch3, mcast_vlan_id, [onu1.port], 10)
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
+
+ # 12. Setup the OLT to pass authentication (EAPOL) to controller from ONU2
+ # and test it.
+ self.installEapolRule(onu_port2)
+ self.sendEapolIn(onu_port2)
+
+ # 13. Setup the OLT to pass unicast traffic for the ONU2
+ self.installDoubleTaggingRules(s_vlan_id, onu2.c_vlan_id, self.getCookieBlock(), onu_port2)
+
+ # 14. Verify that unicast traffic works in both directions (use a few hundred frames in both ways)
+ self.testPacketFlow(s_vlan_id, onu2.c_vlan_id, onu_port2) # this tests just one packet in each way
+ self.testSustainedPacketFlow(s_vlan_id, onu2.c_vlan_id, 100, onu=onu_port2)
+
+ # 15. Verify that both ONU1 can send/receive unicast and ONU1 can still see its 2 channels
+ self.testMcastFlow(mcast_groups.ch2, mcast_vlan_id, [onu1.port], 10)
+ self.testMcastFlow(mcast_groups.ch3, mcast_vlan_id, [onu1.port], 10)
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
+
+ # 16. Setup IGMP forwarding toward the controller
+ # Note: this is actually not needed since the rule is not port specific - maybe it should be?
+ #self.installIgmpRule(self.getCookieBlock())
+ #self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
+
+ # 17. Send periodic IGMP queries out toward the ONU2 and verify its arrival
+ self.testIgmpQueryOut(onu_port2)
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10) # to check that unicast still flows
+ self.testSustainedPacketFlow(s_vlan_id, onu2.c_vlan_id, 10, onu=onu_port2) # to check that unicast still flows
+
+ # 18. Send in an IGMP join request for one channel (specific multicast address) and verify that
+ # the controller receives it.
+ self.sendIgmpReport(join=[mcast_groups.ch1], onu=onu_port2)
+
+ # 19. Setup flows for forwarding multicast traffic from the OLT port toward ONU2: ch1
+ self.setupMcastChannel(mcast_groups.ch1, mcast_vlan_id, [onu2.port], ch1_group_id, ch1_cookie)
+
+ # 20. Verify that ch1 multicast packets can reach ONU2, but not ONU1
+ self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu2.port], 10)
+ self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port], expect_to_be_blocked=True)
+
+ # 21. Verify that bidirectional unicast traffic still works across the PON for both ONUs
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id, 10)
+ self.testSustainedPacketFlow(s_vlan_id, onu2.c_vlan_id, 10, onu=onu_port2)
+
+ # 22. Change channel to a new multicast group and verify both multicast receiption as well as
+ # unicast traffic works. This step involves:
+ # - sending a leave message by the ONU to the controller and verify its reception
+ # - sending a join message and verify its reception
+ # - removing the existing multicast flow
+ # - adding a new multicast flow
+ # - verify that original flow no longer flows
+ # - verify new flow
+ # - verify that unicast still works
+ # - verify everything for ONU1 too
+ self.sendIgmpReport(join=[mcast_groups.ch2], leave=[mcast_groups.ch1], onu=onu2.port)
+ self.removeMcastChannel(mcast_groups.ch1, mcast_vlan_id, [onu2.port], ch1_group_id, ch1_cookie)
+ self.updateMcastChannel(ch2_group_id, port_list=[onu1.port, onu2.port])
+ self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port, onu2.port], expect_to_be_blocked=True)
+ self.testMcastFlow(mcast_groups.ch2, mcast_vlan_id, [onu1.port, onu2.port])
+ self.testMcastFlow(mcast_groups.ch3, mcast_vlan_id, [onu1.port])
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id)
+ self.testSustainedPacketFlow(s_vlan_id, onu2.c_vlan_id, onu=onu_port2)
+
+ # 23. Add a second channel for ONU2 while keeping the existing one. Verify all what needs to be verified
+ # (similar as above) - at this point ONU1 is tuned into ch2 and ch3 and ONU2 is
+ # tuned to ch2 and ch5
+ self.sendIgmpReport(join=[mcast_groups.ch2, mcast_groups.ch5], onu=onu2.port)
+ ch5_cookie = self.getCookieBlock()
+ ch5_group_id = 5
+ self.setupMcastChannel(mcast_groups.ch5, mcast_vlan_id, [onu2.port], ch5_group_id, ch5_cookie)
+ self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port, onu2.port], expect_to_be_blocked=True)
+ self.testMcastFlow(mcast_groups.ch2, mcast_vlan_id, [onu1.port, onu2.port])
+ self.testMcastFlow(mcast_groups.ch3, mcast_vlan_id, [onu1.port])
+ self.testMcastFlow(mcast_groups.ch5, mcast_vlan_id, [onu2.port])
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id)
+ self.testSustainedPacketFlow(s_vlan_id, onu2.c_vlan_id, onu=onu_port2)
+
+ # 24. Flip a channel for ONU1 and verify everything (ch3 -> ch4)
+ self.sendIgmpReport(leave=[mcast_groups.ch3], join=[mcast_groups.ch2, mcast_groups.ch4], onu=onu1.port)
+ self.removeMcastChannel(mcast_groups.ch3, mcast_vlan_id, [onu1.port], ch3_group_id, ch3_cookie)
+ ch4_cookie = self.getCookieBlock()
+ ch4_group_id = 4
+ self.setupMcastChannel(mcast_groups.ch4, mcast_vlan_id, [onu1.port], ch4_group_id, ch4_cookie)
+ self.testMcastFlow(mcast_groups.ch1, mcast_vlan_id, [onu1.port, onu2.port], expect_to_be_blocked=True)
+ self.testMcastFlow(mcast_groups.ch2, mcast_vlan_id, [onu1.port, onu2.port])
+ self.testMcastFlow(mcast_groups.ch3, mcast_vlan_id, [onu1.port, onu2.port], expect_to_be_blocked=True)
+ self.testMcastFlow(mcast_groups.ch4, mcast_vlan_id, [onu1.port])
+ self.testMcastFlow(mcast_groups.ch5, mcast_vlan_id, [onu2.port])
+ self.testSustainedPacketFlow(s_vlan_id, onu1.c_vlan_id)
+ self.testSustainedPacketFlow(s_vlan_id, onu2.c_vlan_id, onu=onu_port2)
+
+ # 25. Tear down the test.
+ self.resetOlt()
diff --git a/olt.py b/olt.py
index 7330177..f68c884 100644
--- a/olt.py
+++ b/olt.py
@@ -9,14 +9,13 @@
import ofp
import time
+from oltconstants import *
from oftest.testutils import *
from IGMP import IGMPv3, IGMPv3gr, IGMP_TYPE_V3_MEMBERSHIP_REPORT,\
IGMP_V3_GR_TYPE_INCLUDE,\
IGMP_V3_GR_TYPE_EXCLUDE
-from oltconstants import *
-
class EapolPacketIn(OltBaseTest):
"""Verify packet-ins are sent for EAPOL packets """
diff --git a/oltbase.py b/oltbase.py
index 2baf37c..56d43a6 100644
--- a/oltbase.py
+++ b/oltbase.py
@@ -1,8 +1,8 @@
import oftest.base_tests as base_tests
from oftest import config
import ofp
-from oftest.testutils import *
from oltconstants import *
+from oftest.testutils import *
import copy
import logging
from IGMP import IGMPv3, IGMPv3gr, IGMP_TYPE_MEMBERSHIP_QUERY, \
@@ -76,7 +76,7 @@
buffer_id=ofp.OFP_NO_BUFFER,
priority=1000)
else:
- request = ofp.message.flow_delete(
+ request = ofp.message.flow_delete_strict(
table_id=test_param_get("table", 0),
cookie=42,
match=match,
@@ -101,6 +101,24 @@
verify_packet_in(self, pkt, in_port, ofp.OFPR_ACTION)
verify_packets(self, pkt, [])
+ def sendEapolOut(self, out_port=None):
+ """Send out an EAPOL frame and verify that it is sent out at the given ONU port"""
+ out_port = onu_port if out_port is None else out_port
+ pkt = simple_eth_packet(eth_dst='01:02:03:04:05:06', eth_src='01:02:03:de:ad:be',
+ eth_type=0x888e, pktlen=300)
+
+ msg = ofp.message.packet_out(
+ in_port=ofp.OFPP_CONTROLLER,
+ actions=[ofp.action.output(port=out_port)],
+ buffer_id=ofp.OFP_NO_BUFFER,
+ data=str(pkt))
+
+ rv = self.controller.message_send(msg)
+ self.assertTrue(rv == 0, "Error sending output message")
+ verify_no_errors(self.controller)
+
+ verify_packet(self, pkt, out_port)
+
def testPacketFlow(self,
s_vlan_id,
c_vlan_id,
@@ -150,7 +168,7 @@
print "pkt # %d" % (i+1,)
self.testPacketFlow(s_vlan_id, c_vlan_id, onu=onu, verify_blocked_flows=False)
- def installDoubleTaggingRules(self, s_vlan_id, c_vlan_id, cookie=42, onu = None):
+ def installDoubleTaggingRules(self, s_vlan_id, c_vlan_id, cookie=42, onu=None):
inport = onu_port if onu is None else onu
@@ -443,6 +461,30 @@
do_barrier(self.controller)
verify_no_errors(self.controller)
+ def updateMcastChannel(self, group_id, port_list):
+ """Modify the group port list by adding/removing ports"""
+
+ buckets = [
+ ofp.common.bucket(
+ watch_port=ofp.OFPP_ANY,
+ watch_group=ofp.OFPG_ANY,
+ actions=[
+ ofp.action.pop_vlan(),
+ ofp.action.output(port=port)
+ ])
+ for port in port_list
+ ]
+
+ msg = ofp.message.group_mod(
+ command=ofp.OFPGC_MODIFY,
+ group_type=ofp.OFPGT_ALL,
+ group_id=group_id,
+ buckets=buckets)
+
+ self.controller.message_send(msg)
+ do_barrier(self.controller)
+ verify_no_errors(self.controller)
+
def ip2int(self, ip):
"""Convert a dot-notated string IP address"""
digits = [int(d) for d in ip.split('.')]
diff --git a/oltconstants.py b/oltconstants.py
index b97ab5a..29a6db5 100644
--- a/oltconstants.py
+++ b/oltconstants.py
@@ -1,4 +1,5 @@
-from oftest.testutils import *
+import logging
+from oftest.testutils import test_param_get, MINSIZE
import oftest.packet as scapy
from oftest import config
@@ -11,7 +12,7 @@
onu_port2 = test_param_get("onu_port2", 131)
olt_port = test_param_get("olt_port", 258)
device_type = test_param_get("device_type", "normal") # options: "normal", "pmc", "cpqd"
-
+fake_dataplane = test_param_get("fake_dataplane", False)
def createAllGroupAdd(group_id, ports=[]):
buckets = []
@@ -101,3 +102,28 @@
pkt = pkt / ("D" * (pktlen - len(pkt)))
return pkt
+
+# Monkey-patch the oftest library to allow faking the data-plane
+# This mode allows oftest to glide through the OF programming steps while
+# passing all packet receiving tests. This is useful when you want to
+# verify that your agent's OF interface handles all OF protocol messages
+# from the controller, but in the absence of actual data-plane tests.
+# In order to achieve this, in this mode the following chnages are made:
+# - All verify_packet_in() tests pass
+if fake_dataplane:
+ logging.info('Monkey-patching for fake-dataplane operations')
+
+ from oftest import testutils
+
+ def patched_verify_packet_in(test, data, in_port, reason, controller=None):
+ logging.debug('Faking verify_packet_in')
+ msg = ofp.message.packet_in(reason=reason, data=data)
+ return msg
+
+ testutils.verify_packet_in = patched_verify_packet_in
+
+ def patched_verify_packet(test, pkt, of_port):
+ logging.debug('Faking verify_packet')
+ return
+
+ testutils.verify_packet = patched_verify_packet