Python API: Change from cPython to CFFI. - vpp

diff options

author	Ole Troan <ot@cisco.com>	2017-03-08 12:02:24 +0100
committer	Ole Troan <ot@cisco.com>	2017-03-15 21:43:30 +0100
commit	3cc4971882235a539bc6177e8e4b4d92129b3a12 (patch)
tree	c850b1ea3348da7cd632e782008fbe4e3c6b4c87 /build-root
parent	9d676afbb779da5186cb3869925ef6d7d3d04db1 (diff)

Python API: Change from cPython to CFFI.

Change-Id: I03e52466fb3f909ae52b8fba601168f3eadbd972 Signed-off-by: Ole Troan <ot@cisco.com>

Diffstat (limited to 'build-root')

-rw-r--r--

build-root/rpm/vpp.spec

1 files changed, 3 insertions, 9 deletions

#!/usr/bin/env python3 """ L2BD ARP term Test """ import unittest import random import copy from socket import AF_INET, AF_INET6, inet_pton, inet_ntop from scapy.packet import Raw from scapy.layers.l2 import Ether, ARP from scapy.layers.inet import IP from scapy.utils6 import ( in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, in6_mactoifaceid, in6_ismaddr, ) from scapy.layers.inet6 import ( IPv6, UDP, ICMPv6ND_NS, ICMPv6ND_RS, ICMPv6ND_RA, ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation, ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types, ) from framework import VppTestCase, VppTestRunner from util import Host, ppp class TestL2bdArpTerm(VppTestCase): """L2BD arp termination Test Case""" @classmethod def setUpClass(cls): """ Perform standard class setup (defined by class method setUpClass in class VppTestCase) before running the test case, set test case related variables and configure VPP. """ super(TestL2bdArpTerm, cls).setUpClass() try: # Create pg interfaces n_bd = 1 cls.ifs_per_bd = ifs_per_bd = 3 n_ifs = n_bd * ifs_per_bd cls.create_pg_interfaces(range(n_ifs)) # Set up all interfaces for i in cls.pg_interfaces: i.admin_up() cls.hosts = set() except Exception: super(TestL2bdArpTerm, cls).tearDownClass() raise @classmethod def tearDownClass(cls): super(TestL2bdArpTerm, cls).tearDownClass() def setUp(self): """ Clear trace and packet infos before running each test. """ self.reset_packet_infos() super(TestL2bdArpTerm, self).setUp() def tearDown(self): """ Show various debug prints after each test. """ super(TestL2bdArpTerm, self).tearDown() def show_commands_at_teardown(self): self.logger.info(self.vapi.ppcli("show l2fib verbose")) # many tests delete bridge-domain 1 as the last task. don't output # the details of a non-existent bridge-domain. if self.vapi.l2_fib_table_dump(bd_id=1): self.logger.info(self.vapi.ppcli("show bridge-domain 1 detail")) def add_del_arp_term_hosts(self, entries, bd_id=1, is_add=1, is_ipv6=0): for e in entries: ip = e.ip4 if is_ipv6 == 0 else e.ip6 self.vapi.bd_ip_mac_add_del( is_add=is_add, entry={"bd_id": bd_id, "ip": ip, "mac": e.mac} ) @classmethod def mac_list(cls, b6_range): return ["00:00:ca:fe:00:%02x" % b6 for b6 in b6_range] @classmethod def ip4_host(cls, subnet, host, mac): return Host(mac=mac, ip4="172.17.1%02u.%u" % (subnet, host)) @classmethod def ip4_hosts(cls, subnet, start, mac_list): return { cls.ip4_host(subnet, start + j, mac_list[j]) for j in range(len(mac_list)) } @classmethod def ip6_host(cls, subnet, host, mac): return Host(mac=mac, ip6="fd01:%x::%x" % (subnet, host)) @classmethod def ip6_hosts(cls, subnet, start, mac_list): return { cls.ip6_host(subnet, start + j, mac_list[j]) for j in range(len(mac_list)) } @classmethod def bd_swifs(cls, b): n = cls.ifs_per_bd start = (b - 1) * n return [cls.pg_interfaces[j] for j in range(start, start + n)] def bd_add_del(self, bd_id=1, is_add=1): if is_add: self.vapi.bridge_domain_add_del_v2( bd_id=bd_id, is_add=is_add, flood=1, uu_flood=1, forward=1, learn=1 ) for swif in self.bd_swifs(bd_id): swif_idx = swif.sw_if_index self.vapi.sw_interface_set_l2_bridge( rx_sw_if_index=swif_idx, bd_id=bd_id, enable=is_add ) if not is_add: self.vapi.bridge_domain_add_del_v2(bd_id=bd_id, is_add=is_add) @classmethod def arp_req(cls, src_host, host): return Ether(dst="ff:ff:ff:ff:ff:ff", src=src_host.mac) / ARP( op="who-has", hwsrc=src_host.bin_mac, pdst=host.ip4, psrc=src_host.ip4 ) @classmethod def arp_reqs(cls, src_host, entries): return [cls.arp_req(src_host, e) for e in entries] @classmethod def garp_req(cls, host): return cls.arp_req(host, host) @classmethod def garp_reqs(cls, entries): return [cls.garp_req(e) for e in entries] def arp_resp_host(self, src_host, arp_resp): ether = arp_resp[Ether] self.assertEqual(ether.dst, src_host.mac) arp = arp_resp[ARP] self.assertEqual(arp.hwtype, 1) self.assertEqual(arp.ptype, 0x800) self.assertEqual(arp.hwlen, 6) self.assertEqual(arp.plen, 4) arp_opts = {"who-has": 1, "is-at": 2} self.assertEqual(arp.op, arp_opts["is-at"]) self.assertEqual(arp.hwdst, src_host.mac) self.assertEqual(arp.pdst, src_host.ip4) return Host(mac=arp.hwsrc, ip4=arp.psrc) def arp_resp_hosts(self, src_host, pkts): return {self.arp_resp_host(src_host, p) for p in pkts} @staticmethod def inttoip4(ip): o1 = int(ip / 16777216) % 256 o2 = int(ip / 65536) % 256 o3 = int(ip / 256) % 256 o4 = int(ip) % 256 return "%s.%s.%s.%s" % (o1, o2, o3, o4) def arp_event_host(self, e): return Host(str(e.mac), ip4=str(e.ip)) def arp_event_hosts(self, evs): return {self.arp_event_host(e) for e in evs} def nd_event_host(self, e): return Host(str(e.mac), ip6=str(e.ip)) def nd_event_hosts(self, evs): return {self.nd_event_host(e) for e in evs} @classmethod def ns_req(cls, src_host, host): nsma = in6_getnsma(inet_pton(AF_INET6, "fd10::ffff")) d = inet_ntop(AF_INET6, nsma) return ( Ether(dst="ff:ff:ff:ff:ff:ff", src=src_host.mac) / IPv6(dst=d, src=src_host.ip6) / ICMPv6ND_NS(tgt=host.ip6) / ICMPv6NDOptSrcLLAddr(lladdr=src_host.mac) ) @classmethod def ns_reqs_dst(cls, entries, dst_host): return [cls.ns_req(e, dst_host) for e in entries] @classmethod def ns_reqs_src(cls, src_host, entries): return [cls.ns_req(src_host, e) for e in entries] def na_resp_host(self, src_host, rx): self.assertEqual(rx[Ether].dst, src_host.mac) self.assertEqual(in6_ptop(rx[IPv6].dst), in6_ptop(src_host.ip6)) self.assertTrue(rx.haslayer(ICMPv6ND_NA)) self.assertTrue(rx.haslayer(ICMPv6NDOptDstLLAddr)) na = rx[ICMPv6ND_NA] return Host(mac=na.lladdr, ip6=na.tgt) def na_resp_hosts(self, src_host, pkts): return {self.na_resp_host(src_host, p) for p in pkts} def set_bd_flags(self, bd_id, **args): """ Enable/disable defined feature(s) of the bridge domain. :param int bd_id: Bridge domain ID. :param list args: List of feature/status pairs. Allowed features: \ learn, forward, flood, uu_flood and arp_term. Status False means \ disable, status True means enable the feature. :raise: ValueError in case of unknown feature in the input. """ for flag in args: if flag == "learn": feature_bitmap = 1 << 0 elif flag == "forward": feature_bitmap = 1 << 1 elif flag == "flood": feature_bitmap = 1 << 2 elif flag == "uu_flood": feature_bitmap = 1 << 3 elif flag == "arp_term": feature_bitmap = 1 << 4 else: raise ValueError("Unknown feature used: %s" % flag) is_set = 1 if args[flag] else 0 self.vapi.bridge_flags(bd_id=bd_id, is_set=is_set, flags=feature_bitmap) self.logger.info("Bridge domain ID %d updated" % bd_id) def verify_arp(self, src_host, req_hosts, resp_hosts, bd_id=1): reqs = self.arp_reqs(src_host, req_hosts) for swif in self.bd_swifs(bd_id): swif.add_stream(reqs) self.pg_enable_capture(self.pg_interfaces) self.pg_start() for swif in self.bd_swifs(bd_id): resp_pkts = swif.get_capture(len(resp_hosts)) resps = self.arp_resp_hosts(src_host, resp_pkts) self.assertEqual(len(resps ^ resp_hosts), 0) def verify_nd(self, src_host, req_hosts, resp_hosts, bd_id=1): reqs = self.ns_reqs_src(src_host, req_hosts) for swif in self.bd_swifs(bd_id): swif.add_stream(reqs) self.pg_enable_capture(self.pg_interfaces) self.pg_start() for swif in self.bd_swifs(bd_id): resp_pkts = swif.get_capture(len(resp_hosts)) resps = self.na_resp_hosts(src_host, resp_pkts) self.assertEqual(len(resps ^ resp_hosts), 0) def test_l2bd_arp_term_01(self): """L2BD arp term - add 5 hosts, verify arp responses""" src_host = self.ip4_host(50, 50, "00:00:11:22:33:44") self.bd_add_del(1, is_add=1) self.set_bd_flags(1, arp_term=True, flood=False, uu_flood=False, learn=False) macs = self.mac_list(range(1, 5)) hosts = self.ip4_hosts(4, 1, macs) self.add_del_arp_term_hosts(hosts, is_add=1) self.verify_arp(src_host, hosts, hosts) type(self).hosts = hosts def test_l2bd_arp_term_02(self): """L2BD arp term - delete 3 hosts, verify arp responses""" src_host = self.ip4_host(50, 50, "00:00:11:22:33:44") macs = self.mac_list(range(1, 3)) deleted = self.ip4_hosts(4, 1, macs) self.add_del_arp_term_hosts(deleted, is_add=0) remaining = self.hosts - deleted self.verify_arp(src_host, self.hosts, remaining) type(self).hosts = remaining self.bd_add_del(1, is_add=0) def test_l2bd_arp_term_03(self): """L2BD arp term - recreate BD1, readd 3 hosts, verify arp responses""" src_host = self.ip4_host(50, 50, "00:00:11:22:33:44") self.bd_add_del(1, is_add=1) self.set_bd_flags(1, arp_term=True, flood=False, uu_flood=False, learn=False) macs = self.mac_list(range(1, 3)) readded = self.ip4_hosts(4, 1, macs) self.add_del_arp_term_hosts(readded, is_add=1) self.verify_arp(src_host, self.hosts | readded, readded) type(self).hosts = readded def test_l2bd_arp_term_04(self): """L2BD arp term - 2 IP4 addrs per host""" src_host = self.ip4_host(50, 50, "00:00:11:22:33:44") macs = self.mac_list(range(1, 3)) sub5_hosts = self.ip4_hosts(5, 1, macs) self.add_del_arp_term_hosts(sub5_hosts, is_add=1) hosts = self.hosts | sub5_hosts self.verify_arp(src_host, hosts, hosts) type(self).hosts = hosts self.bd_add_del(1, is_add=0) def test_l2bd_arp_term_05(self): """L2BD arp term - create and update 10 IP4-mac pairs""" src_host = self.ip4_host(50, 50, "00:00:11:22:33:44") self.bd_add_del(1, is_add=1) self.set_bd_flags(1, arp_term=True, flood=False, uu_flood=False, learn=False) macs1 = self.mac_list(range(10, 20)) hosts1 = self.ip4_hosts(5, 1, macs1) self.add_del_arp_term_hosts(hosts1, is_add=1) self.verify_arp(src_host, hosts1, hosts1) macs2 = self.mac_list(range(20, 30)) hosts2 = self.ip4_hosts(5, 1, macs2) self.add_del_arp_term_hosts(hosts2, is_add=1) self.verify_arp(src_host, hosts1, hosts2) self.bd_add_del(1, is_add=0) def test_l2bd_arp_term_06(self): """L2BD arp/ND term - hosts with both ip4/ip6""" src_host4 = self.ip4_host(50, 50, "00:00:11:22:33:44") src_host6 = self.ip6_host(50, 50, "00:00:11:22:33:44") self.bd_add_del(1, is_add=1) # enable flood to make sure requests are not flooded self.set_bd_flags(1, arp_term=True, flood=True, uu_flood=False, learn=False) macs = self.mac_list(range(10, 20)) hosts6 = self.ip6_hosts(5, 1, macs) hosts4 = self.ip4_hosts(5, 1, macs) self.add_del_arp_term_hosts(hosts4, is_add=1) self.add_del_arp_term_hosts(hosts6, is_add=1, is_ipv6=1) self.verify_arp(src_host4, hosts4, hosts4) self.verify_nd(src_host6, hosts6, hosts6) self.bd_add_del(1, is_add=0) def test_l2bd_arp_term_07(self): """L2BD ND term - Add and Del hosts, verify ND replies""" src_host6 = self.ip6_host(50, 50, "00:00:11:22:33:44") self.bd_add_del(1, is_add=1) self.set_bd_flags(1, arp_term=True, flood=False, uu_flood=False, learn=False) macs = self.mac_list(range(10, 20)) hosts6 = self.ip6_hosts(5, 1, macs) self.add_del_arp_term_hosts(hosts6, is_add=1, is_ipv6=1) self.verify_nd(src_host6, hosts6, hosts6) del_macs = self.mac_list(range(10, 15)) deleted = self.ip6_hosts(5, 1, del_macs) self.add_del_arp_term_hosts(deleted, is_add=0, is_ipv6=1) self.verify_nd(src_host6, hosts6, hosts6 - deleted) self.bd_add_del(1, is_add=0) def test_l2bd_arp_term_08(self): """L2BD ND term - Add and update IP+mac, verify ND replies""" src_host = self.ip6_host(50, 50, "00:00:11:22:33:44") self.bd_add_del(1, is_add=1) self.set_bd_flags(1, arp_term=True, flood=False, uu_flood=False, learn=False) macs1 = self.mac_list(range(10, 20)) hosts = self.ip6_hosts(5, 1, macs1) self.add_del_arp_term_hosts(hosts, is_add=1, is_ipv6=1) self.verify_nd(src_host, hosts, hosts) macs2 = self.mac_list(range(20, 30)) updated = self.ip6_hosts(5, 1, macs2) self.add_del_arp_term_hosts(updated, is_add=1, is_ipv6=1) self.verify_nd(src_host, hosts, updated) self.bd_add_del(1, is_add=0) def test_l2bd_arp_term_09(self): """L2BD arp term - send garps, verify arp event reports""" self.vapi.want_l2_arp_term_events(enable=1) self.bd_add_del(1, is_add=1) self.set_bd_flags(1, arp_term=True, flood=False, uu_flood=False, learn=False) macs = self.mac_list(range(90, 95)) hosts = self.ip4_hosts(5, 1, macs) garps = self.garp_reqs(hosts) self.bd_swifs(1)[0].add_stream(garps) self.pg_enable_capture(self.pg_interfaces) self.pg_start() evs = [ self.vapi.wait_for_event(1, "l2_arp_term_event") for i in range(len(hosts)) ] ev_hosts = self.arp_event_hosts(evs) self.assertEqual(len(ev_hosts ^ hosts), 0) def test_l2bd_arp_term_10(self): """L2BD arp term - send duplicate garps, verify suppression""" macs = self.mac_list(range(70, 71)) hosts = self.ip4_hosts(6, 1, macs) """ send the packet 5 times expect one event """ garps = self.garp_reqs(hosts) * 5 self.bd_swifs(1)[0].add_stream(garps) self.pg_enable_capture(self.pg_interfaces) self.pg_start() evs = [ self.vapi.wait_for_event(1, "l2_arp_term_event") for i in range(len(hosts)) ] ev_hosts = self.arp_event_hosts(evs) self.assertEqual(len(ev_hosts ^ hosts), 0) def test_l2bd_arp_term_11(self): """L2BD arp term - disable ip4 arp events,send garps, verify no events""" self.vapi.want_l2_arp_term_events(enable=0) macs = self.mac_list(range(90, 95)) hosts = self.ip4_hosts(5, 1, macs) garps = self.garp_reqs(hosts) self.bd_swifs(1)[0].add_stream(garps) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.sleep(1) self.assertEqual(len(self.vapi.collect_events()), 0) self.bd_add_del(1, is_add=0) def test_l2bd_arp_term_12(self): """L2BD ND term - send NS packets verify reports""" self.vapi.want_l2_arp_term_events(enable=1) dst_host = self.ip6_host(50, 50, "00:00:11:22:33:44") self.bd_add_del(1, is_add=1) self.set_bd_flags(1, arp_term=True, flood=False, uu_flood=False, learn=False) macs = self.mac_list(range(10, 15)) hosts = self.ip6_hosts(5, 1, macs) reqs = self.ns_reqs_dst(hosts, dst_host) self.bd_swifs(1)[0].add_stream(reqs) self.pg_enable_capture(self.pg_interfaces) self.pg_start() evs = [ self.vapi.wait_for_event(2, "l2_arp_term_event") for i in range(len(hosts)) ] ev_hosts = self.nd_event_hosts(evs) self.assertEqual(len(ev_hosts ^ hosts), 0) def test_l2bd_arp_term_13(self): """L2BD ND term - send duplicate ns, verify suppression""" dst_host = self.ip6_host(50, 50, "00:00:11:22:33:44") macs = self.mac_list(range(16, 17)) hosts = self.ip6_hosts(5, 1, macs) reqs = self.ns_reqs_dst(hosts, dst_host) * 5 self.bd_swifs(1)[0].add_stream(reqs) self.pg_enable_capture(self.pg_interfaces) self.pg_start() evs = [ self.vapi.wait_for_event(2, "l2_arp_term_event") for i in range(len(hosts)) ] ev_hosts = self.nd_event_hosts(evs) self.assertEqual(len(ev_hosts ^ hosts), 0) def test_l2bd_arp_term_14(self): """L2BD ND term - disable ip4 arp events,send ns, verify no events""" self.vapi.want_l2_arp_term_events(enable=0) dst_host = self.ip6_host(50, 50, "00:00:11:22:33:44") macs = self.mac_list(range(10, 15)) hosts = self.ip6_hosts(5, 1, macs) reqs = self.ns_reqs_dst(hosts, dst_host) self.bd_swifs(1)[0].add_stream(reqs) self.pg_enable_capture(self.pg_interfaces) self.pg_start() self.sleep(1) self.assertEqual(len(self.vapi.collect_events()), 0) self.bd_add_del(1, is_add=0) if __name__ == "__main__": unittest.main(testRunner=VppTestRunner)


context:
space:
mode: