1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
|
#!/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])
|
