test_primitive_non-string_parameter.py

   1 # A simple test to exercise the libraries' functionality
   2 import asyncio
   3 import functools
   4 import os
   5 import sys
   6 import logging
   7 import unittest
   8 import yaml
   9 from n2vc.vnf import N2VC
  10
  11 NSD_YAML = """
  12 nsd:nsd-catalog:
  13     nsd:
  14     -   id: multicharmvdu-ns
  15         name: multicharmvdu-ns
  16         short-name: multicharmvdu-ns
  17         description: NS with 1 VNF
  18         version: '1.0'
  19         logo: osm.png
  20         constituent-vnfd:
  21         -   vnfd-id-ref: multicharmvdu-vnf
  22             member-vnf-index: '1'
  23         vld:
  24         -   id: datanet
  25             name: datanet
  26             short-name: datanet
  27             type: ELAN
  28             vnfd-connection-point-ref:
  29             -   vnfd-id-ref: multicharmvdu-vnf
  30                 member-vnf-index-ref: '1'
  31                 vnfd-connection-point-ref: vnf-data
  32 """
  33
  34 VNFD_YAML = """
  35 vnfd:vnfd-catalog:
  36     vnfd:
  37     -   id: multicharmvdu-vnf
  38         name: multicharmvdu-vnf
  39         short-name: multicharmvdu-vnf
  40         version: '1.0'
  41         description: A VNF consisting of 1 VDUs w/charm
  42         logo: osm.png
  43         connection-point:
  44         -   id: vnf-data
  45             name: vnf-data
  46             short-name: vnf-data
  47             type: VPORT
  48         mgmt-interface:
  49             cp: vnf-data
  50         internal-vld:
  51         -   id: internal
  52             name: internal
  53             short-name: internal
  54             type: ELAN
  55             internal-connection-point:
  56             -   id-ref: dataVM-internal
  57         vdu:
  58         -   id: dataVM
  59             name: dataVM
  60             image: xenial
  61             count: '1'
  62             vm-flavor:
  63                 vcpu-count: '1'
  64                 memory-mb: '1024'
  65                 storage-gb: '10'
  66             interface:
  67             -   name: dataVM-eth0
  68                 position: '1'
  69                 type: INTERNAL
  70                 virtual-interface:
  71                     type: VIRTIO
  72                 internal-connection-point-ref: dataVM-internal
  73             -   name: dataVM-xe0
  74                 position: '2'
  75                 type: EXTERNAL
  76                 virtual-interface:
  77                     type: VIRTIO
  78                 external-connection-point-ref: vnf-data
  79             internal-connection-point:
  80             -   id: dataVM-internal
  81                 name: dataVM-internal
  82                 short-name: dataVM-internal
  83                 type: VPORT
  84             vdu-configuration:
  85                 juju:
  86                     charm: simple
  87                 initial-config-primitive:
  88                 -   seq: '1'
  89                     name: config
  90                     parameter:
  91                     -   name: ssh-hostname
  92                         value: <rw_mgmt_ip>
  93                     -   name: ssh-username
  94                         value: ubuntu
  95                     -   name: ssh-password
  96                         value: ubuntu
  97                 -   seq: '2'
  98                     name: touch
  99                     parameter:
 100                     -   name: filename
 101                         value: '/home/ubuntu/first-touch-dataVM'
 102                 -   seq: '3'
 103                     name: testint
 104                     parameter:
 105                     -   name: interval
 106                         value: 20
 107                 config-primitive:
 108                 -   name: touch
 109                     parameter:
 110                     -   name: filename
 111                         data-type: STRING
 112                         default-value: '/home/ubuntu/touched'
 113 """
 114
 115 class PythonTest(unittest.TestCase):
 116     n2vc = None
 117
 118     def setUp(self):
 119
 120         self.log = logging.getLogger()
 121         self.log.level = logging.DEBUG
 122
 123         self.loop = asyncio.get_event_loop()
 124
 125         # self.loop = asyncio.new_event_loop()
 126         # asyncio.set_event_loop(None)
 127
 128         # Extract parameters from the environment in order to run our test
 129         vca_host = os.getenv('VCA_HOST', '127.0.0.1')
 130         vca_port = os.getenv('VCA_PORT', 17070)
 131         vca_user = os.getenv('VCA_USER', 'admin')
 132         vca_charms = os.getenv('VCA_CHARMS', None)
 133         vca_secret = os.getenv('VCA_SECRET', None)
 134         self.n2vc = N2VC(
 135             log=self.log,
 136             server=vca_host,
 137             port=vca_port,
 138             user=vca_user,
 139             secret=vca_secret,
 140             artifacts=vca_charms,
 141         )
 142
 143     def tearDown(self):
 144         self.loop.run_until_complete(self.n2vc.logout())
 145
 146     def get_descriptor(self, descriptor):
 147         desc = None
 148         try:
 149             tmp = yaml.load(descriptor)
 150
 151             # Remove the envelope
 152             root = list(tmp.keys())[0]
 153             if root == "nsd:nsd-catalog":
 154                 desc = tmp['nsd:nsd-catalog']['nsd'][0]
 155             elif root == "vnfd:vnfd-catalog":
 156                 desc = tmp['vnfd:vnfd-catalog']['vnfd'][0]
 157         except ValueError:
 158             assert False
 159         return desc
 160
 161     def n2vc_callback(self, model_name, application_name, workload_status, task=None):
 162         """We pass the vnfd when setting up the callback, so expect it to be
 163         returned as a tuple."""
 164         if workload_status and not task:
 165             self.log.debug("Callback: workload status \"{}\"".format(workload_status))
 166
 167             if workload_status in ["blocked"]:
 168                 task = asyncio.ensure_future(
 169                     self.n2vc.ExecutePrimitive(
 170                         model_name,
 171                         application_name,
 172                         "config",
 173                         None,
 174                         params={
 175                             'ssh-hostname': '10.195.8.78',
 176                             'ssh-username': 'ubuntu',
 177                             'ssh-password': 'ubuntu'
 178                         }
 179                     )
 180                 )
 181                 task.add_done_callback(functools.partial(self.n2vc_callback, None, None, None))
 182                 pass
 183             elif workload_status in ["active"]:
 184                 self.log.debug("Removing charm")
 185                 task = asyncio.ensure_future(
 186                     self.n2vc.RemoveCharms(model_name, application_name, self.n2vc_callback)
 187                 )
 188                 task.add_done_callback(functools.partial(self.n2vc_callback, None, None, None))
 189
 190     def test_deploy_application(self):
 191         stream_handler = logging.StreamHandler(sys.stdout)
 192         self.log.addHandler(stream_handler)
 193         try:
 194             self.log.info("Log handler installed")
 195             nsd = self.get_descriptor(NSD_YAML)
 196             vnfd = self.get_descriptor(VNFD_YAML)
 197
 198             if nsd and vnfd:
 199
 200                 vca_charms = os.getenv('VCA_CHARMS', None)
 201
 202                 params = {}
 203                 vnf_index = 0
 204
 205                 def deploy():
 206                     """An inner function to do the deployment of a charm from
 207                     either a vdu or vnf.
 208                     """
 209                     charm_dir = "{}/{}".format(vca_charms, charm)
 210
 211                     # Setting this to an IP that will fail the initial config.
 212                     # This will be detected in the callback, which will execute
 213                     # the "config" primitive with the right IP address.
 214                     params['rw_mgmt_ip'] = '10.195.8.78'
 215
 216                     # self.loop.run_until_complete(n.CreateNetworkService(nsd))
 217                     ns_name = "default"
 218
 219                     vnf_name = self.n2vc.FormatApplicationName(
 220                         ns_name,
 221                         vnfd['name'],
 222                         str(vnf_index),
 223                     )
 224
 225                     self.loop.run_until_complete(
 226                         self.n2vc.DeployCharms(
 227                             ns_name,
 228                             vnf_name,
 229                             vnfd,
 230                             charm_dir,
 231                             params,
 232                             {},
 233                             self.n2vc_callback
 234                         )
 235                     )
 236
 237                 # Check if the VDUs in this VNF have a charm
 238                 for vdu in vnfd['vdu']:
 239                     vdu_config = vdu.get('vdu-configuration')
 240                     if vdu_config:
 241                         juju = vdu_config['juju']
 242                         self.assertIsNotNone(juju)
 243
 244                         charm = juju['charm']
 245                         self.assertIsNotNone(charm)
 246
 247                         params['initial-config-primitive'] = vdu_config['initial-config-primitive']
 248
 249                         deploy()
 250                         vnf_index += 1
 251
 252                 # Check if this VNF has a charm
 253                 vnf_config = vnfd.get("vnf-configuration")
 254                 if vnf_config:
 255                     juju = vnf_config['juju']
 256                     self.assertIsNotNone(juju)
 257
 258                     charm = juju['charm']
 259                     self.assertIsNotNone(charm)
 260
 261                     params['initial-config-primitive'] = vnf_config['initial-config-primitive']
 262
 263                     deploy()
 264                     vnf_index += 1
 265
 266                 self.loop.run_forever()
 267
 268                 # self.loop.run_until_complete(n.GetMetrics(vnfd, nsd=nsd))
 269
 270                 # Test actions
 271                 #  ExecutePrimitive(self, nsd, vnfd, vnf_member_index, primitive, callback, *callback_args, **params):
 272
 273                 # self.loop.run_until_complete(n.DestroyNetworkService(nsd))
 274
 275                 # self.loop.run_until_complete(self.n2vc.logout())
 276         finally:
 277             self.log.removeHandler(stream_handler)