misc: update my email-address

[vpp.git] / test / test_abf.py

diff --git a/test/test_abf.py b/test/test_abf.py
index ecd8c53..097476b 100644 (file)
--- a/test/test_abf.py
+++ b/test/test_abf.py
@@ -1,19 +1,24 @@
-#!/usr/bin/env python
+#!/usr/bin/env python3

 
 from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
 import unittest
 
 from framework import VppTestCase, VppTestRunner
 from vpp_ip import DpoProto
 
 from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
 import unittest
 
 from framework import VppTestCase, VppTestRunner
 from vpp_ip import DpoProto

-from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsLabel, VppIpTable
+from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsLabel, \
+    VppIpTable, FibPathProto
+from vpp_acl import AclRule, VppAcl

 
 from scapy.packet import Raw
 from scapy.layers.l2 import Ether
 from scapy.layers.inet import IP, UDP
 from scapy.layers.inet6 import IPv6
 
 from scapy.packet import Raw
 from scapy.layers.l2 import Ether
 from scapy.layers.inet import IP, UDP
 from scapy.layers.inet6 import IPv6

+from ipaddress import IPv4Network, IPv6Network

 
 from vpp_object import VppObject
 
 
 from vpp_object import VppObject
 

+NUM_PKTS = 67
+

 
 def find_abf_policy(test, id):
     policies = test.vapi.abf_policy_dump()
 
 def find_abf_policy(test, id):
     policies = test.vapi.abf_policy_dump()


@@ -43,30 +48,9 @@ class VppAbfPolicy(VppObject):
         self.policy_id = policy_id
         self.acl = acl
         self.paths = paths
         self.policy_id = policy_id
         self.acl = acl
         self.paths = paths

-
-    def encode_paths(self):
-        br_paths = []
-        for p in self.paths:
-            lstack = []
-            for l in p.nh_labels:
-                if type(l) == VppMplsLabel:
-                    lstack.append(l.encode())
-                else:
-                    lstack.append({'label': l, 'ttl': 255})
-            n_labels = len(lstack)
-            while (len(lstack) < 16):
-                lstack.append({})
-            br_paths.append({'next_hop': p.nh_addr,
-                             'weight': 1,
-                             'afi': p.proto,
-                             'sw_if_index': 0xffffffff,
-                             'preference': 0,
-                             'table_id': p.nh_table_id,
-                             'next_hop_id': p.next_hop_id,
-                             'is_udp_encap': p.is_udp_encap,
-                             'n_labels': n_labels,
-                             'label_stack': lstack})
-        return br_paths
+        self.encoded_paths = []
+        for path in self.paths:
+            self.encoded_paths.append(path.encode())

 
     def add_vpp_config(self):
         self._test.vapi.abf_policy_add_del(
 
     def add_vpp_config(self):
         self._test.vapi.abf_policy_add_del(


@@ -74,7 +58,7 @@ class VppAbfPolicy(VppObject):
             {'policy_id': self.policy_id,
              'acl_index': self.acl.acl_index,
              'n_paths': len(self.paths),
             {'policy_id': self.policy_id,
              'acl_index': self.acl.acl_index,
              'n_paths': len(self.paths),

-             'paths': self.encode_paths()})
+             'paths': self.encoded_paths})

         self._test.registry.register(self, self._test.logger)
 
     def remove_vpp_config(self):
         self._test.registry.register(self, self._test.logger)
 
     def remove_vpp_config(self):


@@ -83,14 +67,11 @@ class VppAbfPolicy(VppObject):
             {'policy_id': self.policy_id,
              'acl_index': self.acl.acl_index,
              'n_paths': len(self.paths),
             {'policy_id': self.policy_id,
              'acl_index': self.acl.acl_index,
              'n_paths': len(self.paths),

-             'paths': self.encode_paths()})
+             'paths': self.encoded_paths})

 
     def query_vpp_config(self):
         return find_abf_policy(self._test, self.policy_id)
 
 
     def query_vpp_config(self):
         return find_abf_policy(self._test, self.policy_id)
 

-    def __str__(self):
-        return self.object_id()
-

     def object_id(self):
         return ("abf-policy-%d" % self.policy_id)
 
     def object_id(self):
         return ("abf-policy-%d" % self.policy_id)
 


@@ -131,9 +112,6 @@ class VppAbfAttach(VppObject):
                                    self.policy_id,
                                    self.sw_if_index)
 
                                    self.policy_id,
                                    self.sw_if_index)
 

-    def __str__(self):
-        return self.object_id()
-

     def object_id(self):
         return ("abf-attach-%d-%d" % (self.policy_id, self.sw_if_index))
 
     def object_id(self):
         return ("abf-attach-%d-%d" % (self.policy_id, self.sw_if_index))
 


@@ -141,6 +119,14 @@ class VppAbfAttach(VppObject):
 class TestAbf(VppTestCase):
     """ ABF Test Case """
 
 class TestAbf(VppTestCase):
     """ ABF Test Case """
 

+    @classmethod
+    def setUpClass(cls):
+        super(TestAbf, cls).setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestAbf, cls).tearDownClass()
+

     def setUp(self):
         super(TestAbf, self).setUp()
 
     def setUp(self):
         super(TestAbf, self).setUp()
 


@@ -157,7 +143,6 @@ class TestAbf(VppTestCase):
         for i in self.pg_interfaces:
             i.unconfig_ip4()
             i.unconfig_ip6()
         for i in self.pg_interfaces:
             i.unconfig_ip4()
             i.unconfig_ip6()

-            i.ip6_disable()

             i.admin_down()
         super(TestAbf, self).tearDown()
 
             i.admin_down()
         super(TestAbf, self).tearDown()
 


@@ -178,18 +163,11 @@ class TestAbf(VppTestCase):
         #
         # Rule 1
         #
         #
         # Rule 1
         #

-        rule_1 = ({'is_permit': 1,
-                   'is_ipv6': 0,
-                   'proto': 17,
-                   'srcport_or_icmptype_first': 1234,
-                   'srcport_or_icmptype_last': 1234,
-                   'src_ip_prefix_len': 32,
-                   'src_ip_addr': inet_pton(AF_INET, "1.1.1.1"),
-                   'dstport_or_icmpcode_first': 1234,
-                   'dstport_or_icmpcode_last': 1234,
-                   'dst_ip_prefix_len': 32,
-                   'dst_ip_addr': inet_pton(AF_INET, "1.1.1.2")})
-        acl_1 = self.vapi.acl_add_replace(acl_index=4294967295, r=[rule_1])
+        rule_1 = AclRule(is_permit=1, proto=17, ports=1234,
+                         src_prefix=IPv4Network("1.1.1.1/32"),
+                         dst_prefix=IPv4Network("1.1.1.2/32"))
+        acl_1 = VppAcl(self, rules=[rule_1])
+        acl_1.add_vpp_config()

 
         #
         # ABF policy for ACL 1 - path via interface 1
 
         #
         # ABF policy for ACL 1 - path via interface 1


@@ -213,8 +191,8 @@ class TestAbf(VppTestCase):
                      dst=self.pg0.local_mac) /
                IP(src="1.1.1.1", dst="1.1.1.2") /
                UDP(sport=1234, dport=1234) /
                      dst=self.pg0.local_mac) /
                IP(src="1.1.1.1", dst="1.1.1.2") /
                UDP(sport=1234, dport=1234) /

-               Raw('\xa5' * 100))
-        self.send_and_expect(self.pg0, p_1*65, self.pg1)
+               Raw(b'\xa5' * 100))
+        self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg1)

 
         #
         # Attach a 'better' priority policy to the same interface
 
         #
         # Attach a 'better' priority policy to the same interface


@@ -226,7 +204,7 @@ class TestAbf(VppTestCase):
         attach_2 = VppAbfAttach(self, 11, self.pg0.sw_if_index, 40)
         attach_2.add_vpp_config()
 
         attach_2 = VppAbfAttach(self, 11, self.pg0.sw_if_index, 40)
         attach_2.add_vpp_config()
 

-        self.send_and_expect(self.pg0, p_1*65, self.pg2)
+        self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg2)

 
         #
         # Attach a policy with priority in the middle
 
         #
         # Attach a policy with priority in the middle


@@ -238,13 +216,13 @@ class TestAbf(VppTestCase):
         attach_3 = VppAbfAttach(self, 12, self.pg0.sw_if_index, 45)
         attach_3.add_vpp_config()
 
         attach_3 = VppAbfAttach(self, 12, self.pg0.sw_if_index, 45)
         attach_3.add_vpp_config()
 

-        self.send_and_expect(self.pg0, p_1*65, self.pg2)
+        self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg2)

 
         #
         # remove the best priority
         #
         attach_2.remove_vpp_config()
 
         #
         # remove the best priority
         #
         attach_2.remove_vpp_config()

-        self.send_and_expect(self.pg0, p_1*65, self.pg3)
+        self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg3)

 
         #
         # Attach one of the same policies to Pg1
 
         #
         # Attach one of the same policies to Pg1


@@ -256,15 +234,15 @@ class TestAbf(VppTestCase):
                      dst=self.pg1.local_mac) /
                IP(src="1.1.1.1", dst="1.1.1.2") /
                UDP(sport=1234, dport=1234) /
                      dst=self.pg1.local_mac) /
                IP(src="1.1.1.1", dst="1.1.1.2") /
                UDP(sport=1234, dport=1234) /

-               Raw('\xa5' * 100))
-        self.send_and_expect(self.pg1, p_2 * 65, self.pg3)
+               Raw(b'\xa5' * 100))
+        self.send_and_expect(self.pg1, p_2 * NUM_PKTS, self.pg3)

 
         #
         # detach the policy from PG1, now expect traffic to be dropped
         #
         attach_4.remove_vpp_config()
 
 
         #
         # detach the policy from PG1, now expect traffic to be dropped
         #
         attach_4.remove_vpp_config()
 

-        self.send_and_assert_no_replies(self.pg1, p_2 * 65, "Detached")
+        self.send_and_assert_no_replies(self.pg1, p_2 * NUM_PKTS, "Detached")

 
         #
         # Swap to route via a next-hop in the non-default table
 
         #
         # Swap to route via a next-hop in the non-default table


@@ -285,7 +263,7 @@ class TestAbf(VppTestCase):
         attach_5 = VppAbfAttach(self, 13, self.pg0.sw_if_index, 30)
         attach_5.add_vpp_config()
 
         attach_5 = VppAbfAttach(self, 13, self.pg0.sw_if_index, 30)
         attach_5.add_vpp_config()
 

-        self.send_and_expect(self.pg0, p_1*65, self.pg4)
+        self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg4)

 
         self.pg4.unconfig_ip4()
         self.pg4.set_table_ip4(0)
 
         self.pg4.unconfig_ip4()
         self.pg4.set_table_ip4(0)


@@ -301,27 +279,18 @@ class TestAbf(VppTestCase):
         #
         # Rule 1
         #
         #
         # Rule 1
         #

-        rule_1 = ({'is_permit': 1,
-                   'is_ipv6': 1,
-                   'proto': 17,
-                   'srcport_or_icmptype_first': 1234,
-                   'srcport_or_icmptype_last': 1234,
-                   'src_ip_prefix_len': 128,
-                   'src_ip_addr': inet_pton(AF_INET6, "2001::2"),
-                   'dstport_or_icmpcode_first': 1234,
-                   'dstport_or_icmpcode_last': 1234,
-                   'dst_ip_prefix_len': 128,
-                   'dst_ip_addr': inet_pton(AF_INET6, "2001::1")})
-        acl_1 = self.vapi.acl_add_replace(acl_index=4294967295,
-                                          r=[rule_1])
+        rule_1 = AclRule(is_permit=1, proto=17, ports=1234,
+                         src_prefix=IPv6Network("2001::2/128"),
+                         dst_prefix=IPv6Network("2001::1/128"))
+        acl_1 = VppAcl(self, rules=[rule_1])
+        acl_1.add_vpp_config()

 
         #
         # ABF policy for ACL 1 - path via interface 1
         #
         abf_1 = VppAbfPolicy(self, 10, acl_1,
                              [VppRoutePath("3001::1",
 
         #
         # ABF policy for ACL 1 - path via interface 1
         #
         abf_1 = VppAbfPolicy(self, 10, acl_1,
                              [VppRoutePath("3001::1",

-                                           0xffffffff,
-                                           proto=DpoProto.DPO_PROTO_IP6)])
+                                           0xffffffff)])

         abf_1.add_vpp_config()
 
         attach_1 = VppAbfAttach(self, 10, self.pg0.sw_if_index,
         abf_1.add_vpp_config()
 
         attach_1 = VppAbfAttach(self, 10, self.pg0.sw_if_index,


@@ -335,28 +304,26 @@ class TestAbf(VppTestCase):
                    dst=self.pg0.local_mac) /
              IPv6(src="2001::2", dst="2001::1") /
              UDP(sport=1234, dport=1234) /
                    dst=self.pg0.local_mac) /
              IPv6(src="2001::2", dst="2001::1") /
              UDP(sport=1234, dport=1234) /

-             Raw('\xa5' * 100))
+             Raw(b'\xa5' * 100))

 
         #
         # packets are dropped because there is no route to the policy's
         # next hop
         #
 
         #
         # packets are dropped because there is no route to the policy's
         # next hop
         #

-        self.send_and_assert_no_replies(self.pg1, p * 65, "no route")
+        self.send_and_assert_no_replies(self.pg1, p * NUM_PKTS, "no route")

 
         #
         # add a route resolving the next-hop
         #
         route = VppIpRoute(self, "3001::1", 32,
                            [VppRoutePath(self.pg1.remote_ip6,
 
         #
         # add a route resolving the next-hop
         #
         route = VppIpRoute(self, "3001::1", 32,
                            [VppRoutePath(self.pg1.remote_ip6,

-                                         self.pg1.sw_if_index,
-                                         proto=DpoProto.DPO_PROTO_IP6)],
-                           is_ip6=1)
+                                         self.pg1.sw_if_index)])

         route.add_vpp_config()
 
         #
         # now expect packets forwarded.
         #
         route.add_vpp_config()
 
         #
         # now expect packets forwarded.
         #

-        self.send_and_expect(self.pg0, p * 65, self.pg1)
+        self.send_and_expect(self.pg0, p * NUM_PKTS, self.pg1)

 
 
 if __name__ == '__main__':
 
 
 if __name__ == '__main__':