src/vnet/map/examples/test_map.py

   1 #!/usr/bin/env python
   2
   3 import time,argparse,sys,cmd, unittest
   4 from ipaddress import *
   5
   6 parser = argparse.ArgumentParser(description='VPP MAP test')
   7 parser.add_argument('-i', nargs='*', action="store", dest="inputdir")
   8 args = parser.parse_args()
   9
  10 for dir in args.inputdir:
  11     sys.path.append(dir)
  12 from vpp_papi import *
  13
  14 #
  15 # 1:1 Shared IPv4 address, shared BR (16) VPP CLI
  16 #
  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
  23
  24     for i in range(ip4_pfx.num_addresses):
  25         a = time.clock()
  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
  28         if t == None:
  29             print "map_add_domain failed"
  30             continue
  31         if t.retval != 0:
  32             print "map_add_domain failed", t
  33             continue
  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
  37
  38         if ip6_src_ecmp and not i % mod:
  39             ip6_src = ip6_src + 1
  40
  41         print "Running time:", time.clock() - a
  42
  43 class TestMAP(unittest.TestCase):
  44     '''
  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)
  49         for d in 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)
  56
  57     '''
  58
  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'
  63         psid_offset = 6
  64         psid_len = 6
  65         ea_bits_len = 0
  66         lw46_shared(ip4_pfx, ip6_pfx, ip6_src, ea_bits_len, psid_offset, psid_len)
  67
  68 #
  69 # RX thread, that should sit on blocking vpe_api_read()
  70
  71 #
  72
  73
  74 #
  75 #
  76 #
  77 import threading
  78 class RXThread (threading.Thread):
  79     def __init__(self):
  80         threading.Thread.__init__(self)
  81
  82     def run(self):
  83         print "Starting "
  84         i = 0
  85         while True:
  86             msg = vpe_api_read()
  87             if msg:
  88                 #print msg
  89                 id = unpack('>H', msg[0:2])
  90                 size = unpack('>H', msg[2:4])
  91                 print "Received", id, "of size", size
  92                 i += 1
  93                 #del msg
  94                 continue
  95
  96             #time.sleep(0.001)
  97         return
  98
  99 # Create RX thread
 100 rxthread = RXThread()
 101 rxthread.setDaemon(True)
 102
 103 print "Connect", connect_to_vpe("client124")
 104 import timeit
 105 rxthread.start()
 106 print "After thread started"
 107
 108 #pneum_kill_thread()
 109 print "After thread killed"
 110
 111 #t = show_version(0)
 112 #print "Result from show version", t
 113
 114 print timeit.timeit('t = show_version(0)', number=1000, setup="from __main__ import show_version")
 115 time.sleep(10)
 116 #print timeit.timeit('control_ping(0)', number=10, setup="from __main__ import control_ping")
 117
 118
 119 disconnect_from_vpe()
 120 sys.exit()
 121
 122
 123 print t.program, t.version,t.builddate,t.builddirectory
 124
 125 '''
 126
 127 t = map_domain_dump(0)
 128 if not t:
 129     print('show map domain failed')
 130
 131 for d in t:
 132     print("IP6 prefix:",str(IPv6Address(d.ip6prefix)))
 133     print( "IP4 prefix:",str(IPv4Address(d.ip4prefix)))
 134 '''
 135
 136 suite = unittest.TestLoader().loadTestsFromTestCase(TestMAP)
 137 unittest.TextTestRunner(verbosity=2).run(suite)
 138
 139 disconnect_from_vpe()
 140
 141