Add Tacker monitoring tests

Adding tacker functional test for monitoring feature

Change-Id: I921f32f6265689a30fd1adaa63cc2d29ff84c5a5
Closes-Bug: 1500565
changes/43/234543/4
Santosh Kodicherla 7 years ago
parent 7ae2388903
commit 28051f0ba2

@ -0,0 +1,4 @@
VNF_CIRROS_CREATE_TIMEOUT = 300
VNF_CIRROS_DEAD_TIMEOUT = 250
ACTIVE_SLEEP_TIME = 3
DEAD_SLEEP_TIME = 1

@ -0,0 +1,39 @@
template_name: sample-vnfd-nonparam-respawn
description: demo-example
service_properties:
Id: sample-vnfd
vendor: tacker
version: 1
vdus:
vdu1:
id: vdu1
vm_image: cirros-0.3.4-x86_64-uec
instance_type: m1.tiny
user_data_format: RAW
user_data: |
#!/bin/sh
df -h > /home/cirros/diskinfo
sleep 90
sudo ifdown eth0
network_interfaces:
management:
network: net_mgmt
management: true
pkt_in:
network: net0
pkt_out:
network: net1
placement_policy:
availability_zone: nova
auto-scaling: noop
monitoring_policy: ping
failure_policy: respawn
config:
param0: key0
param1: key1

@ -64,7 +64,8 @@ class BaseTackerTest(base.TestCase):
auth_url=auth_uri)
@classmethod
def wait_until_vnf_status(cls, vnf_id, target_status, timeout):
def wait_until_vnf_status(cls, vnf_id, target_status, timeout,
sleep_interval):
start_time = int(time.time())
while True:
vnf_result = cls.client.show_vnf(vnf_id)
@ -72,8 +73,17 @@ class BaseTackerTest(base.TestCase):
if (status == target_status) or ((int(time.time()) -
start_time) > timeout):
break
time.sleep(5)
time.sleep(sleep_interval)
if (status == target_status):
return target_status
@classmethod
def wait_until_vnf_active(cls, vnf_id, timeout, sleep_interval):
return cls.wait_until_vnf_status(vnf_id, 'ACTIVE', timeout,
sleep_interval)
@classmethod
def wait_until_vnf_active(cls, vnf_id, timeout):
cls.wait_until_vnf_status(vnf_id,'ACTIVE',timeout)
def wait_until_vnf_dead(cls, vnf_id, timeout, sleep_interval):
return cls.wait_until_vnf_status(vnf_id, 'DEAD', timeout,
sleep_interval)

@ -45,8 +45,11 @@ class VnfTestJSON(base.BaseTackerTest):
'vnfd']['id'])
vnf_id = vnf_instance['vnf']['id']
self.wait_until_vnf_active(vnf_id, VNF_CIRROS_CREATE_TIMEOUT)
self.assertEqual(self.client.show_vnf(vnf_id)['vnf']['status'],
sleep_time = 5
vnf_current_status = self.wait_until_vnf_active(vnf_id,
VNF_CIRROS_CREATE_TIMEOUT,
sleep_time)
self.assertEqual(vnf_current_status,
'ACTIVE')
self.assertIsNotNone(self.client.show_vnf(vnf_id)['vnf']['mgmt_url'])
@ -60,4 +63,4 @@ class VnfTestJSON(base.BaseTackerTest):
try:
self.client.delete_vnfd(vnfd_id)
except Exception:
assert False, "vnfd Delete failed"
assert False, "vnfd Delete failed"

@ -0,0 +1,76 @@
# Copyright 2015 Brocade Communications System, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo_config import cfg
from tacker.tests import constants
from tacker.tests.functional.vnfd import base
from tacker.tests.utils import read_file
CONF = cfg.CONF
class VnfTestJSON(base.BaseTackerTest):
def test_create_delete_vnf_monitoring(self):
data = dict()
data['tosca'] = read_file(
'sample_vnfd_no_param_monitoring_respawn.yaml')
toscal = data['tosca']
tosca_arg = {'vnfd': {'attributes': {'vnfd': toscal}}}
#Create vnfd with tosca template
vnfd_instance = self.client.create_vnfd(body=tosca_arg)
self.assertIsNotNone(vnfd_instance)
##Create vnf with vnfd_id
vnfd_id = vnfd_instance['vnfd']['id']
vnf_name = 'test_vnf_with_user_data_respawn'
vnf_arg = {'vnf': {'vnfd_id': vnfd_id, 'name': vnf_name}}
vnf_instance = self.client.create_vnf(body = vnf_arg)
self.assertIsNotNone(vnf_instance)
self.assertIsNotNone(vnf_instance['vnf']['id'])
self.assertIsNotNone(vnf_instance['vnf']['instance_id'])
self.assertEqual(vnf_instance['vnf']['vnfd_id'], vnfd_instance[
'vnfd']['id'])
##Verify vnf is in ACTIVE state, then DEAD state and back ACTIVE again
vnf_id = vnf_instance['vnf']['id']
vnf_current_status = self.wait_until_vnf_active(vnf_id,
constants.VNF_CIRROS_CREATE_TIMEOUT,
constants.ACTIVE_SLEEP_TIME)
self.assertEqual(vnf_current_status, 'ACTIVE')
vnf_current_status = self.wait_until_vnf_dead(vnf_id,
constants.VNF_CIRROS_DEAD_TIMEOUT,
constants.DEAD_SLEEP_TIME)
self.assertEqual(vnf_current_status, 'DEAD')
vnf_current_status = self.wait_until_vnf_active(vnf_id,
constants.VNF_CIRROS_CREATE_TIMEOUT,
constants.ACTIVE_SLEEP_TIME)
self.assertEqual(vnf_current_status, 'ACTIVE')
##Delete vnf_instance with vnf_id
try:
self.client.delete_vnf(vnf_id)
except Exception:
assert False, "vnf Delete failed after the monitor test"
##Delete vnfd_instance
try:
self.client.delete_vnfd(vnfd_id)
except Exception:
assert False, "vnfd Delete failed after the monitor test"
Loading…
Cancel
Save