master-thesis/netpfga/minip4/testdata/gen_testdata-port1.py
Nico Schottelius d97864ffa7 +notes -comment
2019-07-21 16:39:15 +02:00

125 lines
2.9 KiB
Python
Executable file

#!/usr/bin/env python
# from switch_calc_headers import *
from scapy.all import *
from nf_sim_tools import *
from collections import OrderedDict
import sss_sdnet_tuples
########################
# pkt generation tools #
########################
pktsApplied = []
pktsExpected = []
# Pkt lists for SUME simulations
nf_applied = OrderedDict()
nf_applied[0] = []
nf_applied[1] = []
nf_applied[2] = []
nf_applied[3] = []
nf_expected = OrderedDict()
nf_expected[0] = []
nf_expected[1] = []
nf_expected[2] = []
nf_expected[3] = []
nf_port_map = {
"nf0":0b00000001,
"nf1":0b00000100,
"nf2":0b00010000,
"nf3":0b01000000,
"dma0":0b00000010
}
nf_id_map = {
"nf0":0,
"nf1":1,
"nf2":2,
"nf3":3
}
sss_sdnet_tuples.clear_tuple_files()
def applyPkt(pkt, ingress, time):
print("Applying pkt on {} at {}: ".format(ingress, time))
pktsApplied.append(pkt)
sss_sdnet_tuples.sume_tuple_in['src_port'] = nf_port_map[ingress]
sss_sdnet_tuples.sume_tuple_expect['src_port'] = nf_port_map[ingress]
pkt.time = time
nf_applied[nf_id_map[ingress]].append(pkt)
def expPkt(pkt, egress):
pktsExpected.append(pkt)
sss_sdnet_tuples.sume_tuple_expect['dst_port'] = nf_port_map[egress]
sss_sdnet_tuples.write_tuples()
if egress in ["nf0","nf1","nf2","nf3"]:
nf_expected[nf_id_map[egress]].append(pkt)
elif egress == 'bcast':
nf_expected[0].append(pkt)
nf_expected[1].append(pkt)
nf_expected[2].append(pkt)
nf_expected[3].append(pkt)
def print_summary(pkts):
for pkt in pkts:
print "summary = ", pkt.summary()
def write_pcap_files():
wrpcap("src.pcap", pktsApplied)
wrpcap("dst.pcap", pktsExpected)
for i in nf_applied.keys():
if (len(nf_applied[i]) > 0):
wrpcap('nf{0}_applied.pcap'.format(i), nf_applied[i])
for i in nf_expected.keys():
if (len(nf_expected[i]) > 0):
wrpcap('nf{0}_expected.pcap'.format(i), nf_expected[i])
# i = 0..3
for i in nf_applied.keys():
print "nf{0}_applied times: ".format(i), [p.time for p in nf_applied[i]]
#####################
# generate testdata #
#####################
MAC1 = "08:11:11:11:11:08"
MAC2 = "08:22:22:22:22:08"
pktCnt = 0
INDEX_WIDTH = 4
REG_DEPTH = 2**INDEX_WIDTH
# Not sure what this is used for
NUM_KEYS = 4
lookup_table = {
0: 0x00000001,
1: 0x00000010,
2: 0x00000100,
3: 0x00001000
}
def test_port1():
""" packets for a certain mac always go to nf0 """
pktCnt = 1
for inport in [ 'nf0', 'nf1', 'nf2', 'nf3' ]:
# Need to be regenerated, as object is modified above!
pkt = Ether(dst=MAC2, src=MAC1)
pkt = pad_pkt(pkt, 64)
# Send on nf0..nf3
applyPkt(pkt, inport, pktCnt)
# Always receive on nf0 - also regenerate here!
pkt = Ether(dst=MAC2, src=MAC1)
pkt = pad_pkt(pkt, 64)
expPkt(pkt, 'nf0')
pktCnt += 1
test_port1()
write_pcap_files()