Implementing Gluon L3VPN configuration in testcase

Extending the Gluon vping test case by implementing the basic
L3VPN Proton configuration:
 - creation of 2 ports
 - creation of 1 VPN
 - creation of 2 vpnbindings to bind the ports to the VPN

Change-Id: I8e1c0c5ec693869b70552726cc20234c2670a9f2
Signed-off-by: Georg Kunz <georg.kunz@ericsson.com>

1 files changed, 153 insertions, 5 deletions

diff --git a/test/functest/gluon-test-suite.py b/test/functest/gluon-test-suite.py
index 2a64843..b26d806 100755
--- a/test/functest/gluon-test-suite.py
+++ b/test/functest/gluon-test-suite.py
@@ -1,14 +1,162 @@
 #!/usr/bin/python
 
+from opnfv.deployment import factory
+import functest.utils.openstack_utils as osutils
+import functest.utils.functest_logger as ft_logger
+
+import json
+import logging
+import os
 import sys
+import time
+
+logger = ft_logger.Logger("netready-gluon-vping").getLogger()
+logger.setLevel(logging.DEBUG)
+
+
+class GluonVPing:
+
+    def __init__(self):
+        self.controller_node = None
+        self.port_ids = []
+        self.vpn_id = None
+        self.vpnbinding_ids = []
+
+    def prepare_test(self):
+        logger.info("Preparing Gluon vPing testcase.")
+
+        undercloud_ip = os.getenv("INSTALLER_IP")
+        handler = factory.Factory.get_handler('apex',
+                                              undercloud_ip,
+                                              'stack',
+                                              pkey_file='/root/.ssh/id_rsa')
+        nodes = handler.get_nodes()
+        self.controller_node = None
+        for node in nodes:
+            if node.is_controller():
+                self.controller_node = node
+                break
+
+        if self.controller_node is None:
+            logger.error("ERROR: no controller node found.")
+            raise Exception('No controller node found')
+
+    def run(self):
+        self._create_ports()
+        self._create_vpn()
+        self._create_vpnbinding()
+        #
+        # TODO boot VMs and ping
+        #
+
+    def _run_protonclient(self, options):
+        cmd = "protonclient --api net-l3vpn " + options
+        if self.controller_node is not None:
+            logger.debug("Running command: %s" % cmd)
+            output = self.controller_node.run_cmd(cmd)
+            if output is None:
+                raise Exception('Remote command failed')
+            else:
+                return output
+
+    def _create_ports(self):
+        def _build_create_port_options(name, mac, tenant_id):
+            return 'port-create' \
+                ' --mac_address ' + mac + \
+                ' --mtu 1500' \
+                ' --admin_state_up True' \
+                ' --name ' + name + \
+                ' --vlan_transparency True' \
+                ' --vnic_type normal' \
+                ' --vif_type ovs' \
+                ' --status ACTIVE' \
+                ' --tenant_id ' + tenant_id
+
+        tenant_id = osutils.get_tenant_id(
+                                osutils.get_keystone_client(),
+                                'admin')
+        for i in [1, 2]:
+            options = _build_create_port_options(
+                            'gluonPort' + str(i),
+                            'c0:2a:14:04:43:0' + str(i),
+                            tenant_id)
+            output = self._run_protonclient(options)
+            port_data = json.loads(output)
+
+            logger.info('Port %s created.' % port_data['id'])
+            self.port_ids.append(port_data['id'])
+            time.sleep(1)
+
+    def _create_vpn(self):
+        vpn_options = 'vpn-create '\
+                        '--name "GluonVPN" ' \
+                        '--ipv4_family 1000:1000 ' \
+                        '--ipv6_family 1000:1000 ' \
+                        '--route_distinguishers 1000:1000'
+        output = self._run_protonclient(vpn_options)
+        vpn_data = json.loads(output)
+
+        logger.info('VPN %s created.' % vpn_data['id'])
+        self.vpn_id = vpn_data['id']
+        time.sleep(1)
+
+    def _create_vpnbinding(self):
+        def _build_create_vpnbinding(port_id, counter):
+            return 'vpnbinding-create' \
+                    ' --interface_id ' + port_id + \
+                    ' --service_id ' + self.vpn_id + \
+                    ' --ipaddress 10.10.' + str(counter) + '.2' \
+                    ' --subnet_prefix 24' \
+                    ' --gateway 10.10.' + str(counter) + '.1'
+
+        counter = 0
+        for port_id in self.port_ids:
+            options = _build_create_vpnbinding(port_id, counter)
+            output = self._run_protonclient(options)
+            vpnbinding_data = json.loads(output)
+
+            logger.info('VPN-binding %s created.'
+                        % vpnbinding_data['interface_id'])
+            self.vpnbinding_ids.append(vpnbinding_data['interface_id'])
+            counter += 1
+            time.sleep(1)
+
+    def _cleanup_ports(self):
+        for port_id in self.port_ids:
+            logger.info("Deleting port %s" % port_id)
+            self._run_protonclient('port-delete ' + str(port_id))
+
+    def _cleanup_vpn(self):
+        logger.info("Deleting VPN %s" % self.vpn_id)
+        self._run_protonclient('vpn-delete ' + str(self.vpn_id))
+
+    def _cleanup_vpnbindings(self):
+        for vpnbinding_id in self.vpnbinding_ids:
+            logger.info("Deleting VPN-binding %s" % vpnbinding_id)
+            self._run_protonclient('vpnbinding-delete ' + str(vpnbinding_id))
+
+    def cleanup(self):
+        logger.info("Cleaning up Gluon vPing testcase...")
+        self._cleanup_vpnbindings()
+        self._cleanup_vpn()
+        self._cleanup_ports()
+
 
 def main():
-    print "Running Gluon test suite..."
-    #
-    # TODO: awesome test
-    #
-    print "Gluon test suite successfully completed."
+    logger.info("Running Gluon test suite...")
+
+    try:
+        gluon_vping = GluonVPing()
+        gluon_vping.prepare_test()
+        gluon_vping.run()
+        gluon_vping.cleanup()
+    except Exception as e:
+        logger.info("Gluon test suite failed: %s" % e)
+        return 1
+
+    logger.info("Gluon test suite successfully completed.")
     return 0
 
+
 if __name__ == '__main__':
     sys.exit(main())