Reorganize source tree to use single autotools instance

[vpp.git] / src / vnet / map / examples / test_map.py

diff --git a/src/vnet/map/examples/test_map.py b/src/vnet/map/examples/test_map.py
new file mode 100755 (executable)
index 0000000..21388d4
--- /dev/null
+++ b/src/vnet/map/examples/test_map.py
@@ -0,0 +1,141 @@
+#!/usr/bin/env python
+
+import time,argparse,sys,cmd, unittest
+from ipaddress import *
+
+parser = argparse.ArgumentParser(description='VPP MAP test')
+parser.add_argument('-i', nargs='*', action="store", dest="inputdir")
+args = parser.parse_args()
+
+for dir in args.inputdir:
+    sys.path.append(dir)
+from vpp_papi import *
+
+#
+# 1:1 Shared IPv4 address, shared BR (16) VPP CLI
+#
+def lw46_shared(ip4_pfx_str, ip6_pfx_str, ip6_src_str, ea_bits_len, psid_offset, psid_len, ip6_src_ecmp = False):
+    ip4_pfx = ip_network(ip4_pfx_str)
+    ip6_src = ip_address(ip6_src_str)
+    ip6_dst = ip_network(ip6_pfx_str)
+    ip6_nul = IPv6Address(u'0::0')
+    mod = ip4_pfx.num_addresses / 1024
+
+    for i in range(ip4_pfx.num_addresses):
+        a = time.clock()
+        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)
+        #print "Return from map_add_domain", t
+        if t == None:
+            print "map_add_domain failed"
+            continue
+        if t.retval != 0:
+            print "map_add_domain failed", t
+            continue
+        for psid in range(0x1 << int(psid_len)):
+            r = map_add_del_rule(0, t.index, 1, (ip6_dst[(i * (0x1<<int(psid_len))) + psid]).packed, psid)
+            #print "Return from map_add_del_rule", r
+
+        if ip6_src_ecmp and not i % mod:
+            ip6_src = ip6_src + 1
+
+        print "Running time:", time.clock() - a
+
+class TestMAP(unittest.TestCase):
+    '''
+    def test_delete_all(self):
+        t = map_domain_dump(0)
+        self.assertNotEqual(t, None)
+        print "Number of domains configured: ", len(t)
+        for d in t:
+            ts = map_del_domain(0, d.domainindex)
+            self.assertNotEqual(ts, None)
+        t = map_domain_dump(0)
+        self.assertNotEqual(t, None)
+        print "Number of domains configured: ", len(t)
+        self.assertEqual(len(t), 0)
+
+    '''
+
+    def test_a_million_rules(self):
+        ip4_pfx = u'192.0.2.0/24'
+        ip6_pfx = u'2001:db8::/32'
+        ip6_src = u'2001:db8::1'
+        psid_offset = 6
+        psid_len = 6
+        ea_bits_len = 0
+        lw46_shared(ip4_pfx, ip6_pfx, ip6_src, ea_bits_len, psid_offset, psid_len)
+
+#
+# RX thread, that should sit on blocking vpe_api_read()
+
+# 
+
+
+#
+#
+#
+import threading
+class RXThread (threading.Thread):
+    def __init__(self):
+        threading.Thread.__init__(self)
+
+    def run(self):
+        print "Starting "
+        i = 0
+        while True:
+            msg = vpe_api_read()
+            if msg:
+                #print msg
+                id = unpack('>H', msg[0:2])
+                size = unpack('>H', msg[2:4])
+                print "Received", id, "of size", size
+                i += 1
+                #del msg
+                continue
+
+            #time.sleep(0.001)
+        return
+
+# Create RX thread
+rxthread = RXThread()
+rxthread.setDaemon(True)
+        
+print "Connect", connect_to_vpe("client124")
+import timeit
+rxthread.start()
+print "After thread started"
+
+#pneum_kill_thread()
+print "After thread killed"
+
+#t = show_version(0)
+#print "Result from show version", t
+
+print timeit.timeit('t = show_version(0)', number=1000, setup="from __main__ import show_version")
+time.sleep(10)
+#print timeit.timeit('control_ping(0)', number=10, setup="from __main__ import control_ping")
+
+
+disconnect_from_vpe()
+sys.exit()
+
+
+print t.program, t.version,t.builddate,t.builddirectory
+
+'''
+
+t = map_domain_dump(0)
+if not t:
+    print('show map domain failed')
+
+for d in t:
+    print("IP6 prefix:",str(IPv6Address(d.ip6prefix)))
+    print( "IP4 prefix:",str(IPv4Address(d.ip4prefix)))
+'''
+
+suite = unittest.TestLoader().loadTestsFromTestCase(TestMAP)
+unittest.TextTestRunner(verbosity=2).run(suite)
+
+disconnect_from_vpe()
+
+