f56b77a076
Change-Id: I73ca19c431743f6b39669c583d9222a6559346ef Signed-off-by: Jan Gelety <jgelety@cisco.com> Signed-off-by: Juraj Sloboda <jsloboda@cisco.com> Signed-off-by: Stefan Kobza <skobza@cisco.com> Signed-off-by: Matej Klotton <mklotton@cisco.com> Signed-off-by: Maciek Konstantynowicz <mkonstan@cisco.com> Signed-off-by: Damjan Marion <damarion@cisco.com>
107 lines
4.0 KiB
Python
107 lines
4.0 KiB
Python
#!/usr/bin/env python
|
|
|
|
from abc import abstractmethod
|
|
|
|
from scapy.layers.l2 import Ether, Raw
|
|
from scapy.layers.inet import IP, UDP
|
|
|
|
|
|
class BridgeDomain(object):
|
|
def __init__(self):
|
|
## Ethernet frame which is send to pg0 interface and is forwarded to pg1
|
|
self.payload_0_1 = (
|
|
Ether(src='00:00:00:00:00:01', dst='00:00:00:00:00:02') /
|
|
IP(src='1.2.3.4', dst='4.3.2.1') /
|
|
UDP(sport=10000, dport=20000) /
|
|
Raw('\xa5' * 100))
|
|
|
|
## Ethernet frame which is send to pg1 interface and is forwarded to pg0
|
|
self.payload_1_0 = (
|
|
Ether(src='00:00:00:00:00:02', dst='00:00:00:00:00:01') /
|
|
IP(src='4.3.2.1', dst='1.2.3.4') /
|
|
UDP(sport=20000, dport=10000) /
|
|
Raw('\xa5' * 100))
|
|
|
|
## Test case must implement this method, so template known how to send
|
|
# encapsulated frame.
|
|
@abstractmethod
|
|
def encapsulate(self, pkt):
|
|
pass
|
|
|
|
## Test case must implement this method, so template known how to get
|
|
# original payload.
|
|
@abstractmethod
|
|
def decapsulate(self, pkt):
|
|
pass
|
|
|
|
## Test case must implement this method, so template known how if the
|
|
# received frame is corectly encapsulated.
|
|
@abstractmethod
|
|
def check_encapsulation(self, pkt):
|
|
pass
|
|
|
|
## On pg0 interface are encapsulated frames, on pg1 are testing frames
|
|
# without encapsulation
|
|
def test_decap(self):
|
|
## Prepare Ethernet frame that will be send encapsulated.
|
|
pkt_to_send = self.encapsulate(self.payload_0_1)
|
|
|
|
## Add packet to list of packets.
|
|
self.pg_add_stream(0, [pkt_to_send, ])
|
|
|
|
## Enable Packet Capture on both ports.
|
|
self.pg_enable_capture([0, 1])
|
|
|
|
## Start all streams
|
|
self.pg_start()
|
|
|
|
## Pick first received frame and check if is same as non-encapsulated
|
|
# frame.
|
|
out = self.pg_get_capture(1)
|
|
self.assertEqual(len(out), 1,
|
|
'Invalid number of packets on '
|
|
'output: {}'.format(len(out)))
|
|
pkt = out[0]
|
|
|
|
# TODO: add error messages
|
|
self.assertEqual(pkt[Ether].src, self.payload_0_1[Ether].src)
|
|
self.assertEqual(pkt[Ether].dst, self.payload_0_1[Ether].dst)
|
|
self.assertEqual(pkt[IP].src, self.payload_0_1[IP].src)
|
|
self.assertEqual(pkt[IP].dst, self.payload_0_1[IP].dst)
|
|
self.assertEqual(pkt[UDP].sport, self.payload_0_1[UDP].sport)
|
|
self.assertEqual(pkt[UDP].dport, self.payload_0_1[UDP].dport)
|
|
self.assertEqual(pkt[Raw], self.payload_0_1[Raw])
|
|
|
|
## Send non-encapsulated Ethernet frame from pg1 interface and expect
|
|
# encapsulated frame on pg0. On pg0 interface are encapsulated frames,
|
|
# on pg1 are testing frames without encapsulation.
|
|
def test_encap(self):
|
|
## Create packet generator stream.
|
|
self.pg_add_stream(1, [self.payload_1_0])
|
|
|
|
## Enable Packet Capture on both ports.
|
|
self.pg_enable_capture([0, 1])
|
|
|
|
## Start all streams.
|
|
self.pg_start()
|
|
|
|
## Pick first received frame and check if is corectly encapsulated.
|
|
out = self.pg_get_capture(0)
|
|
self.assertEqual(len(out), 1,
|
|
'Invalid number of packets on '
|
|
'output: {}'.format(len(out)))
|
|
rcvd = out[0]
|
|
self.check_encapsulation(rcvd)
|
|
|
|
## Get original frame from received packet and check if is same as
|
|
# sended frame.
|
|
rcvd_payload = self.decapsulate(rcvd)
|
|
# TODO: add error messages
|
|
self.assertEqual(rcvd_payload[Ether].src, self.payload_1_0[Ether].src)
|
|
self.assertEqual(rcvd_payload[Ether].dst, self.payload_1_0[Ether].dst)
|
|
self.assertEqual(rcvd_payload[IP].src, self.payload_1_0[IP].src)
|
|
self.assertEqual(rcvd_payload[IP].dst, self.payload_1_0[IP].dst)
|
|
self.assertEqual(rcvd_payload[UDP].sport, self.payload_1_0[UDP].sport)
|
|
self.assertEqual(rcvd_payload[UDP].dport, self.payload_1_0[UDP].dport)
|
|
self.assertEqual(rcvd_payload[Raw], self.payload_1_0[Raw])
|