X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=tests%2Ftest_primitive_non-string_parameter.py;fp=tests%2Ftest_primitive_non-string_parameter.py;h=0000000000000000000000000000000000000000;hp=9a2d5ad78214a4bb320c9f543d738ce419806309;hb=5e08a0e8fa4fd9d0156d28f8f4e53e5b176c704a;hpb=427f57893cb4bf8aebc0b6f56d0cb3f178d23cf1 diff --git a/tests/test_primitive_non-string_parameter.py b/tests/test_primitive_non-string_parameter.py deleted file mode 100644 index 9a2d5ad..0000000 --- a/tests/test_primitive_non-string_parameter.py +++ /dev/null @@ -1,282 +0,0 @@ -# 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: - - 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)