tests: initial asf framework refactoring for 'make test'

Type: refactor

Change-Id: I41455b759a5d302ad5c4247c13634c471e7d49a8
Signed-off-by: Pratikshya Prasai <pratikshyaprasai2112@gmail.com>
Signed-off-by: Saima Yunus <yunus.saima.234@gmail.com>
Signed-off-by: Dave Wallace <dwallacelf@gmail.com>

1 files changed, 196 insertions, 0 deletions

diff --git a/test/asf/test_ipfix_export.py b/test/asf/test_ipfix_export.py
new file mode 100644
index 00000000000..be4239edbec
--- /dev/null
+++ b/test/asf/test_ipfix_export.py
@@ -0,0 +1,196 @@
+#!/usr/bin/env python3
+from __future__ import print_function
+import binascii
+import random
+import socket
+import unittest
+import time
+import re
+
+from asfframework import VppTestCase
+from vpp_object import VppObject
+from vpp_pg_interface import CaptureTimeoutError
+from vpp_ip_route import VppIpRoute, VppRoutePath
+from ipaddress import ip_address, IPv4Address, IPv6Address
+from socket import AF_INET, AF_INET6
+
+
+class TestIpfixExporter(VppTestCase):
+    """Ipfix Exporter Tests"""
+
+    def setUp(self):
+        super(TestIpfixExporter, self).setUp()
+        self.create_pg_interfaces(range(4))
+        for i in self.pg_interfaces:
+            i.admin_up()
+            i.config_ip4()
+            i.resolve_arp()
+            i.config_ip6()
+            i.resolve_ndp()
+            i.disable_ipv6_ra()
+
+    def tearDown(self):
+        super(TestIpfixExporter, self).tearDown()
+        for i in self.pg_interfaces:
+            i.unconfig_ip4()
+            i.unconfig_ip6()
+            i.admin_down()
+
+    def find_exp_by_collector_addr(self, exporters, addr):
+        """Find the exporter in the list of exportes with the given  addr"""
+
+        for exp in exporters:
+            if exp.collector_address == IPv4Address(addr):
+                return exp
+        return None
+
+    def verify_exporter_detail(
+        self, exp, collector_addr, src_addr, collector_port=4739, mtu=1400, interval=20
+    ):
+        self.assertTrue(exp is not None)
+        self.assert_equal(exp.collector_address, collector_addr)
+        self.assert_equal(exp.src_address, src_addr)
+        self.assert_equal(exp.collector_port, collector_port)
+        self.assert_equal(exp.path_mtu, mtu)
+        self.assert_equal(exp.template_interval, interval)
+
+    def test_create_multipe_exporters(self):
+        """test that we can create and dump multiple exporters"""
+
+        mtu = 1400
+        interval = 20
+        port = 4739
+
+        # Old API - always gives us pool index 0.
+        self.vapi.set_ipfix_exporter(
+            collector_address=self.pg1.remote_ip4,
+            src_address=self.pg0.local_ip4,
+            collector_port=4739,
+            path_mtu=mtu,
+            template_interval=interval,
+        )
+
+        exporters = self.vapi.ipfix_exporter_dump()
+        exp = self.find_exp_by_collector_addr(exporters, self.pg1.remote_ip4)
+        self.verify_exporter_detail(
+            exp, IPv4Address(self.pg1.remote_ip4), IPv4Address(self.pg0.local_ip4)
+        )
+
+        exporters = list(self.vapi.vpp.details_iter(self.vapi.ipfix_all_exporter_get))
+        exp = self.find_exp_by_collector_addr(exporters, self.pg1.remote_ip4)
+        self.verify_exporter_detail(
+            exp, IPv4Address(self.pg1.remote_ip4), IPv4Address(self.pg0.local_ip4)
+        )
+
+        # create a 2nd exporter
+        self.vapi.ipfix_exporter_create_delete(
+            collector_address=self.pg2.remote_ip4,
+            src_address=self.pg0.local_ip4,
+            collector_port=4739,
+            path_mtu=mtu,
+            template_interval=interval,
+            is_create=True,
+        )
+
+        exporters = list(self.vapi.vpp.details_iter(self.vapi.ipfix_all_exporter_get))
+        self.assertTrue(len(exporters) == 2)
+        exp = self.find_exp_by_collector_addr(exporters, self.pg1.remote_ip4)
+        self.verify_exporter_detail(
+            exp, IPv4Address(self.pg1.remote_ip4), IPv4Address(self.pg0.local_ip4)
+        )
+        exp = self.find_exp_by_collector_addr(exporters, self.pg2.remote_ip4)
+        self.verify_exporter_detail(
+            exp, IPv4Address(self.pg2.remote_ip4), IPv4Address(self.pg0.local_ip4)
+        )
+
+        # Create a 3rd exporter
+        self.vapi.ipfix_exporter_create_delete(
+            collector_address=self.pg3.remote_ip4,
+            src_address=self.pg0.local_ip4,
+            collector_port=4739,
+            path_mtu=mtu,
+            template_interval=interval,
+            is_create=True,
+        )
+
+        exporters = list(self.vapi.vpp.details_iter(self.vapi.ipfix_all_exporter_get))
+        self.assertTrue(len(exporters) == 3)
+        exp = self.find_exp_by_collector_addr(exporters, self.pg1.remote_ip4)
+        self.verify_exporter_detail(
+            exp, IPv4Address(self.pg1.remote_ip4), IPv4Address(self.pg0.local_ip4)
+        )
+        exp = self.find_exp_by_collector_addr(exporters, self.pg2.remote_ip4)
+        self.verify_exporter_detail(
+            exp, IPv4Address(self.pg2.remote_ip4), IPv4Address(self.pg0.local_ip4)
+        )
+        exp = self.find_exp_by_collector_addr(exporters, self.pg3.remote_ip4)
+        self.verify_exporter_detail(
+            exp, IPv4Address(self.pg3.remote_ip4), IPv4Address(self.pg0.local_ip4)
+        )
+
+        # Modify the 2nd exporter.
+        self.vapi.ipfix_exporter_create_delete(
+            collector_address=self.pg2.remote_ip4,
+            src_address=self.pg0.local_ip4,
+            collector_port=4739,
+            path_mtu=mtu + 1,
+            template_interval=interval + 1,
+            is_create=True,
+        )
+
+        exporters = list(self.vapi.vpp.details_iter(self.vapi.ipfix_all_exporter_get))
+        self.assertTrue(len(exporters) == 3)
+        exp = self.find_exp_by_collector_addr(exporters, self.pg1.remote_ip4)
+        self.verify_exporter_detail(
+            exp, IPv4Address(self.pg1.remote_ip4), IPv4Address(self.pg0.local_ip4)
+        )
+        exp = self.find_exp_by_collector_addr(exporters, self.pg2.remote_ip4)
+        self.verify_exporter_detail(
+            exp,
+            IPv4Address(self.pg2.remote_ip4),
+            IPv4Address(self.pg0.local_ip4),
+            mtu=mtu + 1,
+            interval=interval + 1,
+        )
+        exp = self.find_exp_by_collector_addr(exporters, self.pg3.remote_ip4)
+        self.verify_exporter_detail(
+            exp, IPv4Address(self.pg3.remote_ip4), IPv4Address(self.pg0.local_ip4)
+        )
+
+        # Delete 2nd exporter
+        self.vapi.ipfix_exporter_create_delete(
+            collector_address=self.pg2.remote_ip4,
+            src_address=self.pg0.local_ip4,
+            collector_port=4739,
+            path_mtu=mtu,
+            template_interval=interval,
+            is_create=False,
+        )
+
+        exporters = list(self.vapi.vpp.details_iter(self.vapi.ipfix_all_exporter_get))
+        self.assertTrue(len(exporters) == 2)
+        exp = self.find_exp_by_collector_addr(exporters, self.pg1.remote_ip4)
+        self.verify_exporter_detail(
+            exp, IPv4Address(self.pg1.remote_ip4), IPv4Address(self.pg0.local_ip4)
+        )
+        exp = self.find_exp_by_collector_addr(exporters, self.pg3.remote_ip4)
+        self.verify_exporter_detail(
+            exp, IPv4Address(self.pg3.remote_ip4), IPv4Address(self.pg0.local_ip4)
+        )
+
+        # Delete final exporter (exporter in slot 0 can not be deleted)
+        self.vapi.ipfix_exporter_create_delete(
+            collector_address=self.pg3.remote_ip4,
+            src_address=self.pg0.local_ip4,
+            collector_port=4739,
+            path_mtu=mtu,
+            template_interval=interval,
+            is_create=False,
+        )
+
+        exporters = list(self.vapi.vpp.details_iter(self.vapi.ipfix_all_exporter_get))
+        self.assertTrue(len(exporters) == 1)
+        exp = self.find_exp_by_collector_addr(exporters, self.pg1.remote_ip4)
+        self.verify_exporter_detail(
+            exp, IPv4Address(self.pg1.remote_ip4), IPv4Address(self.pg0.local_ip4)
+        )