summaryrefslogtreecommitdiffstats
path: root/vpp-api/python/tests/test_papi.py
blob: bede7171519d159619cfa26e79bd6938c7f974cf (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
#!/usr/bin/env python3

import vpp_papi
import unittest, sys, time, threading, struct, logging
from ipaddress import *

papi_event = threading.Event()
def papi_event_handler(result):
    if result.vl_msg_id == vpp_papi.VL_API_SW_INTERFACE_SET_FLAGS:
        papi_event.set()
        return
    if result.vl_msg_id == vpp_papi.VL_API_VNET_INTERFACE_COUNTERS:
        format = '>' + str(int(len(result.data) / 8)) + 'Q'
        counters = struct.unpack(format, result.data)
        print('Counters:', counters)
        return
    if result.vl_msg_id == vpp_papi.VL_API_VNET_IP6_FIB_COUNTERS:
        print('IP6 FIB Counters:', result.count, len(result.c), len(result))
        i = 0
        # FIB counters allocate a large (1000 bytes) block so message length does not match reality
        for c in struct.iter_unpack('>16sBQQ', result.c):
            # In Python 3.5 we can use a tuple for prefix, length
            print(str(IPv6Address(c[0])) + '/' + str(c[1]), str(c[2]), str(c[3]))
            i += 1
            if i >= result.count:
                break
        return

    print('Unknown message id:', result.vl_msg_id)

class TestPAPI(unittest.TestCase):

    def setUp(self):
        r = vpp_papi.connect("test_papi")
        self.assertEqual(r, 0)

    def tearDown(self):
        r = vpp_papi.disconnect()
        self.assertEqual(r, 0)
        
    def test_show_version(self):
        t = vpp_papi.show_version()
        program = t.program.decode().rstrip('\x00')
        self.assertEqual('vpe', program)

    #
    # Add a few MAP domains, then dump them later
    #
    def test_map(self):
        t = vpp_papi.map_summary_stats()
        print(t)
        ip6 = IPv6Address(u'2001:db8::1').packed
        ip4 = IPv4Address(u'10.0.0.0').packed
        ip6_src = IPv6Address(u'2001:db9::1').packed
        t = vpp_papi.map_add_domain(ip6, ip4, ip6_src, 32, 24, 128, 0, 0, 6, 0, 0)
        print(t)
        self.assertEqual(t.retval, 0)

        ip4 = IPv4Address(u'10.0.1.0').packed
        t = vpp_papi.map_add_domain(ip6, ip4, ip6_src, 32, 24, 128, 0, 0, 6, 0, 0)
        print(t)
        self.assertEqual(t.retval, 0)

        t = vpp_papi.map_summary_stats()
        print(t)
        self.assertEqual(t.total_bindings, 2)

        t = vpp_papi.map_domain_dump()
        print (t)
        self.assertEqual(len(t), 2)

    def test_sw_interface_dump(self):
        #
        # Dump interfaces
        #
        t = vpp_papi.sw_interface_dump(0, b'ignored')
        for interface in t:
            if interface.vl_msg_id == vpp_papi.VL_API_SW_INTERFACE_DETAILS:
                print(interface.interface_name.decode())

    def test_want_interface_events(self):
        pid = 123
        vpp_papi.register_event_callback(papi_event_handler)
        papi_event.clear()
        t = vpp_papi.want_interface_events(True, pid)
        print (t)
        print('Setting interface up')
        t = vpp_papi.sw_interface_set_flags(0, 1, 1, 0)
        print (t)
        self.assertEqual(papi_event.wait(5), True)
        t = vpp_papi.sw_interface_set_flags(0, 0, 0, 0)
        print (t)
        self.assertEqual(papi_event.wait(5), True)

    @unittest.skip("not quite ready yet")
    def test_want_stats(self):
        pid = 123
        vpp_papi.register_event_callback(papi_event_handler)
        papi_event.clear()
        t = vpp_papi.want_stats(True, pid)

        print (t)

        #
        # Wait for some stats
        #
        self.assertEqual(papi_event.wait(30), True)
        t = vpp_papi.want_stats(False, pid)
        print (t)

    def test_tap(self):
        pid = 123
        vpp_papi.register_event_callback(papi_event_handler)
        papi_event.clear()
        t = vpp_papi.want_stats(True, pid)

        print (t)

        t = vpp_papi.tap_connect(1, b'tap', b'foo', 1, 0)
        print (t)
        self.assertEqual(t.retval, 0)
        swifindex = t.sw_if_index

        t = vpp_papi.sw_interface_set_flags(swifindex, 1, 1, 0)
        print (t)
        self.assertEqual(t.retval, 0)        

        ip6 = IPv6Address(u'2001:db8::1').packed
        t = vpp_papi.sw_interface_add_del_address(swifindex, 1, 1, 0, 16, ip6)
        print (t)
        time.sleep(40)


if __name__ == '__main__':
    #logging.basicConfig(level=logging.DEBUG)
    unittest.main()