"""Test the deployment and configuration of a proxy charm. 1. Deploy proxy charm to a unit 2. Execute 'get-ssh-public-key' primitive and get returned value 3. Create LXD container with unit's public ssh key 4. Verify SSH works between unit and container 5. Destroy Juju unit 6. Stop and Destroy LXD container """ import asyncio import functools import os import sys import logging import unittest from . import utils import yaml from n2vc.vnf import N2VC NSD_YAML = """ nsd:nsd-catalog: nsd: - id: singlecharmvdu-ns name: singlecharmvdu-ns short-name: singlecharmvdu-ns description: NS with 1 VNFs singlecharmvdu-vnf connected by datanet and mgmtnet VLs version: '1.0' logo: osm.png constituent-vnfd: - vnfd-id-ref: singlecharmvdu-vnf member-vnf-index: '1' vld: - id: mgmtnet name: mgmtnet short-name: mgmtnet type: ELAN mgmt-network: 'true' vim-network-name: mgmt vnfd-connection-point-ref: - vnfd-id-ref: singlecharmvdu-vnf member-vnf-index-ref: '1' vnfd-connection-point-ref: vnf-mgmt - vnfd-id-ref: singlecharmvdu-vnf member-vnf-index-ref: '2' vnfd-connection-point-ref: vnf-mgmt - id: datanet name: datanet short-name: datanet type: ELAN vnfd-connection-point-ref: - vnfd-id-ref: singlecharmvdu-vnf member-vnf-index-ref: '1' vnfd-connection-point-ref: vnf-data - vnfd-id-ref: singlecharmvdu-vnf member-vnf-index-ref: '2' vnfd-connection-point-ref: vnf-data """ VNFD_YAML = """ vnfd:vnfd-catalog: vnfd: - id: singlecharmvdu-vnf name: singlecharmvdu-vnf short-name: singlecharmvdu-vnf version: '1.0' description: A VNF consisting of 2 VDUs w/charms connected to an internal VL, and one VDU with cloud-init logo: osm.png connection-point: - id: vnf-mgmt name: vnf-mgmt short-name: vnf-mgmt type: VPORT - id: vnf-data name: vnf-data short-name: vnf-data type: VPORT mgmt-interface: cp: vnf-mgmt internal-vld: - id: internal name: internal short-name: internal type: ELAN internal-connection-point: - id-ref: mgmtVM-internal - id-ref: dataVM-internal vdu: - id: mgmtVM name: mgmtVM image: xenial count: '1' vm-flavor: vcpu-count: '1' memory-mb: '1024' storage-gb: '10' interface: - name: mgmtVM-eth0 position: '1' type: EXTERNAL virtual-interface: type: VIRTIO external-connection-point-ref: vnf-mgmt - name: mgmtVM-eth1 position: '2' type: INTERNAL virtual-interface: type: VIRTIO internal-connection-point-ref: mgmtVM-internal internal-connection-point: - id: mgmtVM-internal name: mgmtVM-internal short-name: mgmtVM-internal type: VPORT cloud-init-file: cloud-config.txt vdu-configuration: juju: charm: simple initial-config-primitive: - seq: '1' name: config parameter: - name: ssh-hostname value: - name: ssh-username value: ubuntu - name: ssh-password value: ubuntu - seq: '2' name: touch parameter: - name: filename value: '/home/ubuntu/first-touch-mgmtVM' config-primitive: - name: touch parameter: - name: filename data-type: STRING default-value: '/home/ubuntu/touched' """ class PythonTest(unittest.TestCase): n2vc = None container = None def setUp(self): self.log = logging.getLogger() self.log.level = logging.DEBUG self.loop = asyncio.get_event_loop() # self.container = utils.create_lxd_container() self.n2vc = utils.get_n2vc() def tearDown(self): if self.container: self.container.stop() self.container.delete() self.loop.run_until_complete(self.n2vc.logout()) def n2vc_callback(self, model_name, application_name, workload_status, workload_message, task=None): """We pass the vnfd when setting up the callback, so expect it to be returned as a tuple.""" self.log.debug("[Callback] Workload status '{}' for application {}".format(workload_status, application_name)) self.log.debug("[Callback] Task: \"{}\"".format(task)) if workload_status == "exec_primitive" and task: self.log.debug("Getting Primitive Status") # get the uuid from the task uuid = task.result() # get the status of the action task = asyncio.ensure_future( self.n2vc.GetPrimitiveStatus( model_name, uuid, ) ) task.add_done_callback(functools.partial(self.n2vc_callback, model_name, application_name, "primitive_status", task)) if workload_status == "primitive_status" and task and not self.container: self.log.debug("Creating LXD container") # Get the ssh key result = task.result() pubkey = result['pubkey'] self.container = utils.create_lxd_container(pubkey) mgmtaddr = self.container.state().network['eth0']['addresses'] self.log.debug("Setting config ssh-hostname={}".format(mgmtaddr[0]['address'])) task = asyncio.ensure_future( self.n2vc.ExecutePrimitive( model_name, application_name, "config", None, params={ 'ssh-hostname': mgmtaddr[0]['address'], } ) ) task.add_done_callback(functools.partial(self.n2vc_callback, model_name, application_name, None, None)) if workload_status and not task: self.log.debug("Callback: workload status \"{}\"".format(workload_status)) if workload_status in ["blocked"] and not self.container: self.log.debug("Getting public SSH key") # Execute 'get-ssh-public-key' primitive and get returned value task = asyncio.ensure_future( self.n2vc.ExecutePrimitive( model_name, application_name, "get-ssh-public-key", None, params={ 'ssh-hostname': '10.195.8.78', 'ssh-username': 'ubuntu', 'ssh-password': 'ubuntu' } ) ) task.add_done_callback(functools.partial(self.n2vc_callback, model_name, application_name, "exec_primitive", task)) # 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)) if self.container: utils.destroy_lxd_container(self.container) self.container = None # Stop the test self.loop.call_soon_threadsafe(self.loop.stop) def test_deploy_application(self): """Deploy proxy charm to a unit.""" stream_handler = logging.StreamHandler(sys.stdout) self.log.addHandler(stream_handler) try: self.log.info("Log handler installed") nsd = utils.get_descriptor(NSD_YAML) vnfd = utils.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. # mgmtaddr = self.container.state().network['eth0']['addresses'] # params['rw_mgmt_ip'] = mgmtaddr[0]['address'] # Legacy method is to set the ssh-private-key config # with open(utils.get_juju_private_key(), "r") as f: # pkey = f.readline() # params['ssh-private-key'] = pkey 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() # while self.loop.is_running(): # # await asyncio.sleep(1) # time.sleep(1) # 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)