openstack
/
tacker
Code Issues Proposed changes

Enhancement of HTTP Client 

Provide the option to verify the SSL certificate when accessing an
external server from Tacker. Several parameters have been added to
config to allow verification of SSL certificates when accessing
external NFVO servers, heat servers, and notification endpoints from
Tacker.

Implements: blueprint enhance-http-client
Change-Id: I55b2b53cfe0dc794040d0e46ac13a20524b1d9f0
This commit is contained in:
Yusuke Niimi 2023-02-06 09:05:11 +00:00 committed by Koji Shimizu
parent 4a1b680504
commit aac03ceffc
27 changed files with 2129 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
.zuul.yaml
View File

@ -346,7 +346,7 @@
v2_nfvo:
use_external_nfvo: True
endpoint: http://127.0.0.1:9990
token_endpoint: http://127.0.0.1:9990
token_endpoint: http://127.0.0.1:9990/token
client_id: 229ec984de7547b2b662e968961af5a4
client_password: devstack
use_client_secret_basic: True
@ -613,6 +613,42 @@
vars:
prometheus_setup: true
- job:
name: tacker-functional-devstack-multinode-sol-https-v2
parent: tacker-functional-devstack-multinode-sol
description: |
Multinodes job for SOL devstack-based functional tests
with https request
host-vars:
controller-tacker:
devstack_local_conf:
post-config:
$TACKER_CONF:
v2_vnfm:
notification_verify_cert: true
notification_ca_cert_file: /etc/https_server/ssl/ca.crt
use_oauth2_mtls_for_heat: false
heat_verify_cert: true
heat_ca_cert_file: /etc/https_server/ssl/ca.crt
prometheus_plugin:
fault_management: true
performance_management: true
auto_scaling: true
performance_management_package: tacker.sol_refactored.common.monitoring_plugin_base
performance_management_class: MonitoringPluginStub
v2_nfvo:
use_external_nfvo: true
endpoint: https://localhost:9990
token_endpoint: https://localhost:9990/token
client_id: 229ec984de7547b2b662e968961af5a4
client_password: devstack
nfvo_verify_cert: true
nfvo_ca_cert_file: /etc/https_server/ssl/ca.crt
use_client_secret_basic: true
tox_envlist: dsvm-functional-sol-https-v2
vars:
https_setup: true
- job:
name: tacker-functional-devstack-kubernetes-oidc-auth
parent: tacker-functional-devstack-multinode-sol-kubernetes-v2
@ -683,6 +719,7 @@
- tacker-functional-devstack-multinode-sol-v2-ubuntu-focal
- tacker-functional-devstack-multinode-sol-kubernetes-v2
- tacker-functional-devstack-multinode-sol-multi-tenant
- tacker-functional-devstack-multinode-sol-https-v2
- tacker-functional-devstack-multinode-sol-kubernetes-multi-tenant
- tacker-functional-devstack-kubernetes-oidc-auth
- tacker-functional-devstack-multinode-sol-v2-az-retry

3
doc/source/user/etsi_cnf_auto_healing_fm.rst
View File

@ -109,6 +109,9 @@ For it to work, we need to find ``fault_management`` in
...
[prometheus_plugin]
fault_management = True
[v2_vnfm]
# Enable https access to notification server from Tacker (boolean value)
notification_verify_cert = true
...
After modifying the configuration file, don't forget to restart the

3
doc/source/user/etsi_cnf_auto_scaling_pm.rst
View File

@ -104,6 +104,9 @@ For it to work, we need to find ``performance_management`` in
...
[prometheus_plugin]
performance_management = True
[v2_vnfm]
# Enable https access to notification server from Tacker (boolean value)
notification_verify_cert = true
...
After modifying the configuration file, don't forget to restart the

2
playbooks/devstack/pre.yaml
View File

@ -13,6 +13,8 @@
when: setup_multi_tenant is defined and setup_multi_tenant | bool
- role: setup-multi-az
when: setup_multi_az is defined and setup_multi_az | bool
- role: setup-fake-https-server
when: https_setup is defined and https_setup | bool
- role: bindep
bindep_profile: test
bindep_dir: "{{ zuul_work_dir }}"

5
releasenotes/notes/enhance-http-client-4fc0b823b8e740e9.yaml Normal file
View File

@ -0,0 +1,5 @@
---
features:
- |
Added the ability to verify the certificate of the target server for Tacker
to communicate over https to the NFVO, heat, and notification servers.

8
roles/setup-fake-https-server/defaults/main.yaml Normal file
View File

@ -0,0 +1,8 @@
ssl_dir: "/etc/https_server/ssl"
ca_key: "{{ ssl_dir }}/ca.key"
ca_csr: "{{ ssl_dir }}/ca.csr"
ca_crt: "{{ ssl_dir }}/ca.crt"
serv_key: "{{ ssl_dir }}/https_server.key"
serv_csr: "{{ ssl_dir }}/https_server.csr"
serv_crt: "{{ ssl_dir }}/https_server.crt"
serv_pem: "{{ ssl_dir }}/https_server.pem"

36
roles/setup-fake-https-server/tasks/main.yaml Normal file
View File

@ -0,0 +1,36 @@
- block:
- name: Generate directory for SSL certificate
file:
path: "{{ ssl_dir }}"
state: directory
owner: "root"
group: "root"
mode: "0755"
become: yes
- name: Generate CA key and csr for fake https server
shell: openssl req -newkey rsa:2048 -nodes -subj "/CN=rootca" -keyout {{ ca_key }} -out {{ ca_csr }}
become: yes
- name: Generate CA certificate for fake https server
shell: openssl x509 -req -signkey {{ ca_key }} -days 10000 -in {{ ca_csr }} -out {{ ca_crt }}
become: yes
- name: Generate server key and csr for fake https server
shell: openssl req -newkey rsa:2048 -nodes -subj "/CN=localhost" -keyout {{ serv_key }} -out {{ serv_csr }}
become: yes
- name: Generate server certificate for fake https server
shell: openssl x509 -req -CA {{ ca_crt }} -CAkey {{ ca_key }} -CAcreateserial -days 10000 -in {{ serv_csr }} -out {{ serv_crt }}
become: yes
- name: Generate server pem file for fake https server
shell: cat {{ serv_key }} {{ serv_crt }} > {{ serv_pem }}
become: yes
- name: Update server pem file permission
shell: chmod 755 {{ serv_pem }}
become: yes
when:
- inventory_hostname == 'controller-tacker'

38
tacker/sol_refactored/common/config.py
View File

@ -95,6 +95,34 @@ VNFM_OPTS = [
default=False,
help=_('Enable to delete LCM operation occurrence if True. '
'This is intended to use under development.')),
cfg.BoolOpt('notification_verify_cert',
default=False,
help=_('Enable certificate verification during SSL/TLS '
'communication to notification server.')),
cfg.StrOpt('notification_ca_cert_file',
default='',
help=_('Specifies the root CA certificate to use when the '
'notification_verify_cert option is True.')),
cfg.BoolOpt('use_oauth2_mtls_for_heat',
default=False,
help=_('Enable OAuth2.0 mTLS authentication for heat '
'server.')),
cfg.StrOpt('heat_mtls_ca_cert_file',
default='',
help=_('CA Certificate file used by OAuth2.0 mTLS '
'authentication.')),
cfg.StrOpt('heat_mtls_client_cert_file',
default='',
help=_('Client Certificate file used by OAuth2.0 mTLS '
'authentication.')),
cfg.BoolOpt('heat_verify_cert',
default=False,
help=_('Enable certificate verification during SSL/TLS '
'communication to heat server.')),
cfg.StrOpt('heat_ca_cert_file',
default='',
help=_('Specifies the root CA certificate to use when the '
'heat_verify_cert option is True.'))
]
CONF.register_opts(VNFM_OPTS, 'v2_vnfm')
@ -147,7 +175,15 @@ NFVO_OPTS = [
cfg.BoolOpt('use_client_secret_basic',
default=False,
help=_('Use password authenticatiojn if True, '
'use certificate authentication if False.'))
'use certificate authentication if False.')),
cfg.BoolOpt('nfvo_verify_cert',
default=False,
help=_('Enable certificate verification during SSL/TLS '
'communication to NFVO.')),
cfg.StrOpt('nfvo_ca_cert_file',
default='',
help=_('Specifies the root CA certificate to use when the'
'nfvo_verify_cert option is True.'))
]
CONF.register_opts(NFVO_OPTS, 'v2_nfvo')

24
tacker/sol_refactored/common/fm_subscription_utils.py
View File

@ -50,17 +50,35 @@ def subsc_href(subsc_id, endpoint):
def _get_notification_auth_handle(subsc):
if not subsc.obj_attr_is_set('authentication'):
return http_client.NoAuthHandle()
verify = CONF.v2_vnfm.notification_verify_cert
if verify and CONF.v2_vnfm.notification_ca_cert_file:
verify = CONF.v2_vnfm.notification_ca_cert_file
return http_client.NoAuthHandle(verify=verify)
if subsc.authentication.obj_attr_is_set('paramsBasic'):
param = subsc.authentication.paramsBasic
return http_client.BasicAuthHandle(param.userName, param.password)
verify = CONF.v2_vnfm.notification_verify_cert
if verify and CONF.v2_vnfm.notification_ca_cert_file:
verify = CONF.v2_vnfm.notification_ca_cert_file
return http_client.BasicAuthHandle(
param.userName, param.password, verify=verify)
if subsc.authentication.obj_attr_is_set(
'paramsOauth2ClientCredentials'):
param = subsc.authentication.paramsOauth2ClientCredentials
verify = CONF.v2_vnfm.notification_verify_cert
if verify and CONF.v2_vnfm.notification_ca_cert_file:
verify = CONF.v2_vnfm.notification_ca_cert_file
return http_client.OAuth2AuthHandle(None,
param.tokenEndpoint, param.clientId, param.clientPassword)
param.tokenEndpoint, param.clientId, param.clientPassword,
verify=verify)
if subsc.authentication.obj_attr_is_set('paramsOauth2ClientCert'):
param = subsc.authentication.paramsOauth2ClientCert
ca_cert = CONF.v2_vnfm.notification_mtls_ca_cert_file
client_cert = CONF.v2_vnfm.notification_mtls_client_cert_file
return http_client.OAuth2MtlsAuthHandle(None,
param.tokenEndpoint, param.clientId, ca_cert, client_cert)
# not reach here

47
tacker/sol_refactored/common/http_client.py
View File

@ -120,9 +120,11 @@ class AuthHandle(metaclass=abc.ABCMeta):
class KeystoneTokenAuthHandle(AuthHandle):
def __init__(self, auth_url, context):
def __init__(self, auth_url, context, verify=False,
ca_cert=None, client_cert=None):
self.auth_url = auth_url
self.context = context
self.verify = verify
def get_auth(self, context):
if context is None:
@ -133,7 +135,7 @@ class KeystoneTokenAuthHandle(AuthHandle):
project_domain_id=context.project_domain_id)
def get_session(self, auth, service_type):
_session = session.Session(auth=auth, verify=False)
_session = session.Session(auth=auth, verify=self.verify)
return adapter.Adapter(session=_session,
service_type=service_type)
@ -141,13 +143,15 @@ class KeystoneTokenAuthHandle(AuthHandle):
class KeystonePasswordAuthHandle(AuthHandle):
def __init__(self, auth_url, username, password,
project_name, user_domain_name, project_domain_name):
project_name, user_domain_name, project_domain_name,
verify=False):
self.auth_url = auth_url
self.username = username
self.password = password
self.project_name = project_name
self.user_domain_name = user_domain_name
self.project_domain_name = project_domain_name
self.verify = verify
def get_auth(self, context=None):
return v3.Password(auth_url=self.auth_url,
@ -158,51 +162,56 @@ class KeystonePasswordAuthHandle(AuthHandle):
project_domain_name=self.project_domain_name)
def get_session(self, auth, service_type):
_session = session.Session(auth=auth, verify=False)
_session = session.Session(auth=auth, verify=self.verify)
return adapter.Adapter(session=_session,
service_type=service_type)
class BasicAuthHandle(AuthHandle):
def __init__(self, username, password):
def __init__(self, username, password, verify=False):
self.username = username
self.password = password
self.verify = verify
def get_auth(self, context=None):
return http_basic.HTTPBasicAuth(username=self.username,
password=self.password)
def get_session(self, auth, service_type):
return session.Session(auth=auth, verify=False)
return session.Session(auth=auth, verify=self.verify)
class NoAuthHandle(AuthHandle):
def __init__(self, endpoint=None):
def __init__(self, endpoint=None, verify=False):
self.endpoint = endpoint
self.verify = verify
def get_auth(self, context=None):
return noauth.NoAuth(endpoint=self.endpoint)
def get_session(self, auth, service_type):
return session.Session(auth=auth, verify=False)
return session.Session(auth=auth, verify=self.verify)
class OAuth2AuthPlugin(plugin.FixedEndpointPlugin):
def __init__(self, endpoint, token_endpoint, client_id, client_password):
def __init__(self, endpoint, token_endpoint, client_id, client_password,
verify=False):
super(OAuth2AuthPlugin, self).__init__(endpoint)
self.token_endpoint = token_endpoint
self.client_id = client_id
self.client_password = client_password
self.verify = verify
def get_token(self, session, **kwargs):
auth = BasicAuthHandle(self.client_id,
self.client_password)
self.client_password,
self.verify)
client = HttpClient(auth)
url = self.token_endpoint + '/token'
url = self.token_endpoint
data = {'grant_type': 'client_credentials'}
resp, resp_body = client.do_request(url, "POST",
@ -224,18 +233,20 @@ class OAuth2AuthPlugin(plugin.FixedEndpointPlugin):
class OAuth2AuthHandle(AuthHandle):
def __init__(self, endpoint, token_endpoint, client_id, client_password):
def __init__(self, endpoint, token_endpoint, client_id, client_password,
verify=False):
self.endpoint = endpoint
self.token_endpoint = token_endpoint
self.client_id = client_id
self.client_password = client_password
self.verify = verify
def get_auth(self, context=None):
return OAuth2AuthPlugin(self.endpoint, self.token_endpoint,
self.client_id, self.client_password)
self.client_id, self.client_password, self.verify)
def get_session(self, auth, service_type):
_session = session.Session(auth=auth, verify=False)
_session = session.Session(auth=auth, verify=self.verify)
return adapter.Adapter(session=_session,
service_type=service_type)
@ -270,7 +281,7 @@ class OAuth2MtlsAuthPlugin(plugin.FixedEndpointPlugin):
self.client_cert)
client = HttpClient(auth)
url = f'{self.token_endpoint}'
url = self.token_endpoint
data = {
'grant_type': 'client_credentials',
'client_id': self.client_id
@ -297,11 +308,13 @@ class OAuth2MtlsAuthPlugin(plugin.FixedEndpointPlugin):
class OAuth2MtlsAuthHandle(AuthHandle):
def __init__(self, endpoint, token_endpoint, client_id,
verify_cert, client_cert):
ca_cert, client_cert):
self.endpoint = endpoint
self.token_endpoint = token_endpoint
self.client_id = client_id
self.verify_cert = verify_cert
self.verify_cert = True
if ca_cert:
self.verify_cert = ca_cert
self.client_cert = client_cert
def get_auth(self, context=None):

21
tacker/sol_refactored/common/pm_job_utils.py
View File

@ -94,21 +94,32 @@ def make_pm_job_links(pm_job, endpoint):
def _get_notification_auth_handle(pm_job):
if not pm_job.obj_attr_is_set('authentication'):
return http_client.NoAuthHandle()
verify = CONF.v2_vnfm.notification_verify_cert
if verify and CONF.v2_vnfm.notification_ca_cert_file:
verify = CONF.v2_vnfm.notification_ca_cert_file
return http_client.NoAuthHandle(verify=verify)
if pm_job.authentication.obj_attr_is_set('paramsBasic'):
param = pm_job.authentication.paramsBasic
return http_client.BasicAuthHandle(param.userName, param.password)
verify = CONF.v2_vnfm.notification_verify_cert
if verify and CONF.v2_vnfm.notification_ca_cert_file:
verify = CONF.v2_vnfm.notification_ca_cert_file
return http_client.BasicAuthHandle(param.userName, param.password,
verify=verify)
if pm_job.authentication.obj_attr_is_set(
'paramsOauth2ClientCredentials'):
param = pm_job.authentication.paramsOauth2ClientCredentials
verify = CONF.v2_vnfm.notification_verify_cert
if verify and CONF.v2_vnfm.notification_ca_cert_file:
verify = CONF.v2_vnfm.notification_ca_cert_file
return http_client.OAuth2AuthHandle(
None, param.tokenEndpoint, param.clientId, param.clientPassword)
None, param.tokenEndpoint, param.clientId, param.clientPassword,
verify=verify)
if pm_job.authentication.obj_attr_is_set('paramsOauth2ClientCert'):
param = pm_job.authentication.paramsOauth2ClientCert
verify_cert = CONF.v2_vnfm.notification_mtls_ca_cert_file
ca_cert = CONF.v2_vnfm.notification_mtls_ca_cert_file
client_cert = CONF.v2_vnfm.notification_mtls_client_cert_file
return http_client.OAuth2MtlsAuthHandle(None,
param.tokenEndpoint, param.clientId, verify_cert, client_cert)
param.tokenEndpoint, param.clientId, ca_cert, client_cert)
return None

12
tacker/sol_refactored/common/subscription_utils.py
View File

@ -64,11 +64,19 @@ def _get_notification_auth_handle(subsc):
param.tokenEndpoint, param.clientId, verify_cert, client_cert)
elif 'OAUTH2_CLIENT_CREDENTIALS' in auth.authType:
param = subsc.authentication.paramsOauth2ClientCredentials
verify = CONF.v2_vnfm.notification_verify_cert
if verify and CONF.v2_vnfm.notification_ca_cert_file:
verify = CONF.v2_vnfm.notification_ca_cert_file
return http_client.OAuth2AuthHandle(None,
param.tokenEndpoint, param.clientId, param.clientPassword)
param.tokenEndpoint, param.clientId, param.clientPassword,
verify=verify)
elif 'BASIC' in auth.authType:
param = subsc.authentication.paramsBasic
return http_client.BasicAuthHandle(param.userName, param.password)
verify = CONF.v2_vnfm.notification_verify_cert
if verify and CONF.v2_vnfm.notification_ca_cert_file:
verify = CONF.v2_vnfm.notification_ca_cert_file
return http_client.BasicAuthHandle(param.userName, param.password,
verify=verify)
else:
raise sol_ex.AuthTypeNotFound(auth.authType)
else:

32
tacker/sol_refactored/infra_drivers/openstack/heat_utils.py
View File

@ -16,12 +16,14 @@
from oslo_log import log as logging
from oslo_service import loopingcall
from tacker.sol_refactored.common import config
from tacker.sol_refactored.common import exceptions as sol_ex
from tacker.sol_refactored.common import http_client
LOG = logging.getLogger(__name__)
CONF = config.CONF
CHECK_INTERVAL = 5
@ -29,14 +31,28 @@ CHECK_INTERVAL = 5
class HeatClient(object):
def __init__(self, vim_info):
auth = http_client.KeystonePasswordAuthHandle(
auth_url=vim_info.interfaceInfo['endpoint'],
username=vim_info.accessInfo['username'],
password=vim_info.accessInfo['password'],
project_name=vim_info.accessInfo['project'],
user_domain_name=vim_info.accessInfo['userDomain'],
project_domain_name=vim_info.accessInfo['projectDomain']
)
if CONF.v2_vnfm.use_oauth2_mtls_for_heat:
auth = http_client.OAuth2MtlsAuthHandle(
endpoint=None,
token_endpoint=vim_info.interfaceInfo['endpoint'],
client_id=vim_info.accessInfo['username'],
ca_cert=CONF.v2_vnfm.heat_mtls_ca_cert_file,
client_cert=CONF.v2_vnfm.heat_mtls_client_cert_file
)
else:
verify = CONF.v2_vnfm.heat_verify_cert
if verify and CONF.v2_vnfm.heat_ca_cert_file:
verify = CONF.v2_vnfm.heat_ca_cert_file
auth = http_client.KeystonePasswordAuthHandle(
auth_url=vim_info.interfaceInfo['endpoint'],
username=vim_info.accessInfo['username'],
password=vim_info.accessInfo['password'],
project_name=vim_info.accessInfo['project'],
user_domain_name=vim_info.accessInfo['userDomain'],
project_domain_name=vim_info.accessInfo['projectDomain'],
verify=verify
)
self.client = http_client.HttpClient(auth,
service_type='orchestration')

6
tacker/sol_refactored/nfvo/nfvo_client.py
View File

@ -48,11 +48,15 @@ class NfvoClient(object):
self.is_local = False
self.endpoint = CONF.v2_nfvo.endpoint
if CONF.v2_nfvo.use_client_secret_basic:
verify = CONF.v2_nfvo.nfvo_verify_cert
if verify and CONF.v2_nfvo.nfvo_ca_cert_file:
verify = CONF.v2_nfvo.nfvo_ca_cert_file
auth_handle = http_client.OAuth2AuthHandle(
self.endpoint,
CONF.v2_nfvo.token_endpoint,
CONF.v2_nfvo.client_id,
CONF.v2_nfvo.client_password)
CONF.v2_nfvo.client_password,
verify)
else:
auth_handle = http_client.OAuth2MtlsAuthHandle(
self.endpoint,

8
tacker/tests/functional/common/fake_server.py
View File

@ -17,6 +17,7 @@ import http.server
import inspect
import json
import os
import ssl
import threading
import time
from urllib.parse import urlparse
@ -385,6 +386,13 @@ class FakeServerManager(object):
(self.__class__.__name__,
inspect.currentframe().f_code.co_name))
def set_https_server(self):
CERTFILE = "/etc/https_server/ssl/https_server.pem"
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(CERTFILE)
self.objHttpd.socket = context.wrap_socket(
self.objHttpd.socket, server_side=True)
def start_server(self):
"""Start server in thread."""
LOG.debug('[START] %s()' % inspect.currentframe().f_code.co_name)

0
tacker/tests/functional/sol_https_v2/__init__.py Normal file
View File

322
tacker/tests/functional/sol_https_v2/paramgen.py Normal file
View File

@ -0,0 +1,322 @@
# Copyright (C) 2022 Fujitsu
# All Rights Reserved.
#
# 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.
import datetime
from oslo_utils import uuidutils
from tacker.sol_refactored import objects
def sub_create_https_no_auth(callback_uri):
return {
"callbackUri": callback_uri
}
def sub_create_https_basic_auth(callback_uri):
return {
"callbackUri": callback_uri,
"authentication": {
"authType": [
"BASIC"
],
"paramsBasic": {
"userName": "admin-user",
"password": "devstack"
}
}
}
def sub_create_https_oauth2_auth(callback_uri):
return {
"callbackUri": callback_uri,
"authentication": {
"authType": [
"OAUTH2_CLIENT_CREDENTIALS"
],
"paramsOauth2ClientCredentials": {
"clientId": "229ec984de7547b2b662e968961af5a4",
"clientPassword": "devstack",
"tokenEndpoint": "https://localhost:9990/token"
}
}
}
def create_vnf_min(vnfd_id):
# Omit except for required attributes
# NOTE: Only the following cardinality attributes are set.
# - 1
# - 1..N (1)
return {
"vnfdId": vnfd_id
}
def terminate_vnf_min():
# Omit except for required attributes
# NOTE: Only the following cardinality attributes are set.
# - 1
# - 1..N (1)
return {
"terminationType": "FORCEFUL"
}
def instantiate_vnf_min():
# Omit except for required attributes
# NOTE: Only the following cardinality attributes are set.
# - 1
# - 1..N (1)
return {
"flavourId": "simple"
}
def alert_event_firing(inst_id):
return {
"receiver": "receiver",
"status": "firing",
"alerts": [
{
"status": "firing",
"labels": {
"receiver_type": "tacker",
"function_type": "vnffm",
"vnf_instance_id": inst_id,
"perceived_severity": "WARNING",
"event_type": "PROCESSING_ERROR_ALARM"
},
"annotations": {
"fault_type": "Server Down",
"probable_cause": "Process Terminated",
"fault_details": "pid 12345"
},
"startsAt": "2022-06-21T23:47:36.453Z",
"endsAt": "0001-01-01T00:00:00Z",
"generatorURL": "https://controller147:9090/graph?g0.expr="
"up%7Bjob%3D%22node%22%7D+%3D%3D+0&g0.tab=1",
"fingerprint": "5ef77f1f8a3ecb8d"
}
],
"groupLabels": {},
"commonLabels": {
"alertname": "NodeInstanceDown",
"job": "node"
},
"commonAnnotations": {
"description": "sample"
},
"externalURL": "https://controller147:9093",
"version": "4",
"groupKey": "{}:{}",
"truncatedAlerts": 0
}
def pm_job_https_no_auth(callback_uri, inst_id, host_ip):
return {
"objectType": "Vnf",
"objectInstanceIds": [inst_id],
"criteria": {
"performanceMetric": [
f"VCpuUsageMeanVnf.{inst_id}"],
"collectionPeriod": 5,
"reportingPeriod": 10
},
"callbackUri": callback_uri,
"metadata": {
"monitoring": {
"monitorName": "prometheus",
"driverType": "external",
"targetsInfo": [
{
"prometheusHost": host_ip,
"prometheusHostPort": 50022,
"authInfo": {
"ssh_username": "root",
"ssh_password": "root"
},
"alertRuleConfigPath":
"/tmp",
"prometheusReloadApiEndpoint":
"https://localhost:9990/-/reload",
}
]
}
}
}
def pm_job_https_basic_auth(callback_uri, inst_id, host_ip):
return {
"objectType": "Vnf",
"objectInstanceIds": [inst_id],
"criteria": {
"performanceMetric": [
f"VCpuUsageMeanVnf.{inst_id}"],
"collectionPeriod": 5,
"reportingPeriod": 10,
},
"callbackUri": callback_uri,
"authentication": {
"authType": [
"BASIC"
],
"paramsBasic": {
"userName": "admin-user",
"password": "devstack"
}
},
"metadata": {
"monitoring": {
"monitorName": "prometheus",
"driverType": "external",
"targetsInfo": [
{
"prometheusHost": host_ip,
"prometheusHostPort": 50022,
"authInfo": {
"ssh_username": "root",
"ssh_password": "root"
},
"alertRuleConfigPath":
"/tmp",
"prometheusReloadApiEndpoint":
"https://localhost:9990/-/reload",
}
]
}
}
}
def pm_job_https_oauth2_auth(
callback_uri, inst_id, host_ip):
return {
"objectType": "Vnf",
"objectInstanceIds": [inst_id],
"criteria": {
"performanceMetric": [f"VCpuUsageMeanVnf.{inst_id}"],
"collectionPeriod": 5,
"reportingPeriod": 10,
},
"callbackUri": callback_uri,
"authentication": {
"authType": [
"OAUTH2_CLIENT_CREDENTIALS"
],
"paramsOauth2ClientCredentials": {
"clientId": "229ec984de7547b2b662e968961af5a4",
"clientPassword": "devstack",
"tokenEndpoint": "https://localhost:9990/token"
}
},
"metadata": {
"monitoring": {
"monitorName": "prometheus",
"driverType": "external",
"targetsInfo": [
{
"prometheusHost": host_ip,
"prometheusHostPort": 50022,
"authInfo": {
"ssh_username": "root",
"ssh_password": "root"
},
"alertRuleConfigPath":
"/tmp",
"prometheusReloadApiEndpoint":
"https://localhost:9990/-/reload",
}
]
}
}
}
def pm_event(job_id, inst_id):
return {
"receiver": "receiver",
"status": "firing",
"alerts": [
{
"status": "firing",
"labels": {
"receiver_type": "tacker",
"function_type": "vnfpm",
"job_id": job_id,
"metric": f"VCpuUsageMeanVnf.{inst_id}",
"object_instance_id": inst_id
},
"annotations": {
"value": 99,
},
"startsAt": "2022-06-21T23:47:36.453Z",
"endsAt": "0001-01-01T00:00:00Z",
"generatorURL": "http://controller147:9090/graph?g0.expr=up%7B"
"job%3D%22node%22%7D+%3D%3D+0&g0.tab=1",
"fingerprint": "5ef77f1f8a3ecb8d"
}
],
"groupLabels": {},
"commonLabels": {
"alertname": "NodeInstanceDown",
"job": "node"
},
"commonAnnotations": {
"description": "sample"
},
"externalURL": "http://controller147:9093",
"version": "4",
"groupKey": "{}:{}",
"truncatedAlerts": 0
}
def entries(body, inst_id):
return {
'id': uuidutils.generate_uuid(),
'jobId': body.get("id"),
'entries': [{
'objectType': body.get("objectType"),
'objectInstanceId': inst_id,
'performanceMetric': f"VCpuUsageMeanVnf.{inst_id}",
'performanceValues': [{
'timeStamp': datetime.datetime.now(datetime.timezone.utc),
'value': 0.8
}]
}],
}
def alarm(inst_id):
test_time = datetime.datetime.now(datetime.timezone.utc).isoformat()
return objects.AlarmV1.from_dict({
'id': uuidutils.generate_uuid(),
'managedObjectId': inst_id,
'vnfcInstanceIds': [],
'alarmRaisedTime': test_time,
'ackState': 'UNACKNOWLEDGED',
'perceivedSeverity': "WARNING",
'eventTime': test_time,
'eventType': 'COMMUNICATIONS_ALARM',
'faultType': '',
'probableCause': '',
'isRootCause': False,
'faultDetails': '',
'_links': {}
})

402
tacker/tests/functional/sol_https_v2/test_vnffm_https_basic.py Normal file
View File

@ -0,0 +1,402 @@
# Copyright (C) 2023 Fujitsu
# All Rights Reserved.
#
# 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.
import ddt
import os
import time
from tacker.common import rpc
from tacker import context
from tacker.objects import fields
from tacker.sol_refactored.common import config
from tacker.sol_refactored.conductor import conductor_rpc_v2
from tacker.tests.functional.sol_https_v2 import paramgen
from tacker.tests.functional.sol_separated_nfvo_v2 import fake_grant_v2
from tacker.tests.functional.sol_v2_common import base_v2
from tacker.tests.functional.sol_v2_common import test_vnflcm_basic_common
CONF = config.CONF
@ddt.ddt
class VnfFmWithHttpsRequestTest(test_vnflcm_basic_common.CommonVnfLcmTest):
@classmethod
def setUpClass(cls):
cls.is_https = True
super(VnfFmWithHttpsRequestTest, cls).setUpClass()
rpc.init(CONF)
@classmethod
def tearDownClass(cls):
super(VnfFmWithHttpsRequestTest, cls).tearDownClass()
def setUp(self):
super(VnfFmWithHttpsRequestTest, self).setUp()
def _create_fm_subscription(self, req_body):
path = "/vnffm/v1/subscriptions"
return self.tacker_client.do_request(
path, "POST", body=req_body, version="1.3.0")
def _delete_fm_subscription(self, sub_id):
path = "/vnffm/v1/subscriptions/{}".format(sub_id)
return self.tacker_client.do_request(
path, "DELETE", version="1.3.0")
def _create_fm_alarm(self, req_body):
path = "/alert"
return self.tacker_client.do_request(
path, "POST", body=req_body, version="1.3.0")
def _check_notification(self, callback_url, notify_type):
notify_mock_responses = base_v2.FAKE_SERVER_MANAGER.get_history(
callback_url)
base_v2.FAKE_SERVER_MANAGER.clear_history(
callback_url)
self.assertEqual(1, len(notify_mock_responses))
self.assertEqual(204, notify_mock_responses[0].status_code)
self.assertEqual(notify_type, notify_mock_responses[0].request_body[
'notificationType'])
def test_fm_notification_over_https_no_auth(self):
"""Test FM operations over https with no auth
* About attributes:
Omit except for required attributes.
Only the following cardinality attributes are set.
- 1
- 1..N (1)
* About LCM operations:
This test includes the following operations.
- 1. Create VNF instance
- 2. Instantiate VNF
- 3. Create FM subscription
- 4. Alert-Event (firing)
- 5. FM-Delete-Subscription
- 6. Terminate VNF
- 7. Delete VNF instance
"""
# setup
cur_dir = os.path.dirname(__file__)
basic_lcms_min_path = os.path.join(
cur_dir, "../sol_v2_common/samples/basic_lcms_min")
zip_path_file_1, vnfd_id_1 = self.create_vnf_package(
basic_lcms_min_path, nfvo=True)
vnfd_path = "contents/Definitions/v2_sample2_df_simple.yaml"
self._register_vnf_package_mock_response(vnfd_id_1,
zip_path_file_1)
glance_image = fake_grant_v2.GrantV2.get_sw_image(
basic_lcms_min_path, vnfd_path)
flavour_vdu_dict = fake_grant_v2.GrantV2.get_compute_flavor(
basic_lcms_min_path, vnfd_path)
zone_name_list = self.get_zone_list()
# 1. LCM-Create
create_req = paramgen.create_vnf_min(vnfd_id_1)
resp, body = self.create_vnf_instance(create_req)
self.assertEqual(201, resp.status_code)
inst_id = body['id']
# 2. LCM-Instantiate
self._set_grant_response(
True, 'INSTANTIATE', glance_image=glance_image,
flavour_vdu_dict=flavour_vdu_dict, zone_name_list=zone_name_list)
instantiate_req = paramgen.instantiate_vnf_min()
resp, body = self.instantiate_vnf_instance(inst_id, instantiate_req)
self.assertEqual(202, resp.status_code)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# 3. FM-Create-Subscription
expected_inst_attrs = ['id', 'callbackUri', '_links']
callback_url = os.path.join(base_v2.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
callback_uri = ('https://localhost:'
f'{base_v2.FAKE_SERVER_MANAGER.SERVER_PORT}'
f'{callback_url}')
sub_req = paramgen.sub_create_https_no_auth(callback_uri)
resp, body = self._create_fm_subscription(sub_req)
self.assertEqual(201, resp.status_code)
self.check_resp_headers_in_create(resp)
sub_id = body['id']
self.check_resp_body(body, expected_inst_attrs)
# Test notification
self.assert_notification_get(callback_url)
# 4. Alert-Event (firing)
r = conductor_rpc_v2.PrometheusPluginConductor()
ctx = context.get_admin_context()
alarm = paramgen.alarm(inst_id)
r.store_alarm_info(ctx, alarm)
time.sleep(5)
self._check_notification(callback_url, 'AlarmNotification')
# 5. FM-Delete-Subscription
resp, body = self._delete_fm_subscription(sub_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# 6. LCM-Terminate
terminate_req = paramgen.terminate_vnf_min()
resp, body = self.terminate_vnf_instance(inst_id, terminate_req)
self.assertEqual(202, resp.status_code)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# wait a bit because there is a bit time lag between lcmocc DB
# update and terminate completion.
time.sleep(10)
# check instantiationState of VNF
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
self.assertEqual(fields.VnfInstanceState.NOT_INSTANTIATED,
body['instantiationState'])
# 7. LCM-Delete
resp, body = self.delete_vnf_instance(inst_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# check deletion of VNF instance
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(404, resp.status_code)
def test_fm_notification_over_https_basic_auth(self):
"""Test FM operations over https with basic auth
* About attributes:
Omit except for required attributes.
Only the following cardinality attributes are set.
- 1
- 1..N (1)
* About LCM operations:
This test includes the following operations.
- 1. Create VNF instance
- 2. Instantiate VNF
- 3. Create FM subscription
- 4. Alert-Event (firing)
- 5. FM-Delete-Subscription
- 6. Terminate VNF
- 7. Delete VNF instance
"""
# setup
cur_dir = os.path.dirname(__file__)
basic_lcms_min_path = os.path.join(
cur_dir, "../sol_v2_common/samples/basic_lcms_min")
zip_path_file_1, vnfd_id_1 = self.create_vnf_package(
basic_lcms_min_path, nfvo=True)
vnfd_path = "contents/Definitions/v2_sample2_df_simple.yaml"
self._register_vnf_package_mock_response(vnfd_id_1,
zip_path_file_1)
glance_image = fake_grant_v2.GrantV2.get_sw_image(
basic_lcms_min_path, vnfd_path)
flavour_vdu_dict = fake_grant_v2.GrantV2.get_compute_flavor(
basic_lcms_min_path, vnfd_path)
zone_name_list = self.get_zone_list()
# 1. LCM-Create
create_req = paramgen.create_vnf_min(vnfd_id_1)
resp, body = self.create_vnf_instance(create_req)
self.assertEqual(201, resp.status_code)
inst_id = body['id']
# 2. LCM-Instantiate
self._set_grant_response(
True, 'INSTANTIATE', glance_image=glance_image,
flavour_vdu_dict=flavour_vdu_dict, zone_name_list=zone_name_list)
instantiate_req = paramgen.instantiate_vnf_min()
resp, body = self.instantiate_vnf_instance(inst_id, instantiate_req)
self.assertEqual(202, resp.status_code)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
# 3. FM-Create-Subscription
expected_inst_attrs = ['id', 'callbackUri', '_links']
callback_url = os.path.join(base_v2.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
callback_uri = ('https://localhost:'
f'{base_v2.FAKE_SERVER_MANAGER.SERVER_PORT}'
f'{callback_url}')
sub_req = paramgen.sub_create_https_basic_auth(
callback_uri)
resp, body = self._create_fm_subscription(sub_req)
self.assertEqual(201, resp.status_code)
self.check_resp_headers_in_create(resp)
sub_id = body['id']
self.check_resp_body(body, expected_inst_attrs)
# Test notification
self.assert_notification_get(callback_url)
# 4. Alert-Event (firing)
r = conductor_rpc_v2.PrometheusPluginConductor()
ctx = context.get_admin_context()
alarm = paramgen.alarm(inst_id)
r.store_alarm_info(ctx, alarm)
time.sleep(5)
self._check_notification(callback_url, 'AlarmNotification')
# 5. FM-Delete-Subscription
resp, body = self._delete_fm_subscription(sub_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# 6. LCM-Terminate
terminate_req = paramgen.terminate_vnf_min()
resp, body = self.terminate_vnf_instance(inst_id, terminate_req)
self.assertEqual(202, resp.status_code)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# wait a bit because there is a bit time lag between lcmocc DB
# update and terminate completion.
time.sleep(10)
# check instantiationState of VNF
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
self.assertEqual(fields.VnfInstanceState.NOT_INSTANTIATED,
body['instantiationState'])
# 7. LCM-Delete
resp, body = self.delete_vnf_instance(inst_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# check deletion of VNF instance
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(404, resp.status_code)
def test_fm_notification_over_https_oauth2_cred_auth(self):
"""Test FM operations over https with oauth2 auth
* About attributes:
Omit except for required attributes.
Only the following cardinality attributes are set.
- 1
- 1..N (1)
* About LCM operations:
This test includes the following operations.
- 1. Create VNF instance
- 2. Instantiate VNF
- 3. Create FM subscription
- 4. Alert-Event (firing)
- 5. FM-Delete-Subscription
- 6. Terminate VNF
- 7. Delete VNF instance
"""
# setup
cur_dir = os.path.dirname(__file__)
basic_lcms_min_path = os.path.join(
cur_dir, "../sol_v2_common/samples/basic_lcms_min")
zip_path_file_1, vnfd_id_1 = self.create_vnf_package(
basic_lcms_min_path, nfvo=True)
vnfd_path = "contents/Definitions/v2_sample2_df_simple.yaml"
self._register_vnf_package_mock_response(vnfd_id_1,
zip_path_file_1)
glance_image = fake_grant_v2.GrantV2.get_sw_image(
basic_lcms_min_path, vnfd_path)
flavour_vdu_dict = fake_grant_v2.GrantV2.get_compute_flavor(
basic_lcms_min_path, vnfd_path)
zone_name_list = self.get_zone_list()
# 1. LCM-Create
create_req = paramgen.create_vnf_min(vnfd_id_1)
resp, body = self.create_vnf_instance(create_req)
self.assertEqual(201, resp.status_code)
inst_id = body['id']
# 2. LCM-Instantiate
self._set_grant_response(
True, 'INSTANTIATE', glance_image=glance_image,
flavour_vdu_dict=flavour_vdu_dict, zone_name_list=zone_name_list)
instantiate_req = paramgen.instantiate_vnf_min()
resp, body = self.instantiate_vnf_instance(inst_id, instantiate_req)
self.assertEqual(202, resp.status_code)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
# 3. FM-Create-Subscription
expected_inst_attrs = ['id', 'callbackUri', '_links']
callback_url = os.path.join(base_v2.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
callback_uri = ('https://localhost:'
f'{base_v2.FAKE_SERVER_MANAGER.SERVER_PORT}'
f'{callback_url}')
sub_req = paramgen.sub_create_https_oauth2_auth(
callback_uri)
resp, body = self._create_fm_subscription(sub_req)
self.assertEqual(201, resp.status_code)
self.check_resp_headers_in_create(resp)
sub_id = body['id']
self.check_resp_body(body, expected_inst_attrs)
time.sleep(10)
# Test notification
self.assert_notification_get(callback_url)
# 4. Alert-Event (firing)
r = conductor_rpc_v2.PrometheusPluginConductor()
ctx = context.get_admin_context()
alarm = paramgen.alarm(inst_id)
r.store_alarm_info(ctx, alarm)
time.sleep(5)
self._check_notification(callback_url, 'AlarmNotification')
# 5. FM-Delete-Subscription
resp, body = self._delete_fm_subscription(sub_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# 6. LCM-Terminate
terminate_req = paramgen.terminate_vnf_min()
resp, body = self.terminate_vnf_instance(inst_id, terminate_req)
self.assertEqual(202, resp.status_code)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# wait a bit because there is a bit time lag between lcmocc DB
# update and terminate completion.
time.sleep(10)
# check instantiationState of VNF
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
self.assertEqual(fields.VnfInstanceState.NOT_INSTANTIATED,
body['instantiationState'])
# 7. LCM-Delete
resp, body = self.delete_vnf_instance(inst_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# check deletion of VNF instance
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(404, resp.status_code)

469
tacker/tests/functional/sol_https_v2/test_vnflcm_https_basic.py Normal file
View File

@ -0,0 +1,469 @@
# Copyright (C) 2023 Fujitsu
# All Rights Reserved.
#
# 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.
import ddt
import os
import time
from tacker.sol_refactored.common import config
from tacker.objects import fields
from tacker.tests.functional.sol_https_v2 import paramgen
from tacker.tests.functional.sol_separated_nfvo_v2 import fake_grant_v2
from tacker.tests.functional.sol_v2_common import base_v2
from tacker.tests.functional.sol_v2_common import test_vnflcm_basic_common
CONF = config.CONF
@ddt.ddt
class VnfLcmWithHttpsRequest(test_vnflcm_basic_common.CommonVnfLcmTest):
@classmethod
def setUpClass(cls):
cls.is_https = True
super(VnfLcmWithHttpsRequest, cls).setUpClass()
@classmethod
def tearDownClass(cls):
super(VnfLcmWithHttpsRequest, cls).tearDownClass()
def setUp(self):
super().setUp()
def test_vnflcm_over_https_no_auth(self):
"""Test LCM operations over https with no auth
* About attributes:
Omit except for required attributes.
Only the following cardinality attributes are set.
- 1
- 1..N (1)
* About LCM operations:
This test includes the following operations.
- 1. Create subscription
- 2. Show subscription
- 3. Create VNF instance
- 4. Instantiate VNF
- 5. Show VNF instance
- 6. Terminate VNF
- 7. Delete VNF instance
- 8. Delete subscription
- 9. Show subscription
"""
# setup
cur_dir = os.path.dirname(__file__)
basic_lcms_min_path = os.path.join(
cur_dir, "../sol_v2_common/samples/basic_lcms_min")
zip_path_file_1, vnfd_id_1 = self.create_vnf_package(
basic_lcms_min_path, nfvo=True)
vnfd_path = "contents/Definitions/v2_sample2_df_simple.yaml"
self._register_vnf_package_mock_response(vnfd_id_1,
zip_path_file_1)
glance_image = fake_grant_v2.GrantV2.get_sw_image(
basic_lcms_min_path, vnfd_path)
flavour_vdu_dict = fake_grant_v2.GrantV2.get_compute_flavor(
basic_lcms_min_path, vnfd_path)
zone_name_list = self.get_zone_list()
create_req = paramgen.create_vnf_min(vnfd_id_1)
# 1. LCM-Create-Subscription
callback_url = os.path.join(base_v2.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
callback_uri = ('https://localhost:'
f'{base_v2.FAKE_SERVER_MANAGER.SERVER_PORT}'
f'{callback_url}')
sub_req = paramgen.sub_create_https_no_auth(callback_uri)
resp, body = self.create_subscription(sub_req)
self.assertEqual(201, resp.status_code)
self.check_resp_headers_in_create(resp)
sub_id = body['id']
self.assert_notification_get(callback_url)
# 2. LCM-Show-Subscription
resp, body = self.show_subscription(sub_id)
self.assertEqual(200, resp.status_code)
# 3. LCM-Create
expected_inst_attrs = [
'id',
# 'vnfInstanceName', # omitted
# 'vnfInstanceDescription', # omitted
'vnfdId',
'vnfProvider',
'vnfProductName',
'vnfSoftwareVersion',
'vnfdVersion',
# 'vnfConfigurableProperties', # omitted
# 'vimConnectionInfo', # omitted
'instantiationState',
# 'instantiatedVnfInfo', # omitted
# 'metadata', # omitted
# 'extensions', # omitted
'_links'
]
resp, body = self.create_vnf_instance(create_req)
self.assertEqual(201, resp.status_code)
self.check_resp_headers_in_create(resp)
self.check_resp_body(body, expected_inst_attrs)
inst_id = body['id']
# 4. LCM-Instantiate
self._set_grant_response(
True, 'INSTANTIATE', glance_image=glance_image,
flavour_vdu_dict=flavour_vdu_dict, zone_name_list=zone_name_list)
instantiate_req = paramgen.instantiate_vnf_min()
resp, body = self.instantiate_vnf_instance(inst_id, instantiate_req)
self.assertEqual(202, resp.status_code)
self.check_resp_headers_in_operation_task(resp)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# check instantiationState of VNF
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
self.assertEqual(fields.VnfInstanceState.INSTANTIATED,
body['instantiationState'])
# 5. LCM-Show
self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
# 6. LCM-Terminate
terminate_req = paramgen.terminate_vnf_min()
resp, body = self.terminate_vnf_instance(inst_id, terminate_req)
self.assertEqual(202, resp.status_code)
self.check_resp_headers_in_operation_task(resp)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# wait a bit because there is a bit time lag between lcmocc DB
# update and terminate completion.
time.sleep(10)
# check deletion of Heat-stack
stack_name = "vnf-{}".format(inst_id)
stack_status, _ = self.heat_client.get_status(stack_name)
self.assertIsNone(stack_status)
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
# 7. LCM-Delete
resp, body = self.delete_vnf_instance(inst_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# check deletion of VNF instance
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(404, resp.status_code)
# 8. LCM-Delete-subscription
resp, body = self.delete_subscription(sub_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# 9. LCM-Show-Subscription
resp, body = self.show_subscription(sub_id)
self.assertEqual(404, resp.status_code)
def test_vnflcm_over_https_basic_auth(self, is_nfvo=False):
"""Test LCM operations over https with basic auth
* About attributes:
Omit except for required attributes.
Only the following cardinality attributes are set.
- 1
- 1..N (1)
* About LCM operations:
This test includes the following operations.
- 1. Create subscription
- 2. Show subscription
- 3. Create VNF instance
- 4. Instantiate VNF
- 5. Show VNF instance
- 6. Terminate VNF
- 7. Delete VNF instance
- 8. Delete subscription
- 9. Show subscription
"""
# setup
cur_dir = os.path.dirname(__file__)
basic_lcms_min_path = os.path.join(
cur_dir, "../sol_v2_common/samples/basic_lcms_min")
zip_path_file_1, vnfd_id_1 = self.create_vnf_package(
basic_lcms_min_path, nfvo=True)
vnfd_path = "contents/Definitions/v2_sample2_df_simple.yaml"
self._register_vnf_package_mock_response(vnfd_id_1,
zip_path_file_1)
glance_image = fake_grant_v2.GrantV2.get_sw_image(
basic_lcms_min_path, vnfd_path)
flavour_vdu_dict = fake_grant_v2.GrantV2.get_compute_flavor(
basic_lcms_min_path, vnfd_path)
zone_name_list = self.get_zone_list()
create_req = paramgen.create_vnf_min(vnfd_id_1)
# 1. LCM-Create-Subscription
callback_url = os.path.join(base_v2.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
callback_uri = ('https://localhost:'
f'{base_v2.FAKE_SERVER_MANAGER.SERVER_PORT}'
f'{callback_url}')
sub_req = paramgen.sub_create_https_basic_auth(callback_uri)
resp, body = self.create_subscription(sub_req)
self.assertEqual(201, resp.status_code)
self.check_resp_headers_in_create(resp)
sub_id = body['id']
self.assert_notification_get(callback_url)
# 2. LCM-Show-Subscription
resp, body = self.show_subscription(sub_id)
self.assertEqual(200, resp.status_code)
# 3. LCM-Create
expected_inst_attrs = [
'id',
# 'vnfInstanceName', # omitted
# 'vnfInstanceDescription', # omitted
'vnfdId',
'vnfProvider',
'vnfProductName',
'vnfSoftwareVersion',
'vnfdVersion',
# 'vnfConfigurableProperties', # omitted
# 'vimConnectionInfo', # omitted
'instantiationState',
# 'instantiatedVnfInfo', # omitted
# 'metadata', # omitted
# 'extensions', # omitted
'_links'
]
resp, body = self.create_vnf_instance(create_req)
self.assertEqual(201, resp.status_code)
self.check_resp_headers_in_create(resp)
self.check_resp_body(body, expected_inst_attrs)
inst_id = body['id']
# 4. LCM-Instantiate
self._set_grant_response(
True, 'INSTANTIATE', glance_image=glance_image,
flavour_vdu_dict=flavour_vdu_dict, zone_name_list=zone_name_list)
instantiate_req = paramgen.instantiate_vnf_min()
resp, body = self.instantiate_vnf_instance(inst_id, instantiate_req)
self.assertEqual(202, resp.status_code)
self.check_resp_headers_in_operation_task(resp)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# check instantiationState of VNF
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
self.assertEqual(fields.VnfInstanceState.INSTANTIATED,
body['instantiationState'])
# 5. LCM-Show
self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
# 6. LCM-Terminate
terminate_req = paramgen.terminate_vnf_min()
resp, body = self.terminate_vnf_instance(inst_id, terminate_req)
self.assertEqual(202, resp.status_code)
self.check_resp_headers_in_operation_task(resp)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# wait a bit because there is a bit time lag between lcmocc DB
# update and terminate completion.
time.sleep(10)
# check deletion of Heat-stack
stack_name = "vnf-{}".format(inst_id)
stack_status, _ = self.heat_client.get_status(stack_name)
self.assertIsNone(stack_status)
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
# 7. LCM-Delete
resp, body = self.delete_vnf_instance(inst_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# check deletion of VNF instance
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(404, resp.status_code)
# 8. LCM-Delete-subscription
resp, body = self.delete_subscription(sub_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# 9. LCM-Show-Subscription
resp, body = self.show_subscription(sub_id)
self.assertEqual(404, resp.status_code)
def test_vnflcm_over_https_oauth2_cred_auth(self, is_nfvo=False):
"""Test LCM operations over https with oauth2 auth
* About attributes:
Omit except for required attributes.
Only the following cardinality attributes are set.
- 1
- 1..N (1)
* About LCM operations:
This test includes the following operations.
- 1. Create subscription
- 2. Show subscription
- 3. Create VNF instance
- 4. Instantiate VNF
- 5. Show VNF instance
- 6. Terminate VNF
- 7. Delete VNF instance
- 8. Delete subscription
- 9. Show subscription
"""
# setup
cur_dir = os.path.dirname(__file__)
basic_lcms_min_path = os.path.join(
cur_dir, "../sol_v2_common/samples/basic_lcms_min")
zip_path_file_1, vnfd_id_1 = self.create_vnf_package(
basic_lcms_min_path, nfvo=True)
vnfd_path = "contents/Definitions/v2_sample2_df_simple.yaml"
self._register_vnf_package_mock_response(vnfd_id_1,
zip_path_file_1)
glance_image = fake_grant_v2.GrantV2.get_sw_image(
basic_lcms_min_path, vnfd_path)
flavour_vdu_dict = fake_grant_v2.GrantV2.get_compute_flavor(
basic_lcms_min_path, vnfd_path)
zone_name_list = self.get_zone_list()
create_req = paramgen.create_vnf_min(vnfd_id_1)
# 1. LCM-Create-Subscription
callback_url = os.path.join(base_v2.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
callback_uri = ('https://localhost:'
f'{base_v2.FAKE_SERVER_MANAGER.SERVER_PORT}'
f'{callback_url}')
sub_req = paramgen.sub_create_https_oauth2_auth(callback_uri)
resp, body = self.create_subscription(sub_req)
self.assertEqual(201, resp.status_code)
self.check_resp_headers_in_create(resp)
sub_id = body['id']
self.assert_notification_get(callback_url)
# 2. LCM-Show-Subscription
resp, body = self.show_subscription(sub_id)
self.assertEqual(200, resp.status_code)
# 3. LCM-Create
expected_inst_attrs = [
'id',
# 'vnfInstanceName', # omitted
# 'vnfInstanceDescription', # omitted
'vnfdId',
'vnfProvider',
'vnfProductName',
'vnfSoftwareVersion',
'vnfdVersion',
# 'vnfConfigurableProperties', # omitted
# 'vimConnectionInfo', # omitted
'instantiationState',
# 'instantiatedVnfInfo', # omitted
# 'metadata', # omitted
# 'extensions', # omitted
'_links'
]
resp, body = self.create_vnf_instance(create_req)
self.assertEqual(201, resp.status_code)
self.check_resp_headers_in_create(resp)
self.check_resp_body(body, expected_inst_attrs)
inst_id = body['id']
# 4. LCM-Instantiate
self._set_grant_response(
True, 'INSTANTIATE', glance_image=glance_image,
flavour_vdu_dict=flavour_vdu_dict, zone_name_list=zone_name_list)
instantiate_req = paramgen.instantiate_vnf_min()
resp, body = self.instantiate_vnf_instance(inst_id, instantiate_req)
self.assertEqual(202, resp.status_code)
self.check_resp_headers_in_operation_task(resp)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# check instantiationState of VNF
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
self.assertEqual(fields.VnfInstanceState.INSTANTIATED,
body['instantiationState'])
# 5. LCM-Show
self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
# 6. LCM-Terminate
terminate_req = paramgen.terminate_vnf_min()
resp, body = self.terminate_vnf_instance(inst_id, terminate_req)
self.assertEqual(202, resp.status_code)
self.check_resp_headers_in_operation_task(resp)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# wait a bit because there is a bit time lag between lcmocc DB
# update and terminate completion.
time.sleep(10)
# check deletion of Heat-stack
stack_name = "vnf-{}".format(inst_id)
stack_status, _ = self.heat_client.get_status(stack_name)
self.assertIsNone(stack_status)
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
# 7. LCM-Delete
resp, body = self.delete_vnf_instance(inst_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# check deletion of VNF instance
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(404, resp.status_code)
# 8. LCM-Delete-subscription
resp, body = self.delete_subscription(sub_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# 9. LCM-Show-Subscription
resp, body = self.show_subscription(sub_id)
self.assertEqual(404, resp.status_code)

444
tacker/tests/functional/sol_https_v2/test_vnfpm_https_basic.py Normal file
View File

@ -0,0 +1,444 @@
# Copyright (C) 2022 Fujitsu
# All Rights Reserved.
#
# 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.
import ddt
import os
import time
from tacker.common import rpc
from tacker import context
from tacker.sol_refactored.common import config
from tacker.sol_refactored.conductor import conductor_rpc_v2
from tacker.tests.functional.sol_https_v2 import paramgen
from tacker.tests.functional.sol_separated_nfvo_v2 import fake_grant_v2
from tacker.tests.functional.sol_v2_common import base_v2
from tacker.tests.functional.sol_v2_common import test_vnflcm_basic_common
CONF = config.CONF
@ddt.ddt
class VnfPmWithHttpsRequestTest(test_vnflcm_basic_common.CommonVnfLcmTest):
@classmethod
def setUpClass(cls):
cls.is_https = True
super(VnfPmWithHttpsRequestTest, cls).setUpClass()
cls.fake_prometheus_ip = cls._get_controller_tacker_ip(cls)
rpc.init(CONF)
@classmethod
def tearDownClass(cls):
super(VnfPmWithHttpsRequestTest, cls).tearDownClass()
def setUp(self):
super(VnfPmWithHttpsRequestTest, self).setUp()
base_v2.FAKE_SERVER_MANAGER.set_callback(
'PUT', "/-/reload", status_code=202,
response_headers={"Content-Type": "text/plain"})
def _get_controller_tacker_ip(cls):
cur_dir = os.path.dirname(__file__)
script_path = os.path.join(
cur_dir, "../../../../tools/test-setup-fake-prometheus-server.sh")
with open(script_path, 'r') as f_obj:
content = f_obj.read()
ip = content.split('TEST_REMOTE_URI')[1].split(
'http://')[1].split('"')[0]
return ip
def _create_pm_job(self, req_body):
path = "/vnfpm/v2/pm_jobs"
return self.tacker_client.do_request(
path, "POST", body=req_body, version="2.1.0")
def _create_pm_event(self, req_body):
path = "/pm_event"
return self.tacker_client.do_request(
path, "POST", body=req_body, version="2.1.0")
def _check_notification(self, callback_url, notify_type):
notify_mock_responses = base_v2.FAKE_SERVER_MANAGER.get_history(
callback_url)
base_v2.FAKE_SERVER_MANAGER.clear_history(
callback_url)
self.assertEqual(1, len(notify_mock_responses))
self.assertEqual(204, notify_mock_responses[0].status_code)
self.assertEqual(notify_type, notify_mock_responses[0].request_body[
'notificationType'])
def _delete_pm_job(self, pm_job_id):
path = f"/vnfpm/v2/pm_jobs/{pm_job_id}"
return self.tacker_client.do_request(
path, "DELETE", version="2.1.0")
def test_pm_notification_over_https_no_auth(self):
"""Test PM operations over https with no auth
* About attributes:
Omit except for required attributes.
Only the following cardinality attributes are set.
- 1
- 1..N (1)
* About LCM operations:
This test includes the following operations.
- 1. Create VNF instance
- 2. Instantiate VNF
- 3. PMJob-Create
- 4. PM-Event
- 5. PMJob-Delete
- 6. Terminate VNF
- 7. Delete VNF instance
"""
# setup
cur_dir = os.path.dirname(__file__)
basic_lcms_min_path = os.path.join(
cur_dir, "../sol_v2_common/samples/basic_lcms_min")
zip_path_file_1, vnfd_id_1 = self.create_vnf_package(
basic_lcms_min_path, nfvo=True)
vnfd_path = "contents/Definitions/v2_sample2_df_simple.yaml"
self._register_vnf_package_mock_response(vnfd_id_1,
zip_path_file_1)
glance_image = fake_grant_v2.GrantV2.get_sw_image(
basic_lcms_min_path, vnfd_path)
flavour_vdu_dict = fake_grant_v2.GrantV2.get_compute_flavor(
basic_lcms_min_path, vnfd_path)
zone_name_list = self.get_zone_list()
create_req = paramgen.create_vnf_min(vnfd_id_1)
# 1. LCM-Create
self._set_grant_response(
True, 'INSTANTIATE', glance_image=glance_image,
flavour_vdu_dict=flavour_vdu_dict, zone_name_list=zone_name_list)
create_req = paramgen.create_vnf_min(vnfd_id_1)
resp, body = self.create_vnf_instance(create_req)
self.assertEqual(201, resp.status_code)
inst_id = body['id']
# 2. LCM-Instantiate
self._set_grant_response(
True, 'INSTANTIATE', glance_image=glance_image,
flavour_vdu_dict=flavour_vdu_dict, zone_name_list=zone_name_list)
instantiate_req = paramgen.instantiate_vnf_min()
resp, body = self.instantiate_vnf_instance(inst_id, instantiate_req)
self.assertEqual(202, resp.status_code)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# 3. PMJob-Create
pm_expected_attrs = [
'id',
'objectType',
'objectInstanceIds',
'criteria',
'callbackUri',
'_links'
]
callback_url = os.path.join(base_v2.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
callback_uri = ('https://localhost:'
f'{base_v2.FAKE_SERVER_MANAGER.SERVER_PORT}'
f'{callback_url}')
sub_req = paramgen.pm_job_https_no_auth(
callback_uri, inst_id, self.fake_prometheus_ip)
resp, body = self._create_pm_job(sub_req)
self.assertEqual(201, resp.status_code)
self.check_resp_headers_in_create(resp)
self.check_resp_body(body, pm_expected_attrs)
# Test notification
self.assert_notification_get(callback_url)
pm_job_id = body.get('id')
# 4. PM-Event
r = conductor_rpc_v2.PrometheusPluginConductor()
ctx = context.get_admin_context()
entries = paramgen.entries(body, inst_id)
r.store_job_info(ctx, entries)
time.sleep(5)
self._check_notification(
callback_url, 'PerformanceInformationAvailableNotification')
# 5. PMJob-Delete
resp, body = self._delete_pm_job(pm_job_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# 6. LCM-Terminate
terminate_req = paramgen.terminate_vnf_min()
resp, body = self.terminate_vnf_instance(inst_id, terminate_req)
self.assertEqual(202, resp.status_code)
self.check_resp_headers_in_operation_task(resp)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# wait a bit because there is a bit time lag between lcmocc DB
# update and terminate completion.
time.sleep(10)
# check deletion of Heat-stack
stack_name = "vnf-{}".format(inst_id)
stack_status, _ = self.heat_client.get_status(stack_name)
self.assertIsNone(stack_status)
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
# 7. LCM-Delete
resp, body = self.delete_vnf_instance(inst_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# check deletion of VNF instance
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(404, resp.status_code)
def test_pm_notification_over_https_basic_auth(self):
"""Test PM operations over https with basic auth
* About attributes:
Omit except for required attributes.
Only the following cardinality attributes are set.
- 1
- 1..N (1)
* About LCM operations:
This test includes the following operations.
- 1. Create VNF instance
- 2. Instantiate VNF
- 3. PMJob-Create
- 4. PM-Event
- 5. PMJob-Delete
- 6. Terminate VNF
- 7. Delete VNF instance
"""
# setup
cur_dir = os.path.dirname(__file__)
basic_lcms_min_path = os.path.join(
cur_dir, "../sol_v2_common/samples/basic_lcms_min")
zip_path_file_1, vnfd_id_1 = self.create_vnf_package(
basic_lcms_min_path, nfvo=True)
vnfd_path = "contents/Definitions/v2_sample2_df_simple.yaml"
self._register_vnf_package_mock_response(vnfd_id_1,
zip_path_file_1)
glance_image = fake_grant_v2.GrantV2.get_sw_image(
basic_lcms_min_path, vnfd_path)
flavour_vdu_dict = fake_grant_v2.GrantV2.get_compute_flavor(
basic_lcms_min_path, vnfd_path)
zone_name_list = self.get_zone_list()
create_req = paramgen.create_vnf_min(vnfd_id_1)
# 1. LCM-Create
create_req = paramgen.create_vnf_min(vnfd_id_1)
resp, body = self.create_vnf_instance(create_req)
self.assertEqual(201, resp.status_code)
inst_id = body['id']
# 2. LCM-Instantiate
self._set_grant_response(
True, 'INSTANTIATE', glance_image=glance_image,
flavour_vdu_dict=flavour_vdu_dict, zone_name_list=zone_name_list)
instantiate_req = paramgen.instantiate_vnf_min()
resp, body = self.instantiate_vnf_instance(inst_id, instantiate_req)
self.assertEqual(202, resp.status_code)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# 3. PMJob-Create
pm_expected_attrs = [
'id',
'objectType',
'objectInstanceIds',
'criteria',
'callbackUri',
'_links'
]
callback_url = os.path.join(base_v2.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
callback_uri = ('https://localhost:'
f'{base_v2.FAKE_SERVER_MANAGER.SERVER_PORT}'
f'{callback_url}')
sub_req = paramgen.pm_job_https_basic_auth(
callback_uri, inst_id, self.fake_prometheus_ip)
resp, body = self._create_pm_job(sub_req)
self.assertEqual(201, resp.status_code)
self.check_resp_headers_in_create(resp)
self.check_resp_body(body, pm_expected_attrs)
# Test notification
self.assert_notification_get(callback_url)
pm_job_id = body.get('id')
# 4. PM-Event
r = conductor_rpc_v2.PrometheusPluginConductor()
ctx = context.get_admin_context()
entries = paramgen.entries(body, inst_id)
r.store_job_info(ctx, entries)
time.sleep(5)
self._check_notification(
callback_url, 'PerformanceInformationAvailableNotification')
# 5. PMJob-Delete
resp, body = self._delete_pm_job(pm_job_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# 6. LCM-Terminate
terminate_req = paramgen.terminate_vnf_min()
resp, body = self.terminate_vnf_instance(inst_id, terminate_req)
self.assertEqual(202, resp.status_code)
self.check_resp_headers_in_operation_task(resp)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# wait a bit because there is a bit time lag between lcmocc DB
# update and terminate completion.
time.sleep(10)
# check deletion of Heat-stack
stack_name = "vnf-{}".format(inst_id)
stack_status, _ = self.heat_client.get_status(stack_name)
self.assertIsNone(stack_status)
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
# 7. LCM-Delete
resp, body = self.delete_vnf_instance(inst_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# check deletion of VNF instance
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(404, resp.status_code)
def test_pm_notification_over_https_oauth2_cred_auth(self):
"""Test PM operations over https with oauth2 auth
* About attributes:
Omit except for required attributes.
Only the following cardinality attributes are set.
- 1
- 1..N (1)
* About LCM operations:
This test includes the following operations.
- 1. Create VNF instance
- 2. Instantiate VNF
- 3. PMJob-Create
- 4. PM-Event
- 5. PMJob-Delete
- 6. Terminate VNF
- 7. Delete VNF instance
"""
# setup
cur_dir = os.path.dirname(__file__)
basic_lcms_min_path = os.path.join(
cur_dir, "../sol_v2_common/samples/basic_lcms_min")
zip_path_file_1, vnfd_id_1 = self.create_vnf_package(
basic_lcms_min_path, nfvo=True)
vnfd_path = "contents/Definitions/v2_sample2_df_simple.yaml"
self._register_vnf_package_mock_response(vnfd_id_1,
zip_path_file_1)
glance_image = fake_grant_v2.GrantV2.get_sw_image(
basic_lcms_min_path, vnfd_path)
flavour_vdu_dict = fake_grant_v2.GrantV2.get_compute_flavor(
basic_lcms_min_path, vnfd_path)
zone_name_list = self.get_zone_list()
create_req = paramgen.create_vnf_min(vnfd_id_1)
# 1. LCM-Create
self._set_grant_response(
True, 'INSTANTIATE', glance_image=glance_image,
flavour_vdu_dict=flavour_vdu_dict, zone_name_list=zone_name_list)
create_req = paramgen.create_vnf_min(vnfd_id_1)
resp, body = self.create_vnf_instance(create_req)
self.assertEqual(201, resp.status_code)
inst_id = body['id']
# 2. LCM-Instantiate
instantiate_req = paramgen.instantiate_vnf_min()
resp, body = self.instantiate_vnf_instance(inst_id, instantiate_req)
self.assertEqual(202, resp.status_code)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# 3. PMJob-Create
pm_expected_attrs = [
'id',
'objectType',
'objectInstanceIds',
'criteria',
'callbackUri',
'_links'
]
callback_url = os.path.join(base_v2.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
callback_uri = ('https://localhost:'
f'{base_v2.FAKE_SERVER_MANAGER.SERVER_PORT}'
f'{callback_url}')
sub_req = paramgen.pm_job_https_oauth2_auth(
callback_uri, inst_id, self.fake_prometheus_ip)
resp, body = self._create_pm_job(sub_req)
self.assertEqual(201, resp.status_code)
self.check_resp_headers_in_create(resp)
self.check_resp_body(body, pm_expected_attrs)
# Test notification
self.assert_notification_get(callback_url)
pm_job_id = body.get('id')
# 4. PM-Event
r = conductor_rpc_v2.PrometheusPluginConductor()
ctx = context.get_admin_context()
entries = paramgen.entries(body, inst_id)
r.store_job_info(ctx, entries)
time.sleep(5)
self._check_notification(
callback_url, 'PerformanceInformationAvailableNotification')
# 5. PMJob-Delete
resp, body = self._delete_pm_job(pm_job_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# 6. LCM-Terminate
terminate_req = paramgen.terminate_vnf_min()
resp, body = self.terminate_vnf_instance(inst_id, terminate_req)
self.assertEqual(202, resp.status_code)
self.check_resp_headers_in_operation_task(resp)
lcmocc_id = os.path.basename(resp.headers['Location'])
self.wait_lcmocc_complete(lcmocc_id)
# wait a bit because there is a bit time lag between lcmocc DB
# update and terminate completion.
time.sleep(10)
# check deletion of Heat-stack
stack_name = "vnf-{}".format(inst_id)
stack_status, _ = self.heat_client.get_status(stack_name)
self.assertIsNone(stack_status)
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(200, resp.status_code)
# 7. LCM-Delete
resp, body = self.delete_vnf_instance(inst_id)
self.assertEqual(204, resp.status_code)
self.check_resp_headers_in_delete(resp)
# check deletion of VNF instance
resp, body = self.show_vnf_instance(inst_id)
self.assertEqual(404, resp.status_code)

2
tacker/tests/functional/sol_v2_common/base_v2.py
View File

@ -51,6 +51,8 @@ class BaseSolV2Test(base.BaseTestCase):
super(BaseSolV2Test, cls).setUpClass()
FAKE_SERVER_MANAGER.prepare_http_server()
if getattr(cls, 'is_https', False):
FAKE_SERVER_MANAGER.set_https_server()
FAKE_SERVER_MANAGER.start_server()
cfg.CONF(args=['--config-file', '/etc/tacker/tacker.conf'],

55
tacker/tests/unit/sol_refactored/common/test_fm_subscription_utils.py
View File

@ -16,6 +16,7 @@ import copy
import requests
from unittest import mock
from oslo_config import cfg
from oslo_log import log as logging
from tacker import context
@ -70,7 +71,7 @@ class TestFmSubscriptionUtils(base.BaseTestCase):
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
# execute no_auth
# 1. execute no_auth
subsc_utils.send_notification(subsc_no_auth, notif_data_no_auth)
subsc_basic_auth = copy.deepcopy(subsc_no_auth)
@ -78,7 +79,7 @@ class TestFmSubscriptionUtils(base.BaseTestCase):
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test', password='test'))
# execute basic_auth
# 2. execute basic_auth
subsc_utils.send_notification(subsc_basic_auth, notif_data_no_auth)
subsc_oauth2 = copy.deepcopy(subsc_no_auth)
@ -88,10 +89,56 @@ class TestFmSubscriptionUtils(base.BaseTestCase):
clientId='test', clientPassword='test',
tokenEndpoint='http://127.0.0.1/token')))
# execute oauth2
# 3. execute oauth2
subsc_utils.send_notification(subsc_oauth2, notif_data_no_auth)
self.assertEqual(3, mock_resp.call_count)
subsc_oauth2_mtls = copy.deepcopy(subsc_no_auth)
subsc_oauth2_mtls.authentication = objects.SubscriptionAuthentication(
paramsOauth2ClientCert=(
objects.SubscriptionAuthentication_ParamsOauth2ClientCert(
clientId='test',
certificateRef=objects.
ParamsOauth2ClientCert_CertificateRef(
type='x5t#256',
value='03c6e188d1fe5d3da8c9bc9a8dc531a2'
'b3ecf812b03aede9bec7ba1b410b6b64'
),
tokenEndpoint='http://127.0.0.1/token'
)
)
)
# 4. execute oauth2 mTLS
subsc_utils.send_notification(subsc_oauth2_mtls, notif_data_no_auth)
cfg.CONF.set_override("notification_verify_cert", "True",
group="v2_vnfm")
subsc_no_auth = objects.FmSubscriptionV1.from_dict(
fakes_for_fm.fm_subsc_example)
# 5. execute no_auth
subsc_utils.send_notification(subsc_no_auth, notif_data_no_auth)
subsc_basic_auth = copy.deepcopy(subsc_no_auth)
subsc_basic_auth.authentication = objects.SubscriptionAuthentication(
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test', password='test'))
# 6. execute basic_auth
subsc_utils.send_notification(subsc_basic_auth, notif_data_no_auth)
subsc_oauth2 = copy.deepcopy(subsc_no_auth)
subsc_oauth2.authentication = objects.SubscriptionAuthentication(
paramsOauth2ClientCredentials=(
objects.SubscriptionAuthentication_ParamsOauth2(
clientId='test', clientPassword='test',
tokenEndpoint='http://127.0.0.1/token')))
# 7. execute oauth2
subsc_utils.send_notification(subsc_oauth2, notif_data_no_auth)
self.assertEqual(7, mock_resp.call_count)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_send_notification_error_code(self, mock_resp):

31
tacker/tests/unit/sol_refactored/common/test_pm_job_utils.py
View File

@ -12,6 +12,8 @@
# 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 oslo_utils import uuidutils
import requests
from unittest import mock
@ -173,6 +175,35 @@ class TestPmJobUtils(base.BaseTestCase):
result = pm_job_utils._get_notification_auth_handle(pm_job_3)
self.assertIsInstance(result, http_client.OAuth2MtlsAuthHandle)
cfg.CONF.set_override("notification_verify_cert", "True",
group="v2_vnfm")
pm_job_4 = objects.PmJobV2(
id='pm_job_4',
authentication=pm_job_1_auth)
result = pm_job_utils._get_notification_auth_handle(pm_job_4)
res = type(result).__name__
name = type(http_client.BasicAuthHandle('test', 'test')).__name__
self.assertEqual(name, res)
pm_job_5 = objects.PmJobV2(
id='pm_job_5',
authentication=objects.SubscriptionAuthentication(
authType=["OAUTH2_CLIENT_CREDENTIALS"],
paramsOauth2ClientCredentials=(
objects.SubscriptionAuthentication_ParamsOauth2(
clientId='test',
clientPassword='test',
tokenEndpoint='http://127.0.0.1/token'
))
)
)
result = pm_job_utils._get_notification_auth_handle(pm_job_5)
res = type(result).__name__
name = type(http_client.OAuth2AuthHandle(
None, 'http://127.0.0.1/token', 'test', 'test')).__name__
self.assertEqual(name, res)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_test_notification(self, mock_do_request):
resp_no_auth = requests.Response()

41
tacker/tests/unit/sol_refactored/common/test_subscription_utils.py
View File

@ -15,6 +15,7 @@
import requests
from unittest import mock
from oslo_config import cfg
from oslo_utils import uuidutils
from tacker import context
@ -113,6 +114,46 @@ class TestSubscriptionUtils(base.BaseTestCase):
# execute oauth2 mtls
subsc_utils.send_notification(subsc_oauth2_mtls, notif_data_no_auth)
cfg.CONF.set_override("notification_verify_cert", "True",
group="v2_vnfm")
subsc_no_auth = objects.LccnSubscriptionV2(
id='sub-5', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback')
notif_data_no_auth = objects.VnfLcmOperationOccurrenceNotificationV2(
id=uuidutils.generate_uuid()
)
resp_no_auth = requests.Response()
resp_no_auth.status_code = 204
mock_resp.return_value = (resp_no_auth, None)
# execute no_auth
subsc_utils.send_notification(subsc_no_auth, notif_data_no_auth)
subsc_basic_auth = objects.LccnSubscriptionV2(
id='sub-6', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback',
authentication=objects.SubscriptionAuthentication(
authType=['BASIC'],
paramsBasic=objects.SubscriptionAuthentication_ParamsBasic(
userName='test', password='test')))
# execute basic_auth
subsc_utils.send_notification(subsc_basic_auth, notif_data_no_auth)
subsc_oauth2 = objects.LccnSubscriptionV2(
id='sub-7', verbosity='SHORT',
callbackUri='http://127.0.0.1/callback',
authentication=objects.SubscriptionAuthentication(
authType=['OAUTH2_CLIENT_CREDENTIALS'],
paramsOauth2ClientCredentials=(
objects.SubscriptionAuthentication_ParamsOauth2(
clientId='test', clientPassword='test',
tokenEndpoint='http://127.0.0.1/token'))))
# execute oauth2
subsc_utils.send_notification(subsc_oauth2, notif_data_no_auth)
@mock.patch.object(http_client.HttpClient, 'do_request')
def test_send_notification_error_code(self, mock_resp):
subsc_no_auth = objects.LccnSubscriptionV2(

91
tacker/tests/unit/sol_refactored/infra_drivers/openstack/test_openstack.py
View File

@ -3546,6 +3546,7 @@ class TestOpenstack(base.BaseTestCase):
self.driver = openstack.Openstack()
self.context = context.get_admin_context()
CONF.v2_vnfm.default_graceful_termination_timeout = 0
CONF.v2_vnfm.use_oauth2_mtls_for_heat = True
cur_dir = os.path.dirname(__file__)
sample_dir = os.path.join(cur_dir, "../..", "samples")
@ -4373,3 +4374,93 @@ class TestOpenstack(base.BaseTestCase):
self.assertEqual(expected_zone, use_zone_list[2]['VDU1-1'])
self.assertEqual(use_zone_list[2]['VDU1-1'],
use_zone_list[2]['VDU1-2'])
@mock.patch.object(openstack.heat_utils.HeatClient, 'get_stack_id')
@mock.patch.object(openstack.heat_utils.HeatClient, 'get_status')
@mock.patch.object(openstack.heat_utils.HeatClient, 'create_stack')
@mock.patch.object(openstack.heat_utils.HeatClient, 'update_stack')
@mock.patch.object(openstack.heat_utils.HeatClient, 'get_resources')
@mock.patch.object(openstack.heat_utils.HeatClient, 'get_parameters')
@mock.patch.object(openstack.heat_utils.HeatClient, 'get_template')
def test_https_with_instantiate(self, mock_template, mock_parameters,
mock_resources, mock_update_stack, mock_create_stack,
mock_status, mock_stack_id):
# prepare
req = objects.InstantiateVnfRequest.from_dict(_instantiate_req_example)
inst = objects.VnfInstanceV2(
# required fields
id=uuidutils.generate_uuid(),
vnfdId=SAMPLE_VNFD_ID,
vnfProvider='provider',
vnfProductName='product name',
vnfSoftwareVersion='software version',
vnfdVersion='vnfd version',
instantiationState='INSTANTIATED',
vimConnectionInfo=req.vimConnectionInfo
)
grant_req = objects.GrantRequestV1(
operation=fields.LcmOperationType.INSTANTIATE
)
grant = objects.GrantV1()
mock_resources.return_value = _heat_reses_example
mock_parameters.return_value = _heat_get_parameters_example
mock_template.return_value = _heat_get_template_example
mock_stack_id.return_value = None
# execute
CONF.v2_vnfm.use_oauth2_mtls_for_heat = True
CONF.v2_vnfm.heat_verify_cert = True
self.driver.instantiate(req, inst, grant_req, grant, self.vnfd_1)
mock_create_stack.assert_called_once()
mock_stack_id.return_value = STACK_ID
# execute
self.driver.instantiate(req, inst, grant_req, grant, self.vnfd_1)
mock_update_stack.assert_called_once()
@mock.patch.object(openstack.heat_utils.HeatClient, 'get_stack_id')
@mock.patch.object(openstack.heat_utils.HeatClient, 'get_status')
@mock.patch.object(openstack.heat_utils.HeatClient, 'create_stack')
@mock.patch.object(openstack.heat_utils.HeatClient, 'update_stack')
@mock.patch.object(openstack.heat_utils.HeatClient, 'get_resources')
@mock.patch.object(openstack.heat_utils.HeatClient, 'get_parameters')
@mock.patch.object(openstack.heat_utils.HeatClient, 'get_template')
def test_oauth2_mtls_with_instantiate(self, mock_template, mock_parameters,
mock_resources, mock_update_stack, mock_create_stack,
mock_status, mock_stack_id):
# prepare
req = objects.InstantiateVnfRequest.from_dict(_instantiate_req_example)
inst = objects.VnfInstanceV2(
# required fields
id=uuidutils.generate_uuid(),
vnfdId=SAMPLE_VNFD_ID,
vnfProvider='provider',
vnfProductName='product name',
vnfSoftwareVersion='software version',
vnfdVersion='vnfd version',
instantiationState='INSTANTIATED',
vimConnectionInfo=req.vimConnectionInfo
)
grant_req = objects.GrantRequestV1(
operation=fields.LcmOperationType.INSTANTIATE
)
grant = objects.GrantV1()
mock_resources.return_value = _heat_reses_example
mock_parameters.return_value = _heat_get_parameters_example
mock_template.return_value = _heat_get_template_example
mock_stack_id.return_value = None
CONF.v2_vnfm.use_oauth2_mtls_for_heat = False
CONF.v2_vnfm.heat_verify_cert = True
CONF.v2_vnfm.heat_mtls_ca_cert_file = '/path/to/cacert'
CONF.v2_vnfm.heat_mtls_client_cert_file = '/path/to/clientcert'
# execute
self.driver.instantiate(req, inst, grant_req, grant, self.vnfd_1)
mock_create_stack.assert_called_once()
mock_stack_id.return_value = STACK_ID
# execute
self.driver.instantiate(req, inst, grant_req, grant, self.vnfd_1)
mock_update_stack.assert_called_once()

24
tacker/tests/unit/sol_refactored/nfvo/test_nfvo_client.py
View File

@ -315,6 +315,13 @@ class TestNfvoClient(base.BaseTestCase):
self.nfvo_client.grant_api_version = '1.4.0'
self.nfvo_client.vnfpkgm_api_version = '2.1.0'
cfg.CONF.set_override("nfvo_verify_cert", True, group="v2_nfvo")
self.nfvo_client_https = nfvo_client.NfvoClient()
self.nfvo_client_https.endpoint = 'http://127.0.0.1:9990'
self.nfvo_client_https.client = http_client.HttpClient(auth_handle)
self.nfvo_client_https.grant_api_version = '1.4.0'
self.nfvo_client_https.vnfpkgm_api_version = '2.1.0'
cfg.CONF.set_override("use_external_nfvo", True, group="v2_nfvo")
cfg.CONF.set_override("mtls_ca_cert_file", "/path/to/cacert",
group="v2_nfvo")
@ -326,6 +333,17 @@ class TestNfvoClient(base.BaseTestCase):
self.addCleanup(mock.patch.stopall)
mock.patch('os.makedirs').start()
self.nfvo_client_mtls = nfvo_client.NfvoClient()
self.nfvo_client_mtls.endpoint = 'http://127.0.0.1:9990'
auth_handle_mtls = http_client.OAuth2MtlsAuthHandle(
self.nfvo_client.endpoint,
'http://127.0.0.1:9990/token',
'test',
'/path/to/cacert',
'/path/to/clientcert'
)
self.nfvo_client_mtls.client = http_client.HttpClient(auth_handle_mtls)
self.nfvo_client_mtls.grant_api_version = '1.4.0'
self.nfvo_client_mtls.vnfpkgm_api_version = '2.1.0'
@mock.patch.object(local_nfvo.LocalNfvo, 'onboarded_show')
@mock.patch.object(http_client.HttpClient, 'do_request')
@ -349,6 +367,12 @@ class TestNfvoClient(base.BaseTestCase):
self.context, SAMPLE_VNFD_ID)
self.assertEqual(SAMPLE_VNFD_ID, result.vnfdId)
# external nfvo oauth2 https
self.nfvo_client_https.is_local = False
result = self.nfvo_client_https.get_vnf_package_info_vnfd(
self.context, SAMPLE_VNFD_ID)
self.assertEqual(SAMPLE_VNFD_ID, result.vnfdId)
# external nfvo oauth2 mtls
cfg.CONF.set_override("mtls_client_cert_file", "/path/to/clientcert",
group="v2_nfvo")

6
tox.ini
View File

@ -102,6 +102,12 @@ setenv = {[testenv]setenv}
commands =
stestr --test-path=./tacker/tests/functional/sol_v2_az_retry run --slowest --concurrency 1 {posargs}
[testenv:dsvm-functional-sol-https-v2]
setenv = {[testenv]setenv}
commands =
stestr --test-path=./tacker/tests/functional/sol_https_v2 run --slowest --concurrency 1 {posargs}
[testenv:dsvm-compliance-sol-api]
passenv =
{[testenv]passenv}