3 import time,argparse,sys,cmd, unittest
4 from ipaddress import *
6 parser = argparse.ArgumentParser(description='VPP MAP test')
7 parser.add_argument('-i', nargs='*', action="store", dest="inputdir")
8 args = parser.parse_args()
10 for dir in args.inputdir:
12 from vpp_papi import *
15 # 1:1 Shared IPv4 address, shared BR (16) VPP CLI
17 def lw46_shared(ip4_pfx_str, ip6_pfx_str, ip6_src_str, ea_bits_len, psid_offset, psid_len, ip6_src_ecmp = False):
18 ip4_pfx = ip_network(ip4_pfx_str)
19 ip6_src = ip_address(ip6_src_str)
20 ip6_dst = ip_network(ip6_pfx_str)
21 ip6_nul = IPv6Address(u'0::0')
22 mod = ip4_pfx.num_addresses / 1024
24 for i in range(ip4_pfx.num_addresses):
26 t = map_add_domain(0, ip6_nul.packed, ip4_pfx[i].packed, ip6_src.packed, 0, 32, 128, ea_bits_len, psid_offset, psid_len, 0, 0)
27 #print "Return from map_add_domain", t
29 print "map_add_domain failed"
32 print "map_add_domain failed", t
34 for psid in range(0x1 << int(psid_len)):
35 r = map_add_del_rule(0, t.index, 1, (ip6_dst[(i * (0x1<<int(psid_len))) + psid]).packed, psid)
36 #print "Return from map_add_del_rule", r
38 if ip6_src_ecmp and not i % mod:
41 print "Running time:", time.clock() - a
43 class TestMAP(unittest.TestCase):
45 def test_delete_all(self):
46 t = map_domain_dump(0)
47 self.assertNotEqual(t, None)
48 print "Number of domains configured: ", len(t)
50 ts = map_del_domain(0, d.domainindex)
51 self.assertNotEqual(ts, None)
52 t = map_domain_dump(0)
53 self.assertNotEqual(t, None)
54 print "Number of domains configured: ", len(t)
55 self.assertEqual(len(t), 0)
59 def test_a_million_rules(self):
60 ip4_pfx = u'192.0.2.0/24'
61 ip6_pfx = u'2001:db8::/32'
62 ip6_src = u'2001:db8::1'
66 lw46_shared(ip4_pfx, ip6_pfx, ip6_src, ea_bits_len, psid_offset, psid_len)
69 # RX thread, that should sit on blocking vpe_api_read()
78 class RXThread (threading.Thread):
80 threading.Thread.__init__(self)
89 id = unpack('>H', msg[0:2])
90 size = unpack('>H', msg[2:4])
91 print "Received", id, "of size", size
100 rxthread = RXThread()
101 rxthread.setDaemon(True)
103 print "Connect", connect_to_vpe("client124")
106 print "After thread started"
109 print "After thread killed"
112 #print "Result from show version", t
114 print timeit.timeit('t = show_version(0)', number=1000, setup="from __main__ import show_version")
116 #print timeit.timeit('control_ping(0)', number=10, setup="from __main__ import control_ping")
119 disconnect_from_vpe()
123 print t.program, t.version,t.builddate,t.builddirectory
127 t = map_domain_dump(0)
129 print('show map domain failed')
132 print("IP6 prefix:",str(IPv6Address(d.ip6prefix)))
133 print( "IP4 prefix:",str(IPv4Address(d.ip4prefix)))
136 suite = unittest.TestLoader().loadTestsFromTestCase(TestMAP)
137 unittest.TextTestRunner(verbosity=2).run(suite)
139 disconnect_from_vpe()