1 """Test the deployment and configuration of a proxy charm.
2 1. Deploy proxy charm to a unit
3 2. Execute 'get-ssh-public-key' primitive and get returned value
4 3. Create LXD container with unit's public ssh key
5 4. Verify SSH works between unit and container
7 6. Stop and Destroy LXD container
17 from n2vc
.vnf
import N2VC
22 - id: singlecharmvdu-ns
23 name: singlecharmvdu-ns
24 short-name: singlecharmvdu-ns
25 description: NS with 1 VNFs singlecharmvdu-vnf connected by datanet and mgmtnet VLs
29 - vnfd-id-ref: singlecharmvdu-vnf
37 vim-network-name: mgmt
38 vnfd-connection-point-ref:
39 - vnfd-id-ref: singlecharmvdu-vnf
40 member-vnf-index-ref: '1'
41 vnfd-connection-point-ref: vnf-mgmt
42 - vnfd-id-ref: singlecharmvdu-vnf
43 member-vnf-index-ref: '2'
44 vnfd-connection-point-ref: vnf-mgmt
49 vnfd-connection-point-ref:
50 - vnfd-id-ref: singlecharmvdu-vnf
51 member-vnf-index-ref: '1'
52 vnfd-connection-point-ref: vnf-data
53 - vnfd-id-ref: singlecharmvdu-vnf
54 member-vnf-index-ref: '2'
55 vnfd-connection-point-ref: vnf-data
61 - id: singlecharmvdu-vnf
62 name: singlecharmvdu-vnf
63 short-name: singlecharmvdu-vnf
65 description: A VNF consisting of 2 VDUs w/charms connected to an internal VL, and one VDU with cloud-init
83 internal-connection-point:
84 - id-ref: mgmtVM-internal
85 - id-ref: dataVM-internal
101 external-connection-point-ref: vnf-mgmt
107 internal-connection-point-ref: mgmtVM-internal
108 internal-connection-point:
109 - id: mgmtVM-internal
110 name: mgmtVM-internal
111 short-name: mgmtVM-internal
113 cloud-init-file: cloud-config.txt
117 initial-config-primitive:
131 value: '/home/ubuntu/first-touch-mgmtVM'
137 default-value: '/home/ubuntu/touched'
142 class PythonTest(unittest
.TestCase
):
147 self
.log
= logging
.getLogger()
148 self
.log
.level
= logging
.DEBUG
150 self
.loop
= asyncio
.get_event_loop()
152 # self.container = utils.create_lxd_container()
153 self
.n2vc
= utils
.get_n2vc()
157 self
.container
.stop()
158 self
.container
.delete()
160 self
.loop
.run_until_complete(self
.n2vc
.logout())
162 def n2vc_callback(self
, model_name
, application_name
, workload_status
, workload_message
, task
=None):
163 """We pass the vnfd when setting up the callback, so expect it to be
164 returned as a tuple."""
165 self
.log
.debug("[Callback] Workload status '{}' for application {}".format(workload_status
, application_name
))
166 self
.log
.debug("[Callback] Task: \"{}\"".format(task
))
168 if workload_status
== "exec_primitive" and task
:
169 self
.log
.debug("Getting Primitive Status")
170 # get the uuid from the task
173 # get the status of the action
174 task
= asyncio
.ensure_future(
175 self
.n2vc
.GetPrimitiveStatus(
180 task
.add_done_callback(functools
.partial(self
.n2vc_callback
, model_name
, application_name
, "primitive_status", task
))
182 if workload_status
== "primitive_status" and task
and not self
.container
:
183 self
.log
.debug("Creating LXD container")
185 result
= task
.result()
186 pubkey
= result
['pubkey']
188 self
.container
= utils
.create_lxd_container(pubkey
)
189 mgmtaddr
= self
.container
.state().network
['eth0']['addresses']
191 self
.log
.debug("Setting config ssh-hostname={}".format(mgmtaddr
[0]['address']))
192 task
= asyncio
.ensure_future(
193 self
.n2vc
.ExecutePrimitive(
199 'ssh-hostname': mgmtaddr
[0]['address'],
203 task
.add_done_callback(functools
.partial(self
.n2vc_callback
, model_name
, application_name
, None, None))
205 if workload_status
and not task
:
206 self
.log
.debug("Callback: workload status \"{}\"".format(workload_status
))
208 if workload_status
in ["blocked"] and not self
.container
:
209 self
.log
.debug("Getting public SSH key")
211 # Execute 'get-ssh-public-key' primitive and get returned value
212 task
= asyncio
.ensure_future(
213 self
.n2vc
.ExecutePrimitive(
216 "get-ssh-public-key",
219 'ssh-hostname': '10.195.8.78',
220 'ssh-username': 'ubuntu',
221 'ssh-password': 'ubuntu'
225 task
.add_done_callback(functools
.partial(self
.n2vc_callback
, model_name
, application_name
, "exec_primitive", task
))
228 # task = asyncio.ensure_future(
229 # self.n2vc.ExecutePrimitive(
235 # 'ssh-hostname': '10.195.8.78',
236 # 'ssh-username': 'ubuntu',
237 # 'ssh-password': 'ubuntu'
241 # task.add_done_callback(functools.partial(self.n2vc_callback, None, None, None))
243 elif workload_status
in ["active"]:
244 self
.log
.debug("Removing charm")
245 task
= asyncio
.ensure_future(
246 self
.n2vc
.RemoveCharms(model_name
, application_name
, self
.n2vc_callback
)
248 task
.add_done_callback(functools
.partial(self
.n2vc_callback
, None, None, None))
251 utils
.destroy_lxd_container(self
.container
)
252 self
.container
= None
255 self
.loop
.call_soon_threadsafe(self
.loop
.stop
)
257 def test_deploy_application(self
):
258 """Deploy proxy charm to a unit."""
259 stream_handler
= logging
.StreamHandler(sys
.stdout
)
260 self
.log
.addHandler(stream_handler
)
262 self
.log
.info("Log handler installed")
263 nsd
= utils
.get_descriptor(NSD_YAML
)
264 vnfd
= utils
.get_descriptor(VNFD_YAML
)
268 vca_charms
= os
.getenv('VCA_CHARMS', None)
274 """An inner function to do the deployment of a charm from
277 charm_dir
= "{}/{}".format(vca_charms
, charm
)
279 # Setting this to an IP that will fail the initial config.
280 # This will be detected in the callback, which will execute
281 # the "config" primitive with the right IP address.
282 # mgmtaddr = self.container.state().network['eth0']['addresses']
283 # params['rw_mgmt_ip'] = mgmtaddr[0]['address']
285 # Legacy method is to set the ssh-private-key config
286 # with open(utils.get_juju_private_key(), "r") as f:
287 # pkey = f.readline()
288 # params['ssh-private-key'] = pkey
292 vnf_name
= self
.n2vc
.FormatApplicationName(
298 self
.loop
.run_until_complete(
299 self
.n2vc
.DeployCharms(
310 # Check if the VDUs in this VNF have a charm
311 for vdu
in vnfd
['vdu']:
312 vdu_config
= vdu
.get('vdu-configuration')
314 juju
= vdu_config
['juju']
315 self
.assertIsNotNone(juju
)
317 charm
= juju
['charm']
318 self
.assertIsNotNone(charm
)
320 params
['initial-config-primitive'] = vdu_config
['initial-config-primitive']
325 # Check if this VNF has a charm
326 vnf_config
= vnfd
.get("vnf-configuration")
328 juju
= vnf_config
['juju']
329 self
.assertIsNotNone(juju
)
331 charm
= juju
['charm']
332 self
.assertIsNotNone(charm
)
334 params
['initial-config-primitive'] = vnf_config
['initial-config-primitive']
339 self
.loop
.run_forever()
340 # while self.loop.is_running():
341 # # await asyncio.sleep(1)
345 # ExecutePrimitive(self, nsd, vnfd, vnf_member_index, primitive, callback, *callback_args, **params):
347 # self.loop.run_until_complete(n.DestroyNetworkService(nsd))
349 # self.loop.run_until_complete(self.n2vc.logout())
351 self
.log
.removeHandler(stream_handler
)