1 # A simple test to exercise the libraries' functionality
9 from n2vc
.vnf
import N2VC
14 - id: multicharmvdu-ns
15 name: multicharmvdu-ns
16 short-name: multicharmvdu-ns
17 description: NS with 2 VNFs multicharmvdu-vnf connected by datanet and mgmtnet VLs
21 - vnfd-id-ref: multicharmvdu-vnf
23 - vnfd-id-ref: multicharmvdu-vnf
31 vim-network-name: mgmt
32 vnfd-connection-point-ref:
33 - vnfd-id-ref: multicharmvdu-vnf
34 member-vnf-index-ref: '1'
35 vnfd-connection-point-ref: vnf-mgmt
36 - vnfd-id-ref: multicharmvdu-vnf
37 member-vnf-index-ref: '2'
38 vnfd-connection-point-ref: vnf-mgmt
43 vnfd-connection-point-ref:
44 - vnfd-id-ref: multicharmvdu-vnf
45 member-vnf-index-ref: '1'
46 vnfd-connection-point-ref: vnf-data
47 - vnfd-id-ref: multicharmvdu-vnf
48 member-vnf-index-ref: '2'
49 vnfd-connection-point-ref: vnf-data
55 - id: multicharmvdu-vnf
56 name: multicharmvdu-vnf
57 short-name: multicharmvdu-vnf
59 description: A VNF consisting of 2 VDUs w/charms connected to an internal VL, and one VDU with cloud-init
77 internal-connection-point:
78 - id-ref: mgmtVM-internal
79 - id-ref: dataVM-internal
95 external-connection-point-ref: vnf-mgmt
101 internal-connection-point-ref: mgmtVM-internal
102 internal-connection-point:
103 - id: mgmtVM-internal
104 name: mgmtVM-internal
105 short-name: mgmtVM-internal
107 cloud-init-file: cloud-config.txt
111 initial-config-primitive:
125 value: '/home/ubuntu/first-touch-mgmtVM'
131 default-value: '/home/ubuntu/touched'
147 internal-connection-point-ref: dataVM-internal
153 external-connection-point-ref: vnf-data
154 internal-connection-point:
155 - id: dataVM-internal
156 name: dataVM-internal
157 short-name: dataVM-internal
162 initial-config-primitive:
176 value: '/home/ubuntu/first-touch-dataVM'
182 default-value: '/home/ubuntu/touched'
185 class PythonTest(unittest
.TestCase
):
190 self
.log
= logging
.getLogger()
191 self
.log
.level
= logging
.DEBUG
193 self
.loop
= asyncio
.get_event_loop()
195 # self.loop = asyncio.new_event_loop()
196 # asyncio.set_event_loop(None)
198 # Extract parameters from the environment in order to run our test
199 vca_host
= os
.getenv('VCA_HOST', '127.0.0.1')
200 vca_port
= os
.getenv('VCA_PORT', 17070)
201 vca_user
= os
.getenv('VCA_USER', 'admin')
202 vca_charms
= os
.getenv('VCA_CHARMS', None)
203 vca_secret
= os
.getenv('VCA_SECRET', None)
210 artifacts
=vca_charms
,
214 self
.loop
.run_until_complete(self
.n2vc
.logout())
216 def get_descriptor(self
, descriptor
):
219 tmp
= yaml
.load(descriptor
)
221 # Remove the envelope
222 root
= list(tmp
.keys())[0]
223 if root
== "nsd:nsd-catalog":
224 desc
= tmp
['nsd:nsd-catalog']['nsd'][0]
225 elif root
== "vnfd:vnfd-catalog":
226 desc
= tmp
['vnfd:vnfd-catalog']['vnfd'][0]
231 def n2vc_callback(self
, model_name
, application_name
, workload_status
, task
=None):
232 """We pass the vnfd when setting up the callback, so expect it to be
233 returned as a tuple."""
234 if workload_status
and not task
:
235 self
.log
.debug("Callback: workload status \"{}\"".format(workload_status
))
237 if workload_status
in ["blocked"]:
238 task
= asyncio
.ensure_future(
239 self
.n2vc
.ExecutePrimitive(
245 'ssh-hostname': '10.195.8.78',
246 'ssh-username': 'ubuntu',
247 'ssh-password': 'ubuntu'
251 task
.add_done_callback(functools
.partial(self
.n2vc_callback
, None, None, None))
253 elif workload_status
in ["active"]:
254 self
.log
.debug("Removing charm")
255 task
= asyncio
.ensure_future(
256 self
.n2vc
.RemoveCharms(model_name
, application_name
, self
.n2vc_callback
)
258 task
.add_done_callback(functools
.partial(self
.n2vc_callback
, None, None, None))
260 def test_deploy_application(self
):
261 stream_handler
= logging
.StreamHandler(sys
.stdout
)
262 self
.log
.addHandler(stream_handler
)
264 self
.log
.info("Log handler installed")
265 nsd
= self
.get_descriptor(NSD_YAML
)
266 vnfd
= self
.get_descriptor(VNFD_YAML
)
270 vca_charms
= os
.getenv('VCA_CHARMS', None)
276 """An inner function to do the deployment of a charm from
279 charm_dir
= "{}/{}".format(vca_charms
, charm
)
281 # Setting this to an IP that will fail the initial config.
282 # This will be detected in the callback, which will execute
283 # the "config" primitive with the right IP address.
284 params
['rw_mgmt_ip'] = '10.195.8.78'
286 # self.loop.run_until_complete(n.CreateNetworkService(nsd))
289 vnf_name
= self
.n2vc
.FormatApplicationName(
295 self
.loop
.run_until_complete(
296 self
.n2vc
.DeployCharms(
307 # Check if the VDUs in this VNF have a charm
308 for vdu
in vnfd
['vdu']:
309 vdu_config
= vdu
.get('vdu-configuration')
311 juju
= vdu_config
['juju']
312 self
.assertIsNotNone(juju
)
314 charm
= juju
['charm']
315 self
.assertIsNotNone(charm
)
317 params
['initial-config-primitive'] = vdu_config
['initial-config-primitive']
322 # Check if this VNF has a charm
323 vnf_config
= vnfd
.get("vnf-configuration")
325 juju
= vnf_config
['juju']
326 self
.assertIsNotNone(juju
)
328 charm
= juju
['charm']
329 self
.assertIsNotNone(charm
)
331 params
['initial-config-primitive'] = vnf_config
['initial-config-primitive']
336 self
.loop
.run_forever()
338 # self.loop.run_until_complete(n.GetMetrics(vnfd, nsd=nsd))
341 # ExecutePrimitive(self, nsd, vnfd, vnf_member_index, primitive, callback, *callback_args, **params):
343 # self.loop.run_until_complete(n.DestroyNetworkService(nsd))
345 # self.loop.run_until_complete(self.n2vc.logout())
347 self
.log
.removeHandler(stream_handler
)