src/plugins/map/examples/test_map.py

   1 #!/usr/bin/env python3
   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(
  18     ip4_pfx_str,
  19     ip6_pfx_str,
  20     ip6_src_str,
  21     ea_bits_len,
  22     psid_offset,
  23     psid_len,
  24     ip6_src_ecmp=False,
  25 ):
  26     ip4_pfx = ip_network(ip4_pfx_str)
  27     ip6_src = ip_address(ip6_src_str)
  28     ip6_dst = ip_network(ip6_pfx_str)
  29     ip6_nul = IPv6Address("0::0")
  30     mod = ip4_pfx.num_addresses / 1024
  31
  32     for i in range(ip4_pfx.num_addresses):
  33         a = time.clock()
  34         t = map_add_domain(
  35             0,
  36             ip6_nul.packed,
  37             ip4_pfx[i].packed,
  38             ip6_src.packed,
  39             0,
  40             32,
  41             128,
  42             ea_bits_len,
  43             psid_offset,
  44             psid_len,
  45             0,
  46             0,
  47         )
  48         # print "Return from map_add_domain", t
  49         if t == None:
  50             print("map_add_domain failed")
  51             continue
  52         if t.retval != 0:
  53             print(f"map_add_domain failed, {t}")
  54             continue
  55         for psid in range(0x1 << int(psid_len)):
  56             r = map_add_del_rule(
  57                 0,
  58                 t.index,
  59                 1,
  60                 (ip6_dst[(i * (0x1 << int(psid_len))) + psid]).packed,
  61                 psid,
  62             )
  63             # print "Return from map_add_del_rule", r
  64
  65         if ip6_src_ecmp and not i % mod:
  66             ip6_src = ip6_src + 1
  67
  68         print(f"Running time: {time.clock() - a}")
  69
  70
  71 class TestMAP(unittest.TestCase):
  72     """
  73     def test_delete_all(self):
  74         t = map_domain_dump(0)
  75         self.assertNotEqual(t, None)
  76         print(f"Number of domains configured: {len(t)}")
  77         for d in t:
  78             ts = map_del_domain(0, d.domainindex)
  79             self.assertNotEqual(ts, None)
  80         t = map_domain_dump(0)
  81         self.assertNotEqual(t, None)
  82         print(f"Number of domains configured: {len(t)}")
  83         self.assertEqual(len(t), 0)/
  84
  85     """
  86
  87     def test_a_million_rules(self):
  88         ip4_pfx = "192.0.2.0/24"
  89         ip6_pfx = "2001:db8::/32"
  90         ip6_src = "2001:db8::1"
  91         psid_offset = 6
  92         psid_len = 6
  93         ea_bits_len = 0
  94         lw46_shared(ip4_pfx, ip6_pfx, ip6_src, ea_bits_len, psid_offset, psid_len)
  95
  96
  97 #
  98 # RX thread, that should sit on blocking vpe_api_read()
  99
 100 #
 101
 102
 103 #
 104 #
 105 #
 106 import threading
 107
 108
 109 class RXThread(threading.Thread):
 110     def __init__(self):
 111         threading.Thread.__init__(self)
 112
 113     def run(self):
 114         print("Starting ")
 115         i = 0
 116         while True:
 117             msg = vpe_api_read()
 118             if msg:
 119                 # print msg
 120                 id = unpack(">H", msg[0:2])
 121                 size = unpack(">H", msg[2:4])
 122                 print(f"Received {id} of size {size}")
 123                 i += 1
 124                 # del msg
 125                 continue
 126
 127             # time.sleep(0.001)
 128         return
 129
 130
 131 # Create RX thread
 132 rxthread = RXThread()
 133 rxthread.setDaemon(True)
 134
 135 print(f"Connect {connect_to_vpe('client124')}")
 136 import timeit
 137
 138 rxthread.start()
 139 print("After thread started")
 140
 141 # pneum_kill_thread()
 142 print("After thread killed")
 143
 144 # t = show_version(0)
 145 # print "Result from show version", t
 146
 147 print(
 148     f"{timeit.timeit('t = show_version(0)', number=1000, setup='from __main__ import show_version')}"
 149 )
 150 time.sleep(10)
 151 # print timeit.timeit('control_ping(0)', number=10, setup="from __main__ import control_ping")
 152
 153
 154 disconnect_from_vpe()
 155 sys.exit()
 156
 157
 158 print(f"{t.program} {t.version}{t.builddate}{t.builddirectory}")
 159
 160 """
 161
 162 t = map_domain_dump(0)
 163 if not t:
 164     print('show map domain failed')
 165
 166 for d in t:
 167     print("IP6 prefix:",str(IPv6Address(d.ip6prefix)))
 168     print( "IP4 prefix:",str(IPv4Address(d.ip4prefix)))
 169 """
 170
 171 suite = unittest.TestLoader().loadTestsFromTestCase(TestMAP)
 172 unittest.TextTestRunner(verbosity=2).run(suite)
 173
 174 disconnect_from_vpe()