-#!/usr/bin/env python
+#!/usr/bin/env python3
-from abc import abstractmethod, ABCMeta
+import abc
+import six
-from scapy.layers.l2 import Ether, Raw
+from scapy.layers.l2 import Ether
+from scapy.packet import Raw
from scapy.layers.inet import IP, UDP
from util import ip4_range
+@six.add_metaclass(abc.ABCMeta)
class BridgeDomain(object):
""" Bridge domain abstraction """
- __metaclass__ = ABCMeta
@property
def frame_request(self):
UDP(sport=20000, dport=10000) /
Raw('\xa5' * 100))
- @abstractmethod
+ @abc.abstractmethod
def ip_range(self, start, end):
""" range of remote ip's """
pass
- @abstractmethod
+ @abc.abstractmethod
def encap_mcast(self, pkt, src_ip, src_mac, vni):
""" Encapsulate mcast packet """
pass
- @abstractmethod
+ @abc.abstractmethod
def encapsulate(self, pkt, vni):
""" Encapsulate packet """
pass
- @abstractmethod
+ @abc.abstractmethod
def decapsulate(self, pkt):
""" Decapsulate packet """
pass
- @abstractmethod
+ @abc.abstractmethod
def check_encapsulation(self, pkt, vni, local_only=False):
""" Verify the encapsulation """
pass
"""
encapsulated_pkt = self.encapsulate(self.frame_request,
- self.single_tunnel_bd)
+ self.single_tunnel_vni)
self.pg0.add_stream([encapsulated_pkt, ])
self.pg_start()
- # Pick first received frame and check if it's corectly encapsulated.
+ # Pick first received frame and check if it's correctly encapsulated.
out = self.pg0.get_capture(1)
pkt = out[0]
- self.check_encapsulation(pkt, self.single_tunnel_bd)
+ self.check_encapsulation(pkt, self.single_tunnel_vni)
payload = self.decapsulate(pkt)
self.assert_eq_pkts(payload, self.frame_reply)
self.pg_start()
- # Get packet from each tunnel and assert it's corectly encapsulated.
+ # Get packet from each tunnel and assert it's correctly encapsulated.
out = self.pg0.get_capture(self.n_ucast_tunnels)
for pkt in out:
self.check_encapsulation(pkt, self.ucast_flood_bd, True)
self.pg_start()
- # Pick first received frame and check if it's corectly encapsulated.
+ # Pick first received frame and check if it's correctly encapsulated.
out = self.pg0.get_capture(1)
pkt = out[0]
self.check_encapsulation(pkt, self.mcast_flood_bd,