+++ /dev/null
-# 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
- data-type: STRING
- value: <rw_mgmt_ip>
- - name: ssh-username
- data-type: STRING
- value: ubuntu
- - name: ssh-password
- data-type: STRING
- value: ubuntu
- - seq: '2'
- name: touch
- parameter:
- - name: filename
- data-type: STRING
- value: '/home/ubuntu/first-touch-dataVM'
- - seq: '3'
- name: testint
- parameter:
- - name: interval
- data-type: INTEGER
- 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)
projects / osm / N2VC.git / blobdiff