def wait_for_event(self, timeout, name=None):
""" Wait for and return next event. """
if name:
- self.test_class.logger.debug("Expecting event within %ss",
- timeout)
- else:
self.test_class.logger.debug("Expecting event '%s' within %ss",
name, timeout)
+ else:
+ self.test_class.logger.debug("Expecting event within %ss",
+ timeout)
if self._events:
self.test_class.logger.debug("Not waiting, event already queued")
limit = time.time() + timeout
'ip_address': ip,
'mac_address': mac})
+ def want_ip4_arp_events(self, enable_disable=1, address=0):
+ return self.api(self.papi.want_ip4_arp_events,
+ {'enable_disable': enable_disable,
+ 'address': address,
+ 'pid': os.getpid(), })
+
+ def want_ip6_nd_events(self, enable_disable=1, address=0):
+ return self.api(self.papi.want_ip6_nd_events,
+ {'enable_disable': enable_disable,
+ 'address': address,
+ 'pid': os.getpid(), })
+
def l2fib_add_del(self, mac, bd_id, sw_if_index, is_add=1, static_mac=0,
filter_mac=0, bvi_mac=0):
"""Create/delete L2 FIB entry.
'traffic_type': traffic_type
})
- def macip_acl_add_replace(self, rules, acl_index=0xFFFFFFFF, tag=""):
+ def macip_acl_add(self, rules, tag=""):
""" Add MACIP acl
:param rules: list of rules for given acl
:param tag: acl tag
"""
- # return self.api(self.papi.macip_acl_add_replace,
- # {'acl_index': acl_index,
- # 'r': rules,
- # 'count': len(rules),
- # 'tag': tag})
return self.api(self.papi.macip_acl_add,
{'r': rules,
'count': len(rules),
'tag': tag})
+ def macip_acl_add_replace(self, rules, acl_index=0xFFFFFFFF, tag=""):
+ """ Add MACIP acl
+
+ :param rules: list of rules for given acl
+ :param tag: acl tag
+ """
+
+ return self.api(self.papi.macip_acl_add_replace,
+ {'acl_index': acl_index,
+ 'r': rules,
+ 'count': len(rules),
+ 'tag': tag})
+
def macip_acl_del(self, acl_index):
"""