tests/test_primitive_non-string_parameter.py

# A simple test to exercise the libraries' functionality
import asyncio
import functools
import os
import sys
import logging
import unittest
import yaml
from n2vc.vnf import N2VC

NSD_YAML = """
nsd:nsd-catalog:
    nsd:
    -   id: multicharmvdu-ns
        name: multicharmvdu-ns
        short-name: multicharmvdu-ns
        description: NS with 1 VNF
        version: '1.0'
        logo: osm.png
        constituent-vnfd:
        -   vnfd-id-ref: multicharmvdu-vnf
            member-vnf-index: '1'
        vld:
        -   id: datanet
            name: datanet
            short-name: datanet
            type: ELAN
            vnfd-connection-point-ref:
            -   vnfd-id-ref: multicharmvdu-vnf
                member-vnf-index-ref: '1'
                vnfd-connection-point-ref: vnf-data
"""

VNFD_YAML = """
vnfd:vnfd-catalog:
    vnfd:
    -   id: multicharmvdu-vnf
        name: multicharmvdu-vnf
        short-name: multicharmvdu-vnf
        version: '1.0'
        description: A VNF consisting of 1 VDUs w/charm
        logo: osm.png
        connection-point:
        -   id: vnf-data
            name: vnf-data
            short-name: vnf-data
            type: VPORT
        mgmt-interface:
            cp: vnf-data
        internal-vld:
        -   id: internal
            name: internal
            short-name: internal
            type: ELAN
            internal-connection-point:
            -   id-ref: dataVM-internal
        vdu:
        -   id: dataVM
            name: dataVM
            image: xenial
            count: '1'
            vm-flavor:
                vcpu-count: '1'
                memory-mb: '1024'
                storage-gb: '10'
            interface:
            -   name: dataVM-eth0
                position: '1'
                type: INTERNAL
                virtual-interface:
                    type: VIRTIO
                internal-connection-point-ref: dataVM-internal
            -   name: dataVM-xe0
                position: '2'
                type: EXTERNAL
                virtual-interface:
                    type: VIRTIO
                external-connection-point-ref: vnf-data
            internal-connection-point:
            -   id: dataVM-internal
                name: dataVM-internal
                short-name: dataVM-internal
                type: VPORT
            vdu-configuration:
                juju:
                    charm: simple
                initial-config-primitive:
                -   seq: '1'
                    name: config
                    parameter:
                    -   name: ssh-hostname
                        value: <rw_mgmt_ip>
                    -   name: ssh-username
                        value: ubuntu
                    -   name: ssh-password
                        value: ubuntu
                -   seq: '2'
                    name: touch
                    parameter:
                    -   name: filename
                        value: '/home/ubuntu/first-touch-dataVM'
                -   seq: '3'
                    name: testint
                    parameter:
                    -   name: interval
                        value: 20
                config-primitive:
                -   name: touch
                    parameter:
                    -   name: filename
                        data-type: STRING
                        default-value: '/home/ubuntu/touched'
"""

class PythonTest(unittest.TestCase):
    n2vc = None

    def setUp(self):

        self.log = logging.getLogger()
        self.log.level = logging.DEBUG

        self.loop = asyncio.get_event_loop()

        # self.loop = asyncio.new_event_loop()
        # asyncio.set_event_loop(None)

        # Extract parameters from the environment in order to run our test
        vca_host = os.getenv('VCA_HOST', '127.0.0.1')
        vca_port = os.getenv('VCA_PORT', 17070)
        vca_user = os.getenv('VCA_USER', 'admin')
        vca_charms = os.getenv('VCA_CHARMS', None)
        vca_secret = os.getenv('VCA_SECRET', None)
        self.n2vc = N2VC(
            log=self.log,
            server=vca_host,
            port=vca_port,
            user=vca_user,
            secret=vca_secret,
            artifacts=vca_charms,
        )

    def tearDown(self):
        self.loop.run_until_complete(self.n2vc.logout())

    def get_descriptor(self, descriptor):
        desc = None
        try:
            tmp = yaml.load(descriptor)

            # Remove the envelope
            root = list(tmp.keys())[0]
            if root == "nsd:nsd-catalog":
                desc = tmp['nsd:nsd-catalog']['nsd'][0]
            elif root == "vnfd:vnfd-catalog":
                desc = tmp['vnfd:vnfd-catalog']['vnfd'][0]
        except ValueError:
            assert False
        return desc

    def n2vc_callback(self, model_name, application_name, workload_status, task=None):
        """We pass the vnfd when setting up the callback, so expect it to be
        returned as a tuple."""
        if workload_status and not task:
            self.log.debug("Callback: workload status \"{}\"".format(workload_status))

            if workload_status in ["blocked"]:
                task = asyncio.ensure_future(
                    self.n2vc.ExecutePrimitive(
                        model_name,
                        application_name,
                        "config",
                        None,
                        params={
                            'ssh-hostname': '10.195.8.78',
                            'ssh-username': 'ubuntu',
                            'ssh-password': 'ubuntu'
                        }
                    )
                )
                task.add_done_callback(functools.partial(self.n2vc_callback, None, None, None))
                pass
            elif workload_status in ["active"]:
                self.log.debug("Removing charm")
                task = asyncio.ensure_future(
                    self.n2vc.RemoveCharms(model_name, application_name, self.n2vc_callback)
                )
                task.add_done_callback(functools.partial(self.n2vc_callback, None, None, None))

    def test_deploy_application(self):
        stream_handler = logging.StreamHandler(sys.stdout)
        self.log.addHandler(stream_handler)
        try:
            self.log.info("Log handler installed")
            nsd = self.get_descriptor(NSD_YAML)
            vnfd = self.get_descriptor(VNFD_YAML)

            if nsd and vnfd:

                vca_charms = os.getenv('VCA_CHARMS', None)

                params = {}
                vnf_index = 0

                def deploy():
                    """An inner function to do the deployment of a charm from
                    either a vdu or vnf.
                    """
                    charm_dir = "{}/{}".format(vca_charms, charm)

                    # Setting this to an IP that will fail the initial config.
                    # This will be detected in the callback, which will execute
                    # the "config" primitive with the right IP address.
                    params['rw_mgmt_ip'] = '10.195.8.78'

                    # self.loop.run_until_complete(n.CreateNetworkService(nsd))
                    ns_name = "default"

                    vnf_name = self.n2vc.FormatApplicationName(
                        ns_name,
                        vnfd['name'],
                        str(vnf_index),
                    )

                    self.loop.run_until_complete(
                        self.n2vc.DeployCharms(
                            ns_name,
                            vnf_name,
                            vnfd,
                            charm_dir,
                            params,
                            {},
                            self.n2vc_callback
                        )
                    )

                # Check if the VDUs in this VNF have a charm
                for vdu in vnfd['vdu']:
                    vdu_config = vdu.get('vdu-configuration')
                    if vdu_config:
                        juju = vdu_config['juju']
                        self.assertIsNotNone(juju)

                        charm = juju['charm']
                        self.assertIsNotNone(charm)

                        params['initial-config-primitive'] = vdu_config['initial-config-primitive']

                        deploy()
                        vnf_index += 1

                # Check if this VNF has a charm
                vnf_config = vnfd.get("vnf-configuration")
                if vnf_config:
                    juju = vnf_config['juju']
                    self.assertIsNotNone(juju)

                    charm = juju['charm']
                    self.assertIsNotNone(charm)

                    params['initial-config-primitive'] = vnf_config['initial-config-primitive']

                    deploy()
                    vnf_index += 1

                self.loop.run_forever()

                # self.loop.run_until_complete(n.GetMetrics(vnfd, nsd=nsd))

                # Test actions
                #  ExecutePrimitive(self, nsd, vnfd, vnf_member_index, primitive, callback, *callback_args, **params):

                # self.loop.run_until_complete(n.DestroyNetworkService(nsd))

                # self.loop.run_until_complete(self.n2vc.logout())
        finally:
            self.log.removeHandler(stream_handler)