diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/test_udp.py | 15 | ||||
| -rw-r--r-- | test/vpp_ip.py | 135 | ||||
| -rw-r--r-- | test/vpp_papi_provider.py | 42 | ||||
| -rw-r--r-- | test/vpp_udp_encap.py | 46 |
4 files changed, 185 insertions, 53 deletions
diff --git a/test/test_udp.py b/test/test_udp.py index 230335ff169..b0c6ee9c46a 100644 --- a/test/test_udp.py +++ b/test/test_udp.py @@ -1,7 +1,7 @@ #!/usr/bin/env python import unittest from framework import VppTestCase, VppTestRunner -from vpp_udp_encap import VppUdpEncap +from vpp_udp_encap import * from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable, VppMplsLabel from scapy.packet import Raw @@ -100,19 +100,24 @@ class TestUdpEncap(VppTestCase): self.pg2.local_ip6, self.pg2.remote_ip6, 332, 442, - table_id=2, - is_ip6=1) + table_id=2) udp_encap_3 = VppUdpEncap(self, 3, self.pg3.local_ip6, self.pg3.remote_ip6, 333, 443, - table_id=3, - is_ip6=1) + table_id=3) udp_encap_0.add_vpp_config() udp_encap_1.add_vpp_config() udp_encap_2.add_vpp_config() udp_encap_3.add_vpp_config() + self.logger.info(self.vapi.cli("sh udp encap")) + + self.assertTrue(find_udp_encap(self, udp_encap_2)) + self.assertTrue(find_udp_encap(self, udp_encap_3)) + self.assertTrue(find_udp_encap(self, udp_encap_0)) + self.assertTrue(find_udp_encap(self, udp_encap_1)) + # # Routes via each UDP encap object - all combinations of v4 and v6. # diff --git a/test/vpp_ip.py b/test/vpp_ip.py new file mode 100644 index 00000000000..912d8430d81 --- /dev/null +++ b/test/vpp_ip.py @@ -0,0 +1,135 @@ +""" + IP Types + +""" + +from ipaddress import ip_address + + +class IpAddressFamily: + ADDRESS_IP4 = 0 + ADDRESS_IP6 = 1 + + +INVALID_INDEX = 0xffffffff + + +def compare_ip_address(api_address, py_address): + if 4 is py_address.version: + if py_address.packed == api_address.ip4.address: + return True + else: + if py_address.packed == api_address.ip6.address: + return True + return False + + +class VppIpAddressUnion(): + def __init__(self, addr): + self.addr = addr + self.ip_addr = ip_address(unicode(self.addr)) + + @property + def version(self): + return self.ip_addr.version + + @property + def address(self): + return self.addr + + def encode(self): + if self.ip_addr.version is 6: + return { + 'ip6': { + 'address': self.ip_addr.packed + }, + } + else: + return { + 'ip4': { + 'address': self.ip_addr.packed + }, + } + + +class VppIpAddress(): + def __init__(self, addr): + self.addr = VppIpAddressUnion(addr) + + def encode(self): + if self.addr.version is 6: + return { + 'af': IpAddressFamily.ADDRESS_IP6, + 'un': self.addr.encode() + } + else: + return { + 'af': IpAddressFamily.ADDRESS_IP4, + 'un': self.addr.encode() + } + + @property + def address(self): + return self.addr.address + + +class VppIpPrefix(): + def __init__(self, addr, len): + self.addr = VppIpAddress(addr) + self.len = len + + def __eq__(self, other): + if self.addr == other.addr and self.len == other.len: + return True + return False + + def encode(self): + return {'address': self.addr.encode(), + 'address_length': self.len} + + @property + def address(self): + return self.addr.address + + +class VppIpMPrefix(): + def __init__(self, saddr, gaddr, len): + self.saddr = saddr + self.gaddr = gaddr + self.len = len + self.ip_saddr = ip_address(unicode(self.saddr)) + self.ip_gaddr = ip_address(unicode(self.gaddr)) + + def encode(self): + + if 6 is self.ip_saddr.version: + prefix = { + 'af': IpAddressFamily.ADDRESS_IP6, + 'grp_address': { + 'ip6': { + 'address': self.ip_gaddr.packed + }, + }, + 'src_address': { + 'ip6': { + 'address': self.ip_saddr.packed + }, + }, + 'grp_address_length': self.len, + } + else: + prefix = { + 'af': IpAddressFamily.ADDRESS_IP4, + 'grp_address': { + 'ip4': { + 'address': self.ip_gaddr.packed + }, + }, + 'src_address': { + 'ip4': { + 'address': self.ip_saddr.packed + }, + }, + 'grp_address_length': self.len, + } + return prefix diff --git a/test/vpp_papi_provider.py b/test/vpp_papi_provider.py index bd4787d3e9e..f9ac7694bf6 100644 --- a/test/vpp_papi_provider.py +++ b/test/vpp_papi_provider.py @@ -1149,15 +1149,13 @@ class VppPapiProvider(object): 'session_id': session_id} ) - def udp_encap_add_del(self, - id, - src_ip, - dst_ip, - src_port, - dst_port, - table_id=0, - is_add=1, - is_ip6=0): + def udp_encap_add(self, + id, + src_ip, + dst_ip, + src_port, + dst_port, + table_id=0): """ Add a GRE tunnel :param id: user provided ID :param src_ip: @@ -1165,21 +1163,23 @@ class VppPapiProvider(object): :param src_port: :param dst_port: :param outer_fib_id: (Default value = 0) - :param is_add: (Default value = 1) - :param is_ipv6: (Default value = 0) """ return self.api( - self.papi.udp_encap_add_del, - {'id': id, - 'is_add': is_add, - 'is_ip6': is_ip6, - 'src_ip': src_ip, - 'dst_ip': dst_ip, - 'src_port': src_port, - 'dst_port': dst_port, - 'table_id': table_id} - ) + self.papi.udp_encap_add, + { + 'udp_encap': { + 'id': id, + 'src_ip': src_ip, + 'dst_ip': dst_ip, + 'src_port': src_port, + 'dst_port': dst_port, + 'table_id': table_id + } + }) + + def udp_encap_del(self, id): + return self.api(self.papi.udp_encap_del, {'id': id}) def udp_encap_dump(self): return self.api(self.papi.udp_encap_dump, {}) diff --git a/test/vpp_udp_encap.py b/test/vpp_udp_encap.py index d4daa7430b3..002f9f4023a 100644 --- a/test/vpp_udp_encap.py +++ b/test/vpp_udp_encap.py @@ -5,13 +5,21 @@ from vpp_object import * from socket import inet_pton, inet_ntop, AF_INET, AF_INET6 +from vpp_ip import * -def find_udp_encap(test, id): +def find_udp_encap(test, ue): encaps = test.vapi.udp_encap_dump() for e in encaps: - if id == e.id: + if ue.id == e.udp_encap.id \ + and compare_ip_address(e.udp_encap.src_ip.un, + ue.src_ip.addr.ip_addr) \ + and compare_ip_address(e.udp_encap.dst_ip.un, + ue.dst_ip.addr.ip_addr) \ + and e.udp_encap.dst_port == ue.dst_port \ + and e.udp_encap.src_port == ue.src_port: return True + return False @@ -24,48 +32,32 @@ class VppUdpEncap(VppObject): dst_ip, src_port, dst_port, - table_id=0, - is_ip6=0): + table_id=0): self._test = test self.id = id self.table_id = table_id - self.is_ip6 = is_ip6 self.src_ip_s = src_ip self.dst_ip_s = dst_ip - if is_ip6: - self.src_ip = inet_pton(AF_INET6, src_ip) - self.dst_ip = inet_pton(AF_INET6, dst_ip) - else: - self.src_ip = inet_pton(AF_INET, src_ip) - self.dst_ip = inet_pton(AF_INET, dst_ip) + self.src_ip = VppIpAddress(src_ip) + self.dst_ip = VppIpAddress(dst_ip) self.src_port = src_port self.dst_port = dst_port def add_vpp_config(self): - self._test.vapi.udp_encap_add_del( + self._test.vapi.udp_encap_add( self.id, - self.src_ip, - self.dst_ip, + self.src_ip.encode(), + self.dst_ip.encode(), self.src_port, self.dst_port, - self.table_id, - is_ip6=self.is_ip6, - is_add=1) + self.table_id) self._test.registry.register(self, self._test.logger) def remove_vpp_config(self): - self._test.vapi.udp_encap_add_del( - self.id, - self.src_ip, - self.dst_ip, - self.src_port, - self.dst_port, - self.table_id, - is_ip6=self.is_ip6, - is_add=0) + self._test.vapi.udp_encap_del(self.id) def query_vpp_config(self): - return find_udp_encap(self._test, self.id) + return find_udp_encap(self._test, self) def __str__(self): return self.object_id() |