--- /dev/null
+from robot.api.deco import keyword
+import base64
+
+
+class Base64Keywords(object):
+ """ Utilities useful for generating UUIDs """
+
+ def __init__(self):
+ super(Base64Keywords, self).__init__()
+
+ @keyword
+ def base64_encode(self, string_to_encode):
+ """generate a base64 encoded string"""
+ return base64.b64encode(self, string_to_encode)
+
+ @keyword
+ def base64_decode(self, string_to_decode):
+ """decode a base64 encoded string"""
+ return base64.b64decode(self, string_to_decode)
--- /dev/null
+# Copyright 2019 AT&T Intellectual Property. 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.
+
+from ONAPLibrary.SNIROKeywords import SNIROKeywords
+from ONAPLibrary.robotlibcore import HybridCore
+
+
+class OOF(HybridCore):
+ """OOF is an ONAP testing library for Robot Framework that provides functionality for interacting with the
+ optimiztion framework. """
+
+ def __init__(self):
+ self.keyword_implementors = [
+ SNIROKeywords()
+ ]
+ HybridCore.__init__(self, self.keyword_implementors)
--- /dev/null
+# Copyright 2019 AT&T Intellectual Property. 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.
+from RequestsLibrary import RequestsLibrary
+from robot.api import logger
+from robot.api.deco import keyword
+from robot.libraries.BuiltIn import BuiltIn
+
+from ONAPLibrary.Utilities import Utilities
+from ONAPLibrary.TemplatingKeywords import TemplatingKeywords
+from ONAPLibrary.Base64Keywords import Base64Keywords
+
+
+class SNIROKeywords(object):
+ """OOF is an ONAP testing library for Robot Framework that provides functionality for interacting with the
+ optimiztion framework. """
+
+ def __init__(self):
+ super(SNIROKeywords, self).__init__()
+ self.application_id = "robot-ete"
+ self.uuid = Utilities()
+ self.templating = TemplatingKeywords()
+ self.base64 = Base64Keywords()
+ self.builtin = BuiltIn()
+
+ @keyword
+ def run_sniro_get_request(self, endpoint, data_path, accept="application/json", auth=None):
+ """Runs OOF-SNIRO Get request"""
+ resp = self.get_request(endpoint, data_path, accept, auth)
+ self.builtin.should_be_equal_as_strings(resp.status_code, "200")
+ return resp
+
+ @keyword
+ def reset_sniro(self, endpoint):
+ logger.debug('Clearing SNIRO data')
+ resp = self.post_request(endpoint, '/reset', None)
+ self.builtin.should_be_equal_as_strings(resp.status_code, "200", 'Clearing SNIRO date failed.')
+
+ @keyword
+ def preload_sniro(self, endpoint, template_directory, template_sniro_data, template_sniro_request,
+ tunnelxconn_ar_name, vgw_name, vbrg_ar_name, vgmux_svc_instance_uuid, vbrg_svc_instance_uuid):
+ self.templating.create_environment("sniro", template_directory)
+ logger.info('Preloading SNIRO for homing service')
+ replace_dict = {'tunnelxconn_ar_name': tunnelxconn_ar_name,
+ 'vgw_name': vgw_name,
+ 'brg_ar_name': vbrg_ar_name,
+ 'vgmux_svc_instance_uuid': vgmux_svc_instance_uuid,
+ 'vbrg_svc_instance_uuid': vbrg_svc_instance_uuid
+ }
+ sniro_data = self.templating.apply_template("sniro", template_sniro_data, replace_dict)
+ base64_sniro_data = self.base64.base64_encode(sniro_data)
+ replace_dict = {'base64_sniro_data': base64_sniro_data}
+ sniro_request = self.templating.apply_template("sniro", template_sniro_request, replace_dict)
+ resp = self.post_request(endpoint, '/', sniro_request)
+ self.builtin.should_be_equal_as_strings(resp.status_code, "200", 'SNIRO preloading failed.')
+ return True
+
+ def post_request(self, endpoint, data_path, data, accept="application/json", auth=None):
+ """Runs an SNIRO post request"""
+ logger.info("Creating session" + endpoint)
+ RequestsLibrary().create_session("so", endpoint, auth=auth)
+ resp = RequestsLibrary().post_request("so", data_path, data=data, headers=self.create_headers(accept))
+ logger.info("Received response from so " + resp.text)
+ return resp
+
+ def get_request(self, endpoint, data_path, accept="application/json", auth=None):
+ """Runs an SNIRO get request"""
+ logger.info("Creating session" + endpoint)
+ RequestsLibrary().create_session("sniro", endpoint, auth=auth)
+ resp = RequestsLibrary().get_request("sniro", data_path, headers=self.create_headers(accept))
+ logger.info("Received response from OOF-SNIRO " + resp.text)
+ return resp
+
+ def create_headers(self, accept="application/json"):
+ """Create the headers that are used by so"""
+ uuid = self.uuid.generate_uuid4()
+ headers = {
+ "Accept": accept,
+ "Content-Type": "application/json",
+ "X-TransactionId": self.application_id + "-" + uuid,
+ "X-FromAppId": self.application_id
+ }
+ return headers
from ONAPLibrary.SocketKeywords import SocketKeywords
from ONAPLibrary.UUIDKeywords import UUIDKeywords
from ONAPLibrary.HTTPKeywords import HTTPKeywords
+from ONAPLibrary.Base64Keywords import Base64Keywords
class Utilities(HybridCore):
- """ DNS Keywords are useful for DNS requests """
+ """ Keywords are useful for helper functions requests """
def __init__(self):
self.keyword_implementors = [
DNSKeywords(),
SocketKeywords(),
UUIDKeywords(),
- HTTPKeywords()
+ HTTPKeywords(),
+ Base64Keywords()
]
HybridCore.__init__(self, self.keyword_implementors)
from vcpeutils.vcpecommon import *
from robot.api import logger
+from datetime import datetime
+import urllib3
+import sys
class SoUtils:
self.so_db_pass = 'password'
self.so_db_port = '30252'
+ # aai urls
+ self.aai_userpass = 'AAI', 'AAI'
+ self.aai_query_port = '8443'
+ self.aai_host = 'aai.onap'
+
+ # properties
+ self.homing_solution = 'sniro' # value is either 'sniro' or 'oof'
+ self.customer_location_used_by_oof = {
+ "customerLatitude": "32.897480",
+ "customerLongitude": "-97.040443",
+ "customerName": "some_company"
+ }
+ self.product_family_id = 'f9457e8c-4afd-45da-9389-46acd9bf5116'
+ self.custom_product_family_id = 'a9a77d5a-123e-4ca2-9eb9-0b015d2ee0fb'
+ self.instance_name_prefix = {
+ 'service': 'svc',
+ 'network': 'net',
+ 'vnf': 'vnf',
+ 'vfmodule': 'vf'
+ }
+
def submit_create_req(self, req_json, req_type, service_instance_id=None, vnf_instance_id=None):
"""
POST {serverRoot}/serviceInstances/v4
'requestParameters': {"userParams": []},
'platform': {"platformName": "Platform-Demonstration"}
}
- self.add_req_info(req_details, instance_name, self.vcpecommon.product_family_id)
+ self.add_req_info(req_details, instance_name, self.product_family_id)
self.add_related_instance(req_details, service_instance_id, service_model)
return {'requestDetails': req_details}
"tenantId": self.tenant_id},
'requestParameters': {"usePreload": 'true'}
}
- self.add_req_info(req_details, instance_name, self.vcpecommon.product_family_id)
+ self.add_req_info(req_details, instance_name, self.product_family_id)
self.add_related_instance(req_details, service_instance_id, service_model)
self.add_related_instance(req_details, vnf_instance_id, vnf_model)
return {'requestDetails': req_details}
},
{
"name": "Customer_Location",
- "value": self.vcpecommon.customer_location_used_by_oof
+ "value": self.customer_location_used_by_oof
},
{
"name": "Homing_Solution",
- "value": self.vcpecommon.homing_solution
+ "value": self.homing_solution
}
],
"subscriptionServiceType": "vCPE",
'aLaCarte': 'false'
}
}
- self.add_req_info(req_details, instance_name, self.vcpecommon.custom_product_family_id)
+ self.add_req_info(req_details, instance_name, self.custom_product_family_id)
self.add_project_info(req_details)
self.add_owning_entity(req_details)
return {'requestDetails': req_details}
name_suffix = '_' + datetime.now().strftime('%Y%m%d%H%M')
# create service
- instance_name = '_'.join([self.vcpecommon.instance_name_prefix['service'],
+ instance_name = '_'.join([self.instance_name_prefix['service'],
parser.svc_model['modelName'][0:10], name_suffix])
instance_name = instance_name.lower()
req = self.generate_custom_service_request(instance_name, parser.svc_model, brg_mac)
self.logger.info('Waiting for AAI traversal to complete...')
for i in range(30):
time.sleep(1)
- if self.vcpecommon.is_node_in_aai(node_type, uuid):
+ if self.is_node_in_aai(node_type, uuid):
return
self.logger.error("AAI traversal didn't finish in 30 seconds. Something is wrong. Type {0}, UUID {1}".format(
# Set Global timestamp for instancenames
global_timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
# create service
- instance_name = '_'.join([self.vcpecommon.instance_name_prefix['service'],
+ instance_name = '_'.join([self.instance_name_prefix['service'],
parser.svc_model['modelName'], global_timestamp, name_suffix])
instance_name = instance_name.lower()
instance_name = instance_name.replace(' ', '')
# create networks
for model in parser.net_models:
base_name = model['modelCustomizationName'].lower().replace('mux_vg', 'mux_gw')
- network_name = '_'.join([self.vcpecommon.instance_name_prefix['network'], base_name, name_suffix])
+ network_name = '_'.join([self.instance_name_prefix['network'], base_name, name_suffix])
network_name = network_name.lower()
self.logger.info('Creating network: ' + network_name)
req = self.generate_vnf_or_network_request(network_name, model, svc_instance_id, parser.svc_model)
# create VNF
if len(parser.vnf_models) == 1:
vnf_model = parser.vnf_models[0]
- vnf_instance_name = '_'.join([self.vcpecommon.instance_name_prefix['vnf'],
+ vnf_instance_name = '_'.join([self.instance_name_prefix['vnf'],
vnf_model['modelCustomizationName'].split(' ')[0], name_suffix])
vnf_instance_name = vnf_instance_name.lower()
vnf_instance_name = vnf_instance_name.replace(' ', '')
sys.exit()
model = parser.vfmodule_models[0]
- vfmodule_instance_name = '_'.join([self.vcpecommon.instance_name_prefix['vfmodule'],
+ vfmodule_instance_name = '_'.join([self.instance_name_prefix['vfmodule'],
model['modelCustomizationName'].split('..')[0], name_suffix])
vfmodule_instance_name = vfmodule_instance_name.lower()
vfmodule_instance_name = vfmodule_instance_name.replace(' ', '')
return None
return svc_instance_id
+
+ def is_node_in_aai(self, node_type, node_uuid):
+ key = None
+ search_node_type = None
+ if node_type == 'service':
+ search_node_type = 'service-instance'
+ key = 'service-instance-id'
+ elif node_type == 'vnf':
+ search_node_type = 'generic-vnf'
+ key = 'vnf-id'
+ else:
+ logging.error('Invalid node_type: ' + node_type)
+ sys.exit()
+
+ url = 'https://{0}:{1}/aai/v11/search/nodes-query?search-node-type={2}&filter={3}:EQUALS:{4}'.format(
+ self.aai_host, self.aai_query_port, search_node_type, key, node_uuid)
+
+ headers = {'Content-Type': 'application/json', 'Accept': 'application/json', 'X-FromAppID': 'vCPE-Robot', 'X-TransactionId': 'get_aai_subscr'}
+ urllib3.disable_warnings()
+ r = requests.get(url, headers=headers, auth=self.aai_userpass, verify=False)
+ response = r.json()
+ self.logger.debug('aai query: ' + url)
+ self.logger.debug('aai response:\n' + json.dumps(response, indent=4, sort_keys=True))
+ return 'result-data' in response
#! /usr/bin/python
-from datetime import datetime
-from vcpeutils.vcpecommon import *
from vcpeutils.csar_parser import *
from robot.api import logger
from past import builtins
-import base64
+import requests
class Preload:
def __init__(self, vcpecommon):
- self.logger = logger
self.vcpecommon = vcpecommon
def replace(self, sz, replace_dict):
for old_string, new_string in list(replace_dict.items()):
sz = sz.replace(old_string, new_string)
if self.vcpecommon.template_variable_symbol in sz:
- self.logger.error('Error! Cannot find a value to replace ' + sz)
+ logger.error('Error! Cannot find a value to replace ' + sz)
return sz
def generate_json(self, template_file, replace_dict):
if self.vcpecommon.template_variable_symbol in v:
data[k] = self.replace(v, replace_dict)
else:
- self.logger.warn('Unexpected line in template: {}. Look for value {}'.format(template_file, v))
+ logger.warn('Unexpected line in template: {}. Look for value {}'.format(template_file, v))
return json_data
- def reset_sniro(self):
- self.logger.debug('Clearing SNIRO data')
- r = requests.post(self.vcpecommon.sniro_url + '/reset', headers=self.vcpecommon.sniro_headers)
- if 2 != r.status_code / 100:
- self.logger.debug(r.content)
- self.logger.error('Clearing SNIRO date failed.')
- sys.exit()
-
- def preload_sniro(self, template_sniro_data, template_sniro_request, tunnelxconn_ar_name, vgw_name, vbrg_ar_name,
- vgmux_svc_instance_uuid, vbrg_svc_instance_uuid):
- self.reset_sniro()
- self.logger.info('Preloading SNIRO for homing service')
- replace_dict = {'${tunnelxconn_ar_name}': tunnelxconn_ar_name,
- '${vgw_name}': vgw_name,
- '${brg_ar_name}': vbrg_ar_name,
- '${vgmux_svc_instance_uuid}': vgmux_svc_instance_uuid,
- '${vbrg_svc_instance_uuid}': vbrg_svc_instance_uuid
- }
- sniro_data = self.generate_json(template_sniro_data, replace_dict)
- self.logger.debug('SNIRO data:')
- self.logger.debug(json.dumps(sniro_data, indent=4, sort_keys=True))
-
- base64_sniro_data = base64.b64encode(json.dumps(sniro_data))
- self.logger.debug('SNIRO data: 64')
- self.logger.debug(base64_sniro_data)
- replace_dict = {'${base64_sniro_data}': base64_sniro_data, '${sniro_ip}': self.vcpecommon.hosts['robot']}
- sniro_request = self.generate_json(template_sniro_request, replace_dict)
- self.logger.debug('SNIRO request:')
- self.logger.debug(json.dumps(sniro_request, indent=4, sort_keys=True))
-
- r = requests.post(self.vcpecommon.sniro_url, headers=self.vcpecommon.sniro_headers, json=sniro_request)
- if 2 != r.status_code / 100:
- response = r.json()
- self.logger.debug(json.dumps(response, indent=4, sort_keys=True))
- self.logger.error('SNIRO preloading failed.')
- sys.exit()
-
- return True
-
def preload_network(self, template_file, network_role, subnet_start_ip, subnet_gateway, common_dict, name_suffix):
"""
:param template_file:
'${subnet_start_ip}': subnet_start_ip,
'${subnet_gateway}': subnet_gateway
}
- self.logger.info('Preloading network ' + network_role)
+ logger.info('Preloading network ' + network_role)
return self.preload(template_file, replace_dict, self.vcpecommon.sdnc_preload_network_url)
def preload(self, template_file, replace_dict, url):
- self.logger.debug(json.dumps(replace_dict, indent=4, sort_keys=True))
+ logger.debug(json.dumps(replace_dict, indent=4, sort_keys=True))
json_data = self.generate_json(template_file, replace_dict)
- self.logger.debug(json.dumps(json_data, indent=4, sort_keys=True))
+ logger.debug(json.dumps(json_data, indent=4, sort_keys=True))
r = requests.post(url, headers=self.vcpecommon.sdnc_headers, auth=self.vcpecommon.sdnc_userpass, json=json_data)
response = r.json()
if int(response.get('output', {}).get('response-code', 0)) != 200:
- self.logger.debug(json.dumps(response, indent=4, sort_keys=True))
- self.logger.error('Preloading failed.')
+ logger.debug(json.dumps(response, indent=4, sort_keys=True))
+ logger.error('Preloading failed.')
return False
return True
'${suffix}': name_suffix
}
replace_dict.update(commont_dict)
- self.logger.info('Preloading vGW')
+ logger.info('Preloading vGW')
return self.preload(template_file, replace_dict, self.vcpecommon.sdnc_preload_vnf_url)
def preload_vgw_gra(self, template_file, brg_mac, commont_dict, name_suffix, vgw_vfmod_name_index):
'${vgw_vfmod_name_index}': vgw_vfmod_name_index
}
replace_dict.update(commont_dict)
- self.logger.info('Preloading vGW-GRA')
+ logger.info('Preloading vGW-GRA')
return self.preload(template_file, replace_dict, self.vcpecommon.sdnc_preload_gra_url)
def preload_vfmodule(self, template_file, service_instance_id, vnf_model, vfmodule_model, common_dict, name_suffix):
'${sdnc_oam_ip}': self.vcpecommon.sdnc_oam_ip,
'${suffix}': name_suffix}
replace_dict.update(common_dict)
- self.logger.info('Preloading VF Module ' + vfmodule_name)
+ logger.info('Preloading VF Module ' + vfmodule_name)
return self.preload(template_file, replace_dict, self.vcpecommon.sdnc_preload_vnf_url)
-
- def preload_all_networks(self, template_file, name_suffix):
- common_dict = {'${' + k + '}': v for k, v in list(self.vcpecommon.common_preload_config.items())}
- for network, v in list(self.vcpecommon.preload_network_config.items()):
- subnet_start_ip, subnet_gateway_ip = v
- if not self.preload_network(template_file, network, subnet_start_ip, subnet_gateway_ip,
- common_dict, name_suffix):
- return None
- return common_dict
-
- def test(self):
- # this is for testing purpose
- name_suffix = datetime.now().strftime('%Y%m%d%H%M')
- vcpecommon = VcpeCommon()
- preloader = Preload(vcpecommon)
-
- network_dict = {'${' + k + '}': v for k, v in list(self.vcpecommon.common_preload_config.items())}
- template_file = 'preload_templates/template.network.json'
- for k, v in list(self.vcpecommon.preload_network_config.items()):
- if not preloader.preload_network(template_file, k, v[0], v[1], network_dict, name_suffix):
- break
-
- print('---------------------------------------------------------------')
- print('Network related replacement dictionary:')
- print(json.dumps(network_dict, indent=4, sort_keys=True))
- print('---------------------------------------------------------------')
-
- keys = ['infra', 'bng', 'gmux', 'brg']
- for key in keys:
- csar_file = self.vcpecommon.find_file(key, 'csar', 'csar')
- template_file = self.vcpecommon.find_file(key, 'json', 'preload_templates')
- if csar_file and template_file:
- parser = CsarParser()
- parser.parse_csar(csar_file)
- service_instance_id = 'test112233'
- preloader.preload_vfmodule(template_file, service_instance_id, parser.vnf_models[0],
- parser.vfmodule_models[0], network_dict, name_suffix)
-
- def test_sniro(self):
- template_sniro_data = self.vcpecommon.find_file('sniro_data', 'json', 'preload_templates')
- template_sniro_request = self.vcpecommon.find_file('sniro_request', 'json', 'preload_templates')
-
- vcperescust_csar = self.vcpecommon.find_file('rescust', 'csar', 'csar')
- parser = CsarParser()
- parser.parse_csar(vcperescust_csar)
- tunnelxconn_ar_name = None
- brg_ar_name = None
- vgw_name = None
- for model in parser.vnf_models:
- if 'tunnel' in model['modelCustomizationName']:
- tunnelxconn_ar_name = model['modelCustomizationName']
- elif 'brg' in model['modelCustomizationName']:
- brg_ar_name = model['modelCustomizationName']
- elif 'vgw' in model['modelCustomizationName']:
- vgw_name = model['modelCustomizationName']
-
- if not (tunnelxconn_ar_name and brg_ar_name and vgw_name):
- self.logger.error('Cannot find all names from %s.', vcperescust_csar)
- sys.exit()
-
- vgmux_svc_instance_uuid = '88888888888888'
- vbrg_svc_instance_uuid = '999999999999999'
-
- self.preload_sniro(template_sniro_data, template_sniro_request, tunnelxconn_ar_name, vgw_name, brg_ar_name,
- vgmux_svc_instance_uuid, vbrg_svc_instance_uuid)
-import json
import logging
import os
-import sys
-import requests
class VcpeCommon:
- #############################################################################################
- # Start: configurations that you must change for a new ONAP installation
- external_net_addr = '10.12.0.0'
- external_net_prefix_len = 16
- #############################################################################################
# set the openstack cloud access credentials here
-
cloud = {
'--os-auth-url': 'http://10.12.25.2:5000',
'--os-username': 'kxi',
'--os-identity-api-version': '3'
}
- common_preload_config = {
- 'oam_onap_net': 'oam_network_2No2',
- 'oam_onap_subnet': 'oam_network_2No2',
- 'public_net': 'external',
- 'public_net_id': '971040b2-7059-49dc-b220-4fab50cb2ad4'
- }
- sdnc_controller_pod = 'dev-sdnc-sdnc-0'
-
#############################################################################################
template_variable_symbol = '${'
cpe_vm_prefix = 'zdcpe'
- #############################################################################################
- # preloading network config
- # key=network role
- # value = [subnet_start_ip, subnet_gateway_ip]
- preload_network_config = {
- 'cpe_public': ['10.2.0.2', '10.2.0.1'],
- 'cpe_signal': ['10.4.0.2', '10.4.0.1'],
- 'brg_bng': ['10.3.0.2', '10.3.0.1'],
- 'bng_mux': ['10.1.0.10', '10.1.0.1'],
- 'mux_gw': ['10.5.0.10', '10.5.0.1']
- }
dcae_ves_collector_name = 'dcae-bootstrap'
global_subscriber_id = 'Demonstration'
self.logger.setLevel(logging.DEBUG)
self.logger.info('Initializing configuration')
- # CHANGEME: vgw_VfModuleModelInvariantUuid is in rescust service csar, look in service-VcpesvcRescust1118-template.yml for groups vgw module metadata. TODO: read this value automcatically
+ # CHANGEME: vgw_VfModuleModelInvariantUuid is in rescust service csar,
+ # look in service-VcpesvcRescust1118-template.yml for groups vgw module metadata.
+ # TODO: read this value automcatically
self.vgw_VfModuleModelInvariantUuid = '26d6a718-17b2-4ba8-8691-c44343b2ecd2'
- # CHANGEME: OOM: this is the address that the brg and bng will nat for sdnc access - 10.0.0.x address of k8 host for sdnc-0 container
- #self.sdnc_oam_ip = self.get_pod_node_oam_ip('sdnc-sdnc-0')
- self.sdnc_oam_ip = 'sdnc.onap'
- # CHANGEME: OOM: this is a k8s host external IP, e.g. oom-k8s-01 IP
- #self.oom_so_sdnc_aai_ip = self.get_pod_node_public_ip('sdnc-sdnc-0')
- #self.oom_so_sdnc_aai_ip = self.get_pod_node_public_ip('sdnc-sdnc-0')
- # CHANGEME: OOM: this is a k8s host external IP, e.g. oom-k8s-01 IP
- #self.oom_dcae_ves_collector = self.oom_so_sdnc_aai_ip
- # CHANGEME: OOM: this is a k8s host external IP, e.g. oom-k8s-01 IP
- #self.mr_ip_addr = self.oom_so_sdnc_aai_ip
- self.mr_ip_addr = 'mr.onap'
- #self.mr_ip_port = '30227'
- self.mr_ip_port = '3904'
- #self.so_nbi_port = '30277' if self.oom_mode else '8080'
-
- #self.sdnc_preloading_port = '30202' if self.oom_mode else '8282'
self.sdnc_preloading_port = '8282'
- #self.aai_query_port = '30233' if self.oom_mode else '8443'
- self.aai_query_port = '8443'
- #self.sniro_port = '30288' if self.oom_mode else '8080'
- self.sniro_port = '8080'
self.host_names = ['so', 'sdnc', 'robot', 'aai', self.dcae_ves_collector_name]
if extra_host_names:
self.host_names.extend(extra_host_names)
# get IP addresses
- #self.hosts = self.get_vm_ip(self.host_names, self.external_net_addr, self.external_net_prefix_len)
- self.hosts = { 'so': 'so.onap', 'sdnc': 'sdnc.onap', 'robot': 'robot.onap', 'aai': 'aai.onap' }
- # this is the keyword used to name vgw stack, must not be used in other stacks
- self.vgw_name_keyword = 'base_vcpe_vgw'
- # this is the file that will keep the index of last assigned SO name
- self.vgw_vfmod_name_index_file= '__var/vgw_vfmod_name_index'
- self.svc_instance_uuid_file = '__var/svc_instance_uuid'
- self.preload_dict_file = '__var/preload_dict'
- self.vgmux_vnf_name_file = '__var/vgmux_vnf_name'
- self.product_family_id = 'f9457e8c-4afd-45da-9389-46acd9bf5116'
- self.custom_product_family_id = 'a9a77d5a-123e-4ca2-9eb9-0b015d2ee0fb'
- self.instance_name_prefix = {
- 'service': 'svc',
- 'network': 'net',
- 'vnf': 'vnf',
- 'vfmodule': 'vf'
- }
- self.aai_userpass = 'AAI', 'AAI'
- self.pub_key = 'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDKXDgoo3+WOqcUG8/5uUbk81+yczgwC4Y8ywTmuQqbNxlY1oQ0YxdMUqUnhitSXs5S/yRuAVOYHwGg2mCs20oAINrP+mxBI544AMIb9itPjCtgqtE2EWo6MmnFGbHB4Sx3XioE7F4VPsh7japsIwzOjbrQe+Mua1TGQ5d4nfEOQaaglXLLPFfuc7WbhbJbK6Q7rHqZfRcOwAMXgDoBqlyqKeiKwnumddo2RyNT8ljYmvB6buz7KnMinzo7qB0uktVT05FH9Rg0CTWH5norlG5qXgP2aukL0gk1ph8iAt7uYLf1ktp+LJI2gaF6L0/qli9EmVCSLr1uJ38Q8CBflhkh'
+ self.hosts = {'so': 'so.onap', 'sdnc': 'sdnc.onap', 'robot': 'robot.onap', 'aai': 'aai.onap'}
self.os_tenant_id = self.cloud['--os-tenant-id']
self.os_region_name = self.cloud['--os-region-name']
- self.common_preload_config['pub_key'] = self.pub_key
- self.sniro_url = 'http://' + self.hosts['robot'] + ':' + self.sniro_port + '/__admin/mappings'
- self.sniro_headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
- self.homing_solution = 'sniro' # value is either 'sniro' or 'oof'
-# self.homing_solution = 'oof'
- self.customer_location_used_by_oof = {
- "customerLatitude": "32.897480",
- "customerLongitude": "-97.040443",
- "customerName": "some_company"
- }
#############################################################################################
# SDNC urls
- self.sdnc_userpass = 'admin', 'Kp8bJ4SXszM0WXlhak3eHlcse2gAw84vaoGGmJvUy2U'
self.sdnc_db_name = 'sdnctl'
self.sdnc_db_user = 'sdnctl'
self.sdnc_db_pass = 'gamma'
self.vpp_inf_url = 'http://{0}:8183/restconf/config/ietf-interfaces:interfaces'
self.vpp_api_headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
self.vpp_api_userpass = ('admin', 'admin')
- self.vpp_ves_url= 'http://{0}:8183/restconf/config/vesagent:vesagent'
-
-
-
- def find_file(self, file_name_keyword, file_ext, search_dir):
- """
- :param file_name_keyword: keyword used to look for the csar file, case insensitive matching, e.g, infra
- :param file_ext: e.g., csar, json
- :param search_dir path to search
- :return: path name of the file
- """
- file_name_keyword = file_name_keyword.lower()
- file_ext = file_ext.lower()
- if not file_ext.startswith('.'):
- file_ext = '.' + file_ext
-
- filenamepath = None
- for file_name in os.listdir(search_dir):
- file_name_lower = file_name.lower()
- if file_name_keyword in file_name_lower and file_name_lower.endswith(file_ext):
- if filenamepath:
- self.logger.error('Multiple files found for *{0}*.{1} in '
- 'directory {2}'.format(file_name_keyword, file_ext, search_dir))
- sys.exit()
- filenamepath = os.path.abspath(os.path.join(search_dir, file_name))
-
- if filenamepath:
- return filenamepath
- else:
- self.logger.error("Cannot find *{0}*{1} in directory {2}".format(file_name_keyword, file_ext, search_dir))
- sys.exit()
+ self.vpp_ves_url = 'http://{0}:8183/restconf/config/vesagent:vesagent'
@staticmethod
def network_name_to_subnet_name(network_name):
else:
self.logger.error("Can't get subnet info from network name: " + network_name)
return False
-
- def is_node_in_aai(self, node_type, node_uuid):
- key = None
- search_node_type = None
- if node_type == 'service':
- search_node_type = 'service-instance'
- key = 'service-instance-id'
- elif node_type == 'vnf':
- search_node_type = 'generic-vnf'
- key = 'vnf-id'
- else:
- logging.error('Invalid node_type: ' + node_type)
- sys.exit()
-
- url = 'https://{0}:{1}/aai/v11/search/nodes-query?search-node-type={2}&filter={3}:EQUALS:{4}'.format(
- self.hosts['aai'], self.aai_query_port, search_node_type, key, node_uuid)
-
- headers = {'Content-Type': 'application/json', 'Accept': 'application/json', 'X-FromAppID': 'vCPE-Robot', 'X-TransactionId': 'get_aai_subscr'}
- requests.packages.urllib3.disable_warnings()
- r = requests.get(url, headers=headers, auth=self.aai_userpass, verify=False)
- response = r.json()
- self.logger.debug('aai query: ' + url)
- self.logger.debug('aai response:\n' + json.dumps(response, indent=4, sort_keys=True))
- return 'result-data' in response
Code Review / testsuite / python-testing-utils.git / commitdiff