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