From f56b77a0764222cc45a9df572df901067a273356 Mon Sep 17 00:00:00 2001 From: Damjan Marion Date: Mon, 3 Oct 2016 19:44:57 +0200 Subject: test: new test infrastructure Change-Id: I73ca19c431743f6b39669c583d9222a6559346ef Signed-off-by: Jan Gelety Signed-off-by: Juraj Sloboda Signed-off-by: Stefan Kobza Signed-off-by: Matej Klotton Signed-off-by: Maciek Konstantynowicz Signed-off-by: Damjan Marion --- test/template_bd.py | 106 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 106 insertions(+) create mode 100644 test/template_bd.py (limited to 'test/template_bd.py') diff --git a/test/template_bd.py b/test/template_bd.py new file mode 100644 index 00000000..473c4228 --- /dev/null +++ b/test/template_bd.py @@ -0,0 +1,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]) -- cgit 1.2.3-korg