openstack/tacker
Code Issues Proposed changes

Merge "Improve the sol-vnflcm job"

Browse Source
This commit is contained in:
Zuul 2025-01-10 04:56:45 +00:00 committed by Gerrit Code Review
commit e07f144f27
5 changed files with 636 additions and 1318 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

289
tacker/tests/functional/sol/vnflcm/base.py
View File

@ -26,6 +26,7 @@ from tacker.objects import fields
from tacker.tests.functional import base
from tacker.tests.functional.common.fake_server import FakeServerManager
from tacker.tests.functional.common import logging_utils
from tacker.tests.functional.sol.vnflcm import fake_vnflcm
from tacker.tests import utils
from tacker.vnfm.infra_drivers.openstack import constants as infra_cnst
@ -188,7 +189,7 @@ def _list_vnf_package(tacker_client, **kwargs):
def _create_instantiate_vnf_request_body(flavour_id,
instantiation_level_id=None, vim_id=None, ext_vl=None,
add_params=None):
ext_managed_vl=None, add_params=None):
request_body = {"flavourId": flavour_id}
if instantiation_level_id:
@ -197,6 +198,9 @@ def _create_instantiate_vnf_request_body(flavour_id,
if ext_vl:
request_body["extVirtualLinks"] = ext_vl
if ext_managed_vl:
request_body["extManagedVirtualLinks"] = ext_managed_vl
if vim_id:
request_body["vimConnectionInfo"] = [
{"id": uuidutils.generate_uuid(),
@ -218,6 +222,7 @@ class BaseVnfLcmTest(base.BaseTackerTest):
# can set it false. For example, BaseVnfLcmMultiTenantTest which
# create their own servers for two different tenants.
prepare_fake_server = True
prepare_network = True
@classmethod
def setUpClass(cls):
@ -258,14 +263,16 @@ class BaseVnfLcmTest(base.BaseTackerTest):
self.vim = self.get_vim(vim_list, 'VIM0')
if not self.vim:
assert False, "vim_list is Empty: Default VIM is missing"
result = self._create_network_settings()
self.ext_networks = result.get('ext_networks')
self.ext_vl = result.get('ext_vl')
self.ext_mngd_networks = result.get('ext_mngd_networks')
self.ext_link_ports = result.get('ext_link_ports')
self.ext_subnets = result.get('ext_subnets')
self.changed_ext_networks = result.get('changed_ext_networks')
self.changed_ext_subnets = result.get('changed_ext_subnets')
if self.prepare_network:
result = self._create_network_settings()
self.ext_networks = result.get('ext_networks')
self.ext_vl = result.get('ext_vl')
self.ext_mngd_networks = result.get('ext_mngd_networks')
self.ext_link_ports = result.get('ext_link_ports')
self.ext_subnets = result.get('ext_subnets')
self.changed_ext_networks = result.get('changed_ext_networks')
self.changed_ext_subnets = result.get('changed_ext_subnets')
@classmethod
def _prepare_start_fake_server(cls, fake_server,
@ -401,10 +408,10 @@ class BaseVnfLcmTest(base.BaseTackerTest):
resource_details = self._get_heat_resource(stack_id,
resource_name,
h_client)
image = self._get_image_id_from_resource_attributes(
image = self._get_image_from_resource_attributes(
resource_details)
if image:
image_id_list.append(image.id)
image_id_list.append(image['id'])
return image_id_list
@ -479,7 +486,8 @@ class BaseVnfLcmTest(base.BaseTackerTest):
def _create_vnf_instance_from_body(self, request_body, http_client=None):
if http_client is None:
http_client = self.http_client
request_body['vnfInstanceName'] = self._testMethodName
if not request_body.get('vnfInstanceName'):
request_body['vnfInstanceName'] = self._testMethodName
resp, response_body = http_client.do_request(
self.base_vnf_instances_url,
"POST",
@ -488,19 +496,31 @@ class BaseVnfLcmTest(base.BaseTackerTest):
return resp, response_body
def _delete_vnf_instance(self, id, http_client=None):
timeout = VNF_TERMINATE_TIMEOUT
if http_client is None:
http_client = self.http_client
url = os.path.join(self.base_vnf_instances_url, id)
resp, body = http_client.do_request(url, "DELETE")
start_time = int(time.time())
while True:
resp, body = self.http_client.do_request(url, "DELETE")
if resp.status_code in (204, 404):
break
if ((int(time.time()) - start_time) > timeout):
error = "Failed to delete vnf instance %s"
self.fail(error % id)
return resp, body
def _show_vnf_instance(self, id, http_client=None):
def _show_vnf_instance(self, id, http_client=None, expected_result=None):
if http_client is None:
http_client = self.http_client
show_url = os.path.join(self.base_vnf_instances_url, id)
resp, vnf_instance = http_client.do_request(show_url, "GET")
if expected_result:
self.assertDictSupersetOf(expected_result, vnf_instance)
return resp, vnf_instance
def _list_vnf_instance(self, http_client=None, **kwargs):
@ -511,27 +531,6 @@ class BaseVnfLcmTest(base.BaseTackerTest):
return resp, vnf_instances
def _wait_vnf_instance(self, id,
http_client=None,
instantiation_state=fields.VnfInstanceState.INSTANTIATED,
timeout=VNF_INSTANTIATE_TIMEOUT):
if http_client is None:
http_client = self.http_client
start_time = int(time.time())
while True:
resp, body = self._show_vnf_instance(id, http_client)
if body['instantiationState'] == instantiation_state:
break
if ((int(time.time()) - start_time) > timeout):
error = ("Vnf instance %(id)s status is %(current)s, "
"expected status should be %(expected)s")
self.fail(error % {"id": id,
"current": body['instantiationState'],
"expected": instantiation_state})
time.sleep(5)
def _instantiate_vnf_instance(self, id, request_body, http_client=None):
if http_client is None:
http_client = self.http_client
@ -643,29 +642,7 @@ class BaseVnfLcmTest(base.BaseTackerTest):
return resp, response_body
def _wait_terminate_vnf_instance(self, id, timeout=None, http_client=None):
if http_client is None:
http_client = self.http_client
start_time = int(time.time())
self._wait_vnf_instance(id,
http_client,
instantiation_state=fields.VnfInstanceState.NOT_INSTANTIATED,
timeout=timeout)
# If gracefulTerminationTimeout is set, check whether vnf
# instantiation_state is set to NOT_INSTANTIATED after
# gracefulTerminationTimeout seconds.
if timeout and int(time.time()) - start_time < timeout:
self.fail("Vnf is terminated before graceful termination"
"timeout period")
else:
return
# wait for status completion
time.sleep(VNF_DELETE_COMPLETION_WAIT)
def _get_heat_stack(self, vnf_instance_id, h_client=None,
def _get_heat_stack(self, vnf_instance, h_client=None,
prefix_id='vnflcm_'):
if h_client is None:
h_client = self.h_client
@ -674,16 +651,16 @@ class BaseVnfLcmTest(base.BaseTackerTest):
except Exception:
return None
target_stack_name = prefix_id + vnf_instance_id
target_stakcs = list(
target_stack_name = prefix_id + vnf_instance['id']
target_stacks = list(
filter(
lambda x: x.stack_name == target_stack_name,
stacks))
if len(target_stakcs) == 0:
if len(target_stacks) == 0:
return None
return target_stakcs[0]
return target_stacks[0]
def _delete_heat_stack(self, stack_id, h_client=None):
if h_client is None:
@ -748,6 +725,53 @@ class BaseVnfLcmTest(base.BaseTackerTest):
return resource
def _get_fixed_ips(self, vnf_instance, request_body):
res_name = None
for extvirlink in request_body['extVirtualLinks']:
if 'extCps' not in extvirlink:
continue
for extcps in extvirlink['extCps']:
if 'cpdId' in extcps:
if res_name is None:
res_name = list()
res_name.append(extcps['cpdId'])
break
if res_name is None:
return []
stack = self._get_heat_stack(vnf_instance)
stack_id = stack.id
stack_resource = self._get_heat_resource_list(stack_id, nested_depth=2)
releations = dict()
for elmt in stack_resource:
if elmt.resource_type != 'OS::Neutron::Port':
continue
if elmt.resource_name not in res_name:
continue
parent = getattr(elmt, 'parent_resource', None)
releations[parent] = elmt.resource_name
details = list()
for (parent_name, resource_name) in releations.items():
for elmt in stack_resource:
if parent_name is None:
detail_stack = self._get_heat_resource(
stack_id, resource_name)
elif parent_name != elmt.resource_name:
continue
else:
detail_stack = self._get_heat_resource(
elmt.physical_resource_id, resource_name)
details.append(detail_stack)
ans_list = list()
for detail in details:
ans_list.append(detail.attributes['fixed_ips'])
return ans_list
def _get_heat_stack_template(self, stack_id, nested_depth=0,
h_client=None):
if h_client is None:
@ -759,13 +783,13 @@ class BaseVnfLcmTest(base.BaseTackerTest):
return template
def _get_image_id_from_resource_attributes(self, stack_resource_details):
def _get_image_from_resource_attributes(self, stack_resource_details):
if stack_resource_details is None:
return None
if not hasattr(stack_resource_details, 'attributes'):
return None
return stack_resource_details.attributes.get('image', {}).get('id')
return stack_resource_details.attributes.get('image', {})
def _get_vnfc_instance_id_list(
self,
@ -849,12 +873,12 @@ class BaseVnfLcmTest(base.BaseTackerTest):
def assert_heat_stack_status(
self,
vnf_instance_id,
vnf_instance,
h_client=None,
expected_stack_status=infra_cnst.STACK_CREATE_COMPLETE):
if h_client is None:
h_client = self.h_client
stack = self._get_heat_stack(vnf_instance_id, h_client)
stack = self._get_heat_stack(vnf_instance, h_client)
self.assertEqual(
expected_stack_status,
stack.stack_status)
@ -863,7 +887,7 @@ class BaseVnfLcmTest(base.BaseTackerTest):
self,
vnf_instance,
h_client=None,
expected_glance_image=None,
expected_glance_image_status=None,
expected_resource_status=None):
if h_client is None:
@ -873,12 +897,13 @@ class BaseVnfLcmTest(base.BaseTackerTest):
resource_details = self._get_heat_resource(stack_id,
resource_name,
h_client)
image = self._get_image_id_from_resource_attributes(
resource_details)
image = self._get_image_from_resource_attributes(resource_details)
if image:
self.assertEqual(expected_glance_image, image.status)
image_detail = self._get_glance_image(image['id'])
self.assertEqual(expected_glance_image_status,
image_detail['status'])
stack = self._get_heat_stack(vnf_instance['id'], h_client)
stack = self._get_heat_stack(vnf_instance, h_client)
resources = self._get_heat_resource_list(stack.id,
nested_depth=2, h_client=h_client)
self.assertIsNotNone(resources)
@ -889,7 +914,7 @@ class BaseVnfLcmTest(base.BaseTackerTest):
resource.resource_status)
# FT-checkpoint: Glance-image
if expected_glance_image:
if expected_glance_image_status:
assert_glance_image(stack.id, resource.resource_name,
h_client)
@ -966,26 +991,6 @@ class BaseVnfLcmTest(base.BaseTackerTest):
time.sleep(RETRY_WAIT_TIME)
def _wait_stack_update(self, vnf_instance_id, expected_status,
h_client=None):
if h_client is None:
h_client = self.h_client
timeout = VNF_HEAL_TIMEOUT
start_time = int(time.time())
while True:
stack = self._get_heat_stack(vnf_instance_id, h_client)
if stack.stack_status == expected_status:
break
if ((int(time.time()) - start_time) > timeout):
error = ("Stack %(id)s status is %(current)s, expected status "
"should be %(expected)s")
self.fail(error % {"vnf_instance_name": vnf_instance_id,
"current": stack.status,
"expected": expected_status})
time.sleep(RETRY_WAIT_TIME)
def assert_create_vnf(self, resp, vnf_instance,
fake_server_manager=None):
if fake_server_manager is None:
@ -1047,12 +1052,13 @@ class BaseVnfLcmTest(base.BaseTackerTest):
resp, vnf_instance = self._show_vnf_instance(vnf_instance_id,
http_client)
self.assert_vnf_state(vnf_instance)
self.assert_instantiation_state(vnf_instance)
self.assert_heat_stack_status(vnf_instance['id'], h_client)
self.assert_heat_stack_status(vnf_instance, h_client)
self.assert_heat_resource_status(
vnf_instance,
h_client,
expected_glance_image='active',
expected_glance_image_status='active',
expected_resource_status='CREATE_COMPLETE')
# FT-checkpoint: Notification
@ -1099,7 +1105,7 @@ class BaseVnfLcmTest(base.BaseTackerTest):
self.assert_instantiation_state(vnf_instance)
self.assert_heat_stack_status(
vnf_instance['id'],
vnf_instance,
h_client,
expected_stack_status=expected_stack_status)
@ -1158,7 +1164,7 @@ class BaseVnfLcmTest(base.BaseTackerTest):
fields.VnfInstanceState.NOT_INSTANTIATED)
# FT-checkpoint: Heat stack status.
stack = self._get_heat_stack(vnf_instance_id, h_client)
stack = self._get_heat_stack(vnf_instance, h_client)
self.assertIsNone(stack)
self.assert_heat_resource_status_is_none(
@ -1225,7 +1231,7 @@ class BaseVnfLcmTest(base.BaseTackerTest):
self.assertIsNotNone(status.get('scaleLevel'))
self.assert_heat_stack_status(
vnf_instance['id'],
vnf_instance,
h_client,
expected_stack_status=expected_stack_status)
@ -1255,6 +1261,61 @@ class BaseVnfLcmTest(base.BaseTackerTest):
if expected_show_res:
self._assert_show_res(vnf_instance, expected_show_res)
def assert_change_ext_conn_vnf(
self,
resp,
request_body,
vnf_instance_id,
before_fixed_ips=None,
http_client=None,
h_client=None,
fake_server_manager=None):
if http_client is None:
http_client = self.http_client
if h_client is None:
h_client = self.h_client
if fake_server_manager is None:
fake_server_manager = FAKE_SERVER_MANAGER
self.assertEqual(202, resp.status_code)
resp, vnf_instance = self._show_vnf_instance(vnf_instance_id,
http_client)
self.assert_vnf_state(vnf_instance)
self.assert_instantiation_state(vnf_instance)
self.assert_heat_stack_status(
vnf_instance,
h_client,
expected_stack_status='UPDATE_COMPLETE')
if before_fixed_ips:
after_fixed_ips = self._get_fixed_ips(vnf_instance, request_body)
self.assertNotEqual(before_fixed_ips, after_fixed_ips)
# FT-checkpoint: Notification
callback_url = os.path.join(
MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
notify_mock_responses = self._filter_notify_history(callback_url,
vnf_instance_id, fake_server_manager=fake_server_manager)
self.assertEqual(3, len(notify_mock_responses))
self.assert_notification_mock_response(
notify_mock_responses[0],
'VnfLcmOperationOccurrenceNotification',
'STARTING')
self.assert_notification_mock_response(
notify_mock_responses[1],
'VnfLcmOperationOccurrenceNotification',
'PROCESSING')
self.assert_notification_mock_response(
notify_mock_responses[2],
'VnfLcmOperationOccurrenceNotification',
'COMPLETED')
def assert_rollback_vnf(self, resp, vnf_instance_id,
fake_server_manager=None):
self.assertEqual(202, resp.status_code)
@ -1346,10 +1407,11 @@ class BaseVnfLcmTest(base.BaseTackerTest):
self.assert_vnf_state(vnf_instance)
self.assert_instantiation_state(vnf_instance)
self.assert_heat_stack_status(
vnf_instance['id'],
h_client,
expected_stack_status=expected_stack_status)
if expected_stack_status:
self.assert_heat_stack_status(
vnf_instance,
h_client,
expected_stack_status=expected_stack_status)
# FT-checkpoint: Notification
callback_url = os.path.join(
@ -1486,6 +1548,27 @@ class BaseVnfLcmTest(base.BaseTackerTest):
self.assertIsNotNone(_links.get('self'))
self.assertIsNotNone(_links.get('self').get('href'))
def register_subscription(self):
callback_url = os.path.join(MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
request_body = fake_vnflcm.Subscription.make_create_request_body(
'http://localhost:{}{}'.format(
FAKE_SERVER_MANAGER.SERVER_PORT,
callback_url))
resp, response_body = self._register_subscription(request_body)
self.assertEqual(201, resp.status_code)
self.assert_http_header_location_for_subscription(resp.headers)
self.assert_notification_get(callback_url)
subscription_id = response_body.get('id')
return subscription_id
def assert_subscription_deletion(self, sub_id):
resp, _ = self._delete_subscription(sub_id)
self.assertEqual(204, resp.status_code)
resp, _ = self._show_subscription(sub_id)
self.assertEqual(404, resp.status_code)
def _filter_notify_history(self, callback_url, vnf_instance_id,
fake_server_manager=None):
if fake_server_manager is None:
@ -1498,13 +1581,13 @@ class BaseVnfLcmTest(base.BaseTackerTest):
h for h in notify_histories
if h.request_body.get('vnfInstanceId') == vnf_instance_id]
def _get_heat_stack_show(self, vnf_instance_id, resource_name=None,
def _get_heat_stack_show(self, vnf_instance, resource_name=None,
h_client=None):
"""Retrieve image name of the resource from stack"""
if h_client is None:
h_client = self.h_client
try:
stack = self._get_heat_stack(vnf_instance_id, h_client)
stack = self._get_heat_stack(vnf_instance, h_client)
stack_info = h_client.stacks.get(stack.id)
stack_dict = stack_info.to_dict()
resource_dict = json.loads(stack_dict['parameters']['nfv'])

920
tacker/tests/functional/sol/vnflcm/test_vnf_instance.py
View File

File diff suppressed because it is too large Load Diff

282
tacker/tests/functional/sol/vnflcm/test_vnflcm_noop.py
View File

@ -13,263 +13,48 @@
# License for the specific language governing permissions and limitations
# under the License.
import os
import time
from oslo_serialization import jsonutils
from oslo_utils import uuidutils
from sqlalchemy import desc
from sqlalchemy.orm import joinedload
from tacker.common import exceptions
from tacker import context as t_context
from tacker.db import api as db_api
from tacker.db.db_sqlalchemy import api
from tacker.db.db_sqlalchemy import models
from tacker.objects import fields
from tacker.tests.functional import base
from tacker.tests.functional.sol.vnflcm import base as vnflcm_base
from tacker.tests import utils
VNF_PACKAGE_UPLOAD_TIMEOUT = 300
VNF_INSTANTIATE_TIMEOUT = 600
VNF_TERMINATE_TIMEOUT = 600
RETRY_WAIT_TIME = 5
class VnfLcmTestNoop(vnflcm_base.BaseVnfLcmTest):
def _create_and_upload_vnf_package(tacker_client, csar_package_name,
user_defined_data):
# create vnf package
body = jsonutils.dumps({"userDefinedData": user_defined_data})
resp, vnf_package = tacker_client.do_request(
'/vnfpkgm/v1/vnf_packages', "POST", body=body)
# upload vnf package
file_path = utils.test_etc_sample("etsi/nfv", csar_package_name)
# Generating unique vnfd id. This is required when multiple workers
# are running concurrently. The call below creates a new temporary
# CSAR with unique vnfd id.
file_path, uniqueid = utils.create_csar_with_unique_vnfd_id(file_path)
with open(file_path, 'rb') as file_object:
resp, resp_body = tacker_client.do_request(
'/vnfpkgm/v1/vnf_packages/{id}/package_content'.format(
id=vnf_package['id']),
"PUT", body=file_object, content_type='application/zip')
# wait for onboard
timeout = VNF_PACKAGE_UPLOAD_TIMEOUT
start_time = int(time.time())
show_url = os.path.join('/vnfpkgm/v1/vnf_packages', vnf_package['id'])
vnfd_id = None
while True:
resp, body = tacker_client.do_request(show_url, "GET")
if body['onboardingState'] == "ONBOARDED":
vnfd_id = body['vnfdId']
break
if ((int(time.time()) - start_time) > timeout):
raise Exception("Failed to onboard vnf package")
time.sleep(1)
# remove temporarily created CSAR file
os.remove(file_path)
return vnf_package['id'], vnfd_id
class VnfLcmTest(base.BaseTackerTest):
prepare_network = False
@classmethod
def setUpClass(cls):
cls.tacker_client = base.BaseTackerTest.tacker_http_client()
cls.vnf_package, cls.vnfd_id = \
_create_and_upload_vnf_package(
cls.tacker_client, "test_inst_terminate_vnf_with_vnflcmnoop",
{"key": "file_functional"})
csar_path, _ = vnflcm_base._create_csar_with_unique_vnfd_id(
utils.test_etc_sample("etsi/nfv",
"test_inst_terminate_vnf_with_vnflcmnoop"))
cls.vnf_package, cls.vnfd_id = (
vnflcm_base._create_and_upload_vnf_package(
cls.tacker_client, {"key": "file_functional"},
csar_path))
super(VnfLcmTest, cls).setUpClass()
super(VnfLcmTestNoop, cls).setUpClass()
@classmethod
def tearDownClass(cls):
# Update vnf package operational state to DISABLED
update_req_body = jsonutils.dumps({
"operationalState": "DISABLED"})
base_path = "/vnfpkgm/v1/vnf_packages"
# Update operational state to DISABLED and delete vnf package
for package_id in [cls.vnf_package]:
resp, resp_body = cls.tacker_client.do_request(
'{base_path}/{id}'.format(id=package_id,
base_path=base_path),
"PATCH", content_type='application/json',
body=update_req_body)
vnflcm_base._delete_vnf_package(cls.tacker_client, package_id)
# Delete vnf package
url = '/vnfpkgm/v1/vnf_packages/%s' % package_id
cls.tacker_client.do_request(url, "DELETE")
super(VnfLcmTest, cls).tearDownClass()
super(VnfLcmTestNoop, cls).tearDownClass()
def setUp(self):
super(VnfLcmTest, self).setUp()
self.base_url = "/vnflcm/v1/vnf_instances"
self.context = t_context.get_admin_context()
vim_list = self.client.list_vims()
if not vim_list:
self.skipTest("Vims are not configured")
vim_id = 'VIM0'
vim = self.get_vim(vim_list, vim_id)
if not vim:
self.skipTest("Openstack VIM '%s' is missing" % vim_id)
self.vim_id = vim['id']
def _instantiate_vnf_instance_request(
self, flavour_id, vim_id=None, additional_param=None):
request_body = {"flavourId": flavour_id}
if vim_id:
request_body["vimConnectionInfo"] = [
{"id": uuidutils.generate_uuid(),
"vimId": vim_id,
"vimType": "openstack"}]
if additional_param:
request_body["additionalParams"] = additional_param
return request_body
def _create_vnf_instance(self, vnfd_id, vnf_instance_name=None,
vnf_instance_description=None):
request_body = {'vnfdId': vnfd_id}
if vnf_instance_name:
request_body['vnfInstanceName'] = vnf_instance_name
if vnf_instance_description:
request_body['vnfInstanceDescription'] = vnf_instance_description
resp, response_body = self.http_client.do_request(
self.base_url, "POST", body=jsonutils.dumps(request_body))
return resp, response_body
def _delete_wait_vnf_instance(self, id):
timeout = VNF_TERMINATE_TIMEOUT
url = os.path.join(self.base_url, id)
start_time = int(time.time())
while True:
resp, body = self.http_client.do_request(url, "DELETE")
if 204 == resp.status_code:
break
if ((int(time.time()) - start_time) > timeout):
error = "Failed to delete vnf instance %s"
self.fail(error % id)
time.sleep(RETRY_WAIT_TIME)
def _delete_vnf_instance(self, id):
self._delete_wait_vnf_instance(id)
# verify vnf instance is deleted
url = os.path.join(self.base_url, id)
resp, body = self.http_client.do_request(url, "GET")
self.assertEqual(404, resp.status_code)
def _show_vnf_instance(self, id, expected_result=None):
show_url = os.path.join(self.base_url, id)
resp, vnf_instance = self.http_client.do_request(show_url, "GET")
self.assertEqual(200, resp.status_code)
if expected_result:
self.assertDictSupersetOf(expected_result, vnf_instance)
return vnf_instance
def _vnf_instance_wait(
self, id,
instantiation_state=fields.VnfInstanceState.INSTANTIATED,
timeout=VNF_INSTANTIATE_TIMEOUT):
show_url = os.path.join(self.base_url, id)
start_time = int(time.time())
while True:
resp, body = self.http_client.do_request(show_url, "GET")
if body['instantiationState'] == instantiation_state:
break
if ((int(time.time()) - start_time) > timeout):
error = ("Vnf instance %(id)s status is %(current)s, "
"expected status should be %(expected)s")
self.fail(error % {"id": id,
"current": body['instantiationState'],
"expected": instantiation_state})
time.sleep(RETRY_WAIT_TIME)
def _instantiate_vnf_instance(self, id, request_body):
url = os.path.join(self.base_url, id, "instantiate")
resp, body = self.http_client.do_request(
url, "POST", body=jsonutils.dumps(request_body))
self.assertEqual(202, resp.status_code)
self._vnf_instance_wait(id)
def _terminate_vnf_instance(self, id, request_body):
url = os.path.join(self.base_url, id, "terminate")
resp, body = self.http_client.do_request(
url, "POST", body=jsonutils.dumps(request_body))
self.assertEqual(202, resp.status_code)
timeout = request_body.get('gracefulTerminationTimeout')
start_time = int(time.time())
self._vnf_instance_wait(
id, instantiation_state=fields.VnfInstanceState.NOT_INSTANTIATED,
timeout=VNF_TERMINATE_TIMEOUT)
# If gracefulTerminationTimeout is set, check whether vnf
# instantiation_state is set to NOT_INSTANTIATED after
# gracefulTerminationTimeout seconds.
if timeout and int(time.time()) - start_time < timeout:
self.fail("Vnf is terminated before graceful termination "
"timeout period")
@db_api.context_manager.reader
def _vnf_notify_get_by_id(self, context, vnf_instance_id,
columns_to_join=None):
query = api.model_query(
context, models.VnfLcmOpOccs,
read_deleted="no", project_only=True).filter_by(
vnf_instance_id=vnf_instance_id).order_by(
desc("created_at"))
if columns_to_join:
for column in columns_to_join:
query = query.options(joinedload(column))
result = query.first()
if not result:
raise exceptions.VnfInstanceNotFound(id=vnf_instance_id)
return result
def _wait_vnflcm_op_occs(
self, context, vnf_instance_id, type, timeout,
operation_state='COMPLETED'):
start_time = int(time.time())
while True:
vnflcm_op_occ = self._vnf_notify_get_by_id(
context, vnf_instance_id)
if vnflcm_op_occ.operation_state == operation_state:
break
if ((int(time.time()) - start_time) > timeout):
raise Exception("Failed to wait {} instance".format(type))
time.sleep(RETRY_WAIT_TIME)
super(VnfLcmTestNoop, self).setUp()
def test_instantiate_terminate_vnf_with_vnflcmnoop(self):
# Create subscription and register it.
subscription_id = self.register_subscription()
self.addCleanup(self._delete_subscription, subscription_id)
# create vnf instance
vnf_instance_name = "vnf_with_instantiation_level-%s" % \
uuidutils.generate_uuid()
@ -281,26 +66,25 @@ class VnfLcmTest(base.BaseTackerTest):
self.assertIsNotNone(vnf_instance['id'])
self.assertEqual(201, resp.status_code)
# instantiate vnf instance
request_body = self._instantiate_vnf_instance_request(
"simple", vim_id=self.vim_id)
self._instantiate_vnf_instance(vnf_instance['id'], request_body)
self._wait_vnflcm_op_occs(self.context, vnf_instance['id'],
'instantiate', VNF_INSTANTIATE_TIMEOUT)
request_body = vnflcm_base._create_instantiate_vnf_request_body(
"simple", vim_id=self.vim['id'])
resp, _ = self._instantiate_vnf_instance(vnf_instance["id"],
request_body)
self.assertEqual(202, resp.status_code)
self._wait_lcm_done('COMPLETED', vnf_instance_id=vnf_instance["id"])
# show vnf instance
vnf_instance = self._show_vnf_instance(vnf_instance['id'])
_, vnf_instance = self._show_vnf_instance(vnf_instance['id'])
self.assertEqual(vnf_instance['instantiationState'], 'INSTANTIATED')
vnflcm_op_occ_ins = self._vnf_notify_get_by_id(
self.context, vnf_instance['id'], columns_to_join=None)
self.assertEqual(vnflcm_op_occ_ins.operation, 'INSTANTIATE')
# terminate vnf instance
terminate_req_body = {
"terminationType": fields.VnfInstanceTerminationType.FORCEFUL,
}
self._terminate_vnf_instance(vnf_instance['id'], terminate_req_body)
self._wait_vnflcm_op_occs(self.context, vnf_instance['id'],
'terminate', VNF_TERMINATE_TIMEOUT)
vnflcm_op_occ_term = self._vnf_notify_get_by_id(
self.context, vnf_instance['id'], columns_to_join=None)
self.assertEqual(vnflcm_op_occ_term.operation, 'TERMINATE')
resp, _ = self._terminate_vnf_instance(
vnf_instance['id'], terminate_req_body)
self.assertEqual(202, resp.status_code)
self._wait_lcm_done('COMPLETED', vnf_instance_id=vnf_instance["id"])
# delete vnf instance
self._delete_vnf_instance(vnf_instance['id'])

371
tacker/tests/functional/sol/vnflcm_userdata/test_vnf_instance_with_user_data.py
View File

@ -117,20 +117,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
- Delete subscription
"""
# Create subscription and register it.
callback_url = os.path.join(vnflcm_base.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
request_body = fake_vnflcm.Subscription.make_create_request_body(
'http://localhost:{}{}'.format(
vnflcm_base.FAKE_SERVER_MANAGER.SERVER_PORT,
callback_url))
resp, response_body = self._register_subscription(request_body)
self.assertEqual(201, resp.status_code)
self.assert_http_header_location_for_subscription(resp.headers)
self.assert_notification_get(callback_url)
subscription_id = response_body.get('id')
self.addCleanup(
self._delete_subscription,
subscription_id)
subscription_id = self.register_subscription()
# Pre Setting: Create vnf package.
sample_name = 'functional5'
@ -152,7 +139,6 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
vnf_instance_id = vnf_instance['id']
self._wait_lcm_done(vnf_instance_id=vnf_instance_id)
self.assert_create_vnf(resp, vnf_instance, vnf_package_id)
self.addCleanup(self._delete_vnf_instance, vnf_instance_id)
# Instantiate vnf instance
request_body = fake_vnflcm.VnfInstances.\
@ -170,7 +156,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
expected_show_res=expected_show_res)
# Scale-out vnf instance
stack = self._get_heat_stack(vnf_instance_id)
stack = self._get_heat_stack(vnf_instance)
pre_stack_resource_list = self._get_heat_resource_list(stack.id, 2)
request_body = fake_vnflcm.VnfInstances.make_scale_request_body(
@ -190,7 +176,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
expected_show_res=expected_show_res)
# Scale-in vnf instance
stack = self._get_heat_stack(vnf_instance_id)
stack = self._get_heat_stack(vnf_instance)
pre_stack_resource_list = self._get_heat_resource_list(stack.id, 2)
request_body = (fake_vnflcm.VnfInstances
@ -210,7 +196,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
expected_show_res=expected_show_res)
# Terminate VNF
stack = self._get_heat_stack(vnf_instance_id)
stack = self._get_heat_stack(vnf_instance)
resources_list = self._get_heat_resource_list(stack.id)
resource_name_list = [r.resource_name for r in resources_list]
glance_image_id_list = self._get_glance_image_list_from_stack_resource(
@ -229,11 +215,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
self.assert_delete_vnf(resp, vnf_instance_id, vnf_package_id)
# Subscription delete
resp, response_body = self._delete_subscription(subscription_id)
self.assertEqual(204, resp.status_code)
resp, _ = self._show_subscription(subscription_id)
self.assertEqual(404, resp.status_code)
self.assert_subscription_deletion(subscription_id)
def test_inst_scaling_heal(self):
"""Test basic life cycle operations with sample VNFD.
@ -253,20 +235,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
- Delete subscription
"""
# Create subscription and register it.
callback_url = os.path.join(vnflcm_base.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
request_body = fake_vnflcm.Subscription.make_create_request_body(
'http://localhost:{}{}'.format(
vnflcm_base.FAKE_SERVER_MANAGER.SERVER_PORT,
callback_url))
resp, response_body = self._register_subscription(request_body)
self.assertEqual(201, resp.status_code)
self.assert_http_header_location_for_subscription(resp.headers)
self.assert_notification_get(callback_url)
subscription_id = response_body.get('id')
self.addCleanup(
self._delete_subscription,
subscription_id)
subscription_id = self.register_subscription()
# Pre Setting: Create vnf package.
sample_name = 'functional5'
@ -288,7 +257,6 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
vnf_instance_id = vnf_instance['id']
self._wait_lcm_done(vnf_instance_id=vnf_instance_id)
self.assert_create_vnf(resp, vnf_instance, vnf_package_id)
self.addCleanup(self._delete_vnf_instance, vnf_instance_id)
# Instantiate vnf instance
request_body = fake_vnflcm.VnfInstances.\
@ -306,7 +274,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
expected_show_res=expected_show_res)
# Scale-out vnf instance
stack = self._get_heat_stack(vnf_instance_id)
stack = self._get_heat_stack(vnf_instance)
pre_stack_resource_list = self._get_heat_resource_list(stack.id, 2)
request_body = fake_vnflcm.VnfInstances.make_scale_request_body(
'SCALE_OUT')
@ -325,7 +293,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
# Heal vnf (do not specify vnfc_instace_id)
# pre check heat status.
self.assert_heat_stack_status(vnf_instance_id)
self.assert_heat_stack_status(vnf_instance)
# Heal
request_body = fake_vnflcm.VnfInstances.make_heal_request_body()
resp, _ = self._heal_vnf_instance(vnf_instance_id, request_body)
@ -341,7 +309,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
expected_show_res=expected_show_res)
# Scale-in vnf instance
stack = self._get_heat_stack(vnf_instance_id)
stack = self._get_heat_stack(vnf_instance)
pre_stack_resource_list = self._get_heat_resource_list(stack.id, 2)
request_body = (fake_vnflcm.VnfInstances
.make_scale_request_body('SCALE_IN'))
@ -359,7 +327,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
expected_show_res=expected_show_res)
# Terminate VNF
stack = self._get_heat_stack(vnf_instance_id)
stack = self._get_heat_stack(vnf_instance)
resources_list = self._get_heat_resource_list(stack.id)
resource_name_list = [r.resource_name for r in resources_list]
glance_image_id_list = self._get_glance_image_list_from_stack_resource(
@ -378,11 +346,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
self.assert_delete_vnf(resp, vnf_instance_id, vnf_package_id)
# Subscription delete
resp, response_body = self._delete_subscription(subscription_id)
self.assertEqual(204, resp.status_code)
resp, _ = self._show_subscription(subscription_id)
self.assertEqual(404, resp.status_code)
self.assert_subscription_deletion(subscription_id)
def test_notification_no_virtual_storage(self):
"""Test notification when virtual storage absent in VNFD.
@ -398,20 +362,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
- Delete subscription
"""
# Create subscription and register it.
callback_url = os.path.join(vnflcm_base.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
request_body = fake_vnflcm.Subscription.make_create_request_body(
'http://localhost:{}{}'.format(
vnflcm_base.FAKE_SERVER_MANAGER.SERVER_PORT,
callback_url))
resp, response_body = self._register_subscription(request_body)
self.assertEqual(201, resp.status_code)
self.assert_http_header_location_for_subscription(resp.headers)
self.assert_notification_get(callback_url)
subscription_id = response_body.get('id')
self.addCleanup(
self._delete_subscription,
subscription_id)
subscription_id = self.register_subscription()
# Pre Setting: Create vnf package.
sample_name = 'user_data_sample_virtual_storage_missing'
@ -433,7 +384,6 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
vnf_instance_id = vnf_instance['id']
self._wait_lcm_done(vnf_instance_id=vnf_instance_id)
self.assert_create_vnf(resp, vnf_instance, vnf_package_id)
self.addCleanup(self._delete_vnf_instance, vnf_instance_id)
# Instantiate vnf instance
request_body = fake_vnflcm.VnfInstances. \
@ -461,7 +411,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
str(notify_mock_responses[2].request_body))
# Terminate VNF
stack = self._get_heat_stack(vnf_instance_id)
stack = self._get_heat_stack(vnf_instance)
resources_list = self._get_heat_resource_list(stack.id)
resource_name_list = [r.resource_name for r in resources_list]
glance_image_id_list = self._get_glance_image_list_from_stack_resource(
@ -481,11 +431,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
self.assert_delete_vnf(resp, vnf_instance_id, vnf_package_id)
# Subscription delete
resp, response_body = self._delete_subscription(subscription_id)
self.assertEqual(204, resp.status_code)
resp, _ = self._show_subscription(subscription_id)
self.assertEqual(404, resp.status_code)
self.assert_subscription_deletion(subscription_id)
def test_vnfdid_filter_in_subscription(self):
"""Test notification when filter exists in subscription.
@ -519,33 +465,20 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
# Create subscription with vnf instance's vnfdid filter.
sub_id_1 = self._gen_sub_and_register_sub(
'with_vnfd_id_filter', vnfd_id)
self.addCleanup(
self._delete_subscription,
sub_id_1)
# Create subscription with other vnfdid filter.
sub_id_2 = self._gen_sub_and_register_sub(
'with_other_vnfd_id', uuidutils.generate_uuid())
self.addCleanup(
self._delete_subscription,
sub_id_2)
# Create subscription without filter.
sub_id_3 = self._gen_sub_and_register_sub(
'no_filter', uuidutils.generate_uuid())
self.addCleanup(
self._delete_subscription,
sub_id_3)
sub_id = self._gen_sub_and_register_sub(self._testMethodName, vnfd_id)
self.addCleanup(
self._delete_subscription,
sub_id)
# Create vnf instance
resp, vnf_instance = self._create_vnf_instance_from_body(
fake_vnflcm.VnfInstances.make_create_request_body(vnfd_id))
vnf_instance_id = vnf_instance['id']
self._wait_lcm_done(vnf_instance_id=vnf_instance_id)
self.assert_create_vnf(resp, vnf_instance, vnf_package_id)
self.addCleanup(self._delete_vnf_instance, vnf_instance_id)
vnflcm_base.FAKE_SERVER_MANAGER.clear_history(
os.path.join(vnflcm_base.MOCK_NOTIFY_CALLBACK_URL,
"with_vnfd_id_filter"))
@ -577,7 +510,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
self._check_subscription(name)
# Terminate VNF
stack = self._get_heat_stack(vnf_instance_id)
stack = self._get_heat_stack(vnf_instance)
resources_list = self._get_heat_resource_list(stack.id)
resource_name_list = [r.resource_name for r in resources_list]
glance_image_id_list = self._get_glance_image_list_from_stack_resource(
@ -601,7 +534,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
# Subscription delete
for subsc_id in [sub_id, sub_id_1, sub_id_2, sub_id_3]:
self._assert_subscription_deletion(subsc_id)
self.assert_subscription_deletion(subsc_id)
def test_stack_update_in_scaling(self):
"""Test basic life cycle operations with sample VNFD.
@ -620,20 +553,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
- Delete subscription
"""
# Create subscription and register it.
callback_url = os.path.join(vnflcm_base.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
request_body = fake_vnflcm.Subscription.make_create_request_body(
'http://localhost:{}{}'.format(
vnflcm_base.FAKE_SERVER_MANAGER.SERVER_PORT,
callback_url))
resp, response_body = self._register_subscription(request_body)
self.assertEqual(201, resp.status_code)
self.assert_http_header_location_for_subscription(resp.headers)
self.assert_notification_get(callback_url)
subscription_id = response_body.get('id')
self.addCleanup(
self._delete_subscription,
subscription_id)
subscription_id = self.register_subscription()
# Pre Setting: Create vnf package.
sample_name = 'stack_update_in_scale'
@ -655,7 +575,6 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
vnf_instance_id = vnf_instance['id']
self._wait_lcm_done(vnf_instance_id=vnf_instance_id)
self.assert_create_vnf(resp, vnf_instance, vnf_package_id)
self.addCleanup(self._delete_vnf_instance, vnf_instance_id)
# Instantiate vnf instance
request_body = (fake_vnflcm.VnfInstances.
@ -672,7 +591,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
self.assertEqual(200, resp.status_code)
# Scale-out vnf instance
stack = self._get_heat_stack(vnf_instance_id)
stack = self._get_heat_stack(vnf_instance)
pre_stack_resource_list = self._get_heat_resource_list(stack.id, 2)
request_body = fake_vnflcm.VnfInstances.make_scale_request_body(
@ -687,7 +606,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
# Scale-in vnf instance
stack = self._get_heat_stack(vnf_instance_id)
stack = self._get_heat_stack(vnf_instance)
pre_stack_resource_list = self._get_heat_resource_list(stack.id, 2)
request_body = (fake_vnflcm.VnfInstances
@ -701,7 +620,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
scale_type='SCALE_IN', expected_stack_status='UPDATE_COMPLETE')
# Terminate VNF
stack = self._get_heat_stack(vnf_instance_id)
stack = self._get_heat_stack(vnf_instance)
resources_list = self._get_heat_resource_list(stack.id)
resource_name_list = [r.resource_name for r in resources_list]
glance_image_id_list = self._get_glance_image_list_from_stack_resource(
@ -720,11 +639,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
self.assert_delete_vnf(resp, vnf_instance_id, vnf_package_id)
# Subscription delete
resp, response_body = self._delete_subscription(subscription_id)
self.assertEqual(204, resp.status_code)
resp, _ = self._show_subscription(subscription_id)
self.assertEqual(404, resp.status_code)
self.assert_subscription_deletion(subscription_id)
def test_inst_update_heal_term(self):
"""Test basic life cycle operations.
@ -748,18 +663,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
- Delete subscription.
"""
# Create subscription and register it.
callback_url = os.path.join(vnflcm_base.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
request_body = fake_vnflcm.Subscription.make_create_request_body(
'http://localhost:{}{}'.format(
vnflcm_base.FAKE_SERVER_MANAGER.SERVER_PORT,
callback_url))
resp, response_body = self._register_subscription(request_body)
self.assertEqual(201, resp.status_code)
self.assert_http_header_location_for_subscription(resp.headers)
self.assert_notification_get(callback_url)
subscription_id = response_body.get('id')
self.addCleanup(self._delete_subscription, subscription_id)
subscription_id = self.register_subscription()
# Subscription show
resp, body = self._wait_show_subscription(subscription_id)
@ -855,7 +759,6 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
self._wait_lcm_done(vnf_instance_id=vnf_instance_id)
self.assert_create_vnf(resp, vnf_instance, vnf_package_id)
vnf_instance_name = vnf_instance['vnfInstanceName']
self.addCleanup(self._delete_vnf_instance, vnf_instance_id)
# Instantiate vnf instance
request_body = fake_vnflcm.VnfInstances.make_inst_request_body(
@ -879,7 +782,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
# Get VNF image after instantiate VDU2
image_before_update = self._get_heat_stack_show(
vnf_instance_id, 'VDU2')
vnf_instance, 'VDU2')
self.assertIsNotNone(
image_before_update,
"failed to retrieve image")
@ -916,13 +819,13 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
self.assert_heal_vnf(resp, vnf_instance_id, vnf_package_id)
# Check of Image changes after heal
image_after_heal = self._get_heat_stack_show(vnf_instance_id, 'VDU2')
image_after_heal = self._get_heat_stack_show(vnf_instance, 'VDU2')
self.assertIsNotNone(
image_after_heal, "failed to retrieve image")
self.assertNotEqual(image_before_update, image_after_heal)
# Terminate VNF
stack = self._get_heat_stack(vnf_instance_id)
stack = self._get_heat_stack(vnf_instance)
resources_list = self._get_heat_resource_list(stack.id)
resource_name_list = [r.resource_name for r in resources_list]
glance_image_id_list = self._get_glance_image_list_from_stack_resource(
@ -947,11 +850,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
self.assert_delete_vnf(resp, vnf_instance_id, vnf_package_id)
# Subscription delete
resp, response_body = self._delete_subscription(subscription_id)
self.assertEqual(204, resp.status_code)
resp, show_body = self._show_subscription(subscription_id)
self.assertEqual(404, resp.status_code)
self.assert_subscription_deletion(subscription_id)
def test_stack_param_heal_term(self):
"""Test basic life cycle operations.
@ -976,18 +875,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
- Delete subscription.
"""
# Create subscription and register it.
callback_url = os.path.join(vnflcm_base.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
request_body = fake_vnflcm.Subscription.make_create_request_body(
'http://localhost:{}{}'.format(
vnflcm_base.FAKE_SERVER_MANAGER.SERVER_PORT,
callback_url))
resp, response_body = self._register_subscription(request_body)
self.assertEqual(201, resp.status_code)
self.assert_http_header_location_for_subscription(resp.headers)
self.assert_notification_get(callback_url)
subscription_id = response_body.get('id')
self.addCleanup(self._delete_subscription, subscription_id)
subscription_id = self.register_subscription()
# Subscription show
resp, body = self._wait_show_subscription(subscription_id)
@ -1083,7 +971,6 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
self._wait_lcm_done(vnf_instance_id=vnf_instance_id)
self.assert_create_vnf(resp, vnf_instance, vnf_package_id)
vnf_instance_name = vnf_instance['vnfInstanceName']
self.addCleanup(self._delete_vnf_instance, vnf_instance_id)
# Instantiate vnf instance
request_body = fake_vnflcm.VnfInstances.make_inst_request_body(
@ -1137,13 +1024,13 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
self.assert_heal_vnf(resp, vnf_instance_id, vnf_package_id)
# Check of stack param desired_capacity existence in stack
stack_after_heal = self._get_heat_stack_show(vnf_instance_id)
stack_after_heal = self._get_heat_stack_show(vnf_instance)
self.assertIsNotNone(
stack_after_heal, "failed to retrieve stack")
self.assertIsNotNone(stack_after_heal.get('desired_capacity'))
# Terminate VNF
stack = self._get_heat_stack(vnf_instance_id)
stack = self._get_heat_stack(vnf_instance)
resources_list = self._get_heat_resource_list(stack.id)
resource_name_list = [r.resource_name for r in resources_list]
glance_image_id_list = self._get_glance_image_list_from_stack_resource(
@ -1168,11 +1055,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
self.assert_delete_vnf(resp, vnf_instance_id, vnf_package_id)
# Subscription delete
resp, response_body = self._delete_subscription(subscription_id)
self.assertEqual(204, resp.status_code)
resp, show_body = self._show_subscription(subscription_id)
self.assertEqual(404, resp.status_code)
self.assert_subscription_deletion(subscription_id)
def test_inst_update_pkgid_heal_all(self):
"""Test basic life cycle operations with pkg update.
@ -1192,18 +1075,7 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
- Delete subscription.
"""
# Create subscription and register it.
callback_url = os.path.join(vnflcm_base.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
request_body = fake_vnflcm.Subscription.make_create_request_body(
'http://localhost:{}{}'.format(
vnflcm_base.FAKE_SERVER_MANAGER.SERVER_PORT,
callback_url))
resp, response_body = self._register_subscription(request_body)
self.assertEqual(201, resp.status_code)
self.assert_http_header_location_for_subscription(resp.headers)
self.assert_notification_get(callback_url)
subscription_id = response_body.get('id')
self.addCleanup(self._delete_subscription, subscription_id)
subscription_id = self.register_subscription()
# Pre Setting: Create vnf package.
sample_name = 'functional'
@ -1225,7 +1097,6 @@ class VnfLcmWithUserDataTest(vnflcm_base.BaseVnfLcmTest):
vnf_instance_id = vnf_instance['id']
self._wait_lcm_done(vnf_instance_id=vnf_instance_id)
self.assert_create_vnf(resp, vnf_instance, vnf_package_id)