--- /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
+
+
+class BaseSDNCKeywords(object):
+ """SDNC is an ONAP testing library for Robot Framework that provides functionality for interacting with the network
+ controller. """
+
+ def __init__(self):
+ super(BaseSDNCKeywords, self).__init__()
+ self.application_id = "robot-ete"
+ self.uuid = Utilities()
+ self.builtin = BuiltIn()
+
+ @keyword
+ def run_get_request(self, endpoint, data_path, accept="application/json", auth=None):
+ """Runs an SDNC 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 run_post_request(self, endpoint, data_path, data, accept="application/json", auth=None):
+ """Runs an SDNC post request"""
+ return self.post_request(endpoint, data_path, data, accept, auth)
+
+ @keyword
+ def run_put_request(self, endpoint, data_path, data, accept="application/json", auth=None):
+ """Runs an SDNC post request"""
+ return self.put_request(endpoint, data_path, data, accept, auth)
+
+ def get_request(self, endpoint, data_path, accept="application/json", auth=None):
+ """Runs an SDNC get request"""
+ logger.info("Creating session" + endpoint)
+ RequestsLibrary().create_session("sdnc", endpoint, auth=auth)
+ resp = RequestsLibrary().get_request("sdnc", data_path, headers=self.create_headers(accept))
+ logger.info("Received response from sdnc " + resp.text)
+ return resp
+
+ def create_headers(self, accept="application/json"):
+ """Create the headers that are used by sdnc"""
+ 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
+
+ def post_request(self, endpoint, data_path, data, accept="application/json", auth=None):
+ """Runs an sdnc post request"""
+ logger.info("Creating session" + endpoint)
+ RequestsLibrary().create_session("sdnc", endpoint, auth=auth)
+ resp = RequestsLibrary().post_request("sdnc", data_path, data=data, headers=self.create_headers(accept))
+ logger.info("Received response from sdnc " + resp.text)
+ return resp
+
+ def put_request(self, endpoint, data_path, data, accept="application/json", auth=None):
+ """Runs an sdnc post request"""
+ logger.info("Creating session" + endpoint)
+ RequestsLibrary().create_session("sdnc", endpoint, auth=auth)
+ resp = RequestsLibrary().put_request("sdnc", data_path, data=data, headers=self.create_headers(accept))
+ logger.info("Received response from sdnc " + resp.text)
+ return resp
+
+ def delete_request(self, endpoint, data_path, data, accept="application/json", auth=None):
+ """Runs an sdnc post request"""
+ logger.info("Creating session" + endpoint)
+ RequestsLibrary().create_session("sdnc", endpoint, auth=auth)
+ resp = RequestsLibrary().delete_request("sdnc", data_path, data=data, headers=self.create_headers(accept))
+ logger.info("Received response from sdnc " + resp.text)
+ return resp
--- /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.BaseSDNCKeywords import BaseSDNCKeywords
+from ONAPLibrary.UUIDKeywords import UUIDKeywords
+from ONAPLibrary.TemplatingKeywords import TemplatingKeywords
+from robot.api.deco import keyword
+from robot.libraries.BuiltIn import BuiltIn
+from robot.libraries.OperatingSystem import OperatingSystem
+
+
+class PreloadSDNCKeywords(object):
+ """SDNC is an ONAP testing library for Robot Framework that provides functionality for interacting with the network
+ controller. """
+
+ def __init__(self):
+ super(PreloadSDNCKeywords, self).__init__()
+ self.application_id = "robot-ete"
+ self.builtin = BuiltIn()
+ self.base_keywords = BaseSDNCKeywords()
+ self.utilities = UUIDKeywords()
+ self.os = OperatingSystem()
+ self.templating = TemplatingKeywords()
+
+ @keyword
+ def preload_vfmodule(self, endpoint, data_path, templates_folder, template, preload_dictionary):
+ """Runs an SDNC request to preload certain data."""
+ self.templating.create_environment("sdnc", templates_folder)
+ data = self.templating.apply_template("sdnc", template, preload_dictionary)
+ return self.base_keywords.post_request(endpoint, data_path, data)
--- /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.PreloadSDNCKeywords import PreloadSDNCKeywords
+from ONAPLibrary.robotlibcore import HybridCore
+from ONAPLibrary.BaseSDNCKeywords import BaseSDNCKeywords
+
+
+class SDNC(HybridCore):
+ """SDNC is an ONAP testing library for Robot Framework that provides functionality for interacting with the network
+ controller. """
+
+ def __init__(self):
+ self.keyword_implementors = [
+ BaseSDNCKeywords(),
+ PreloadSDNCKeywords()
+ ]
+ HybridCore.__init__(self, self.keyword_implementors)
import time
-from vcpeutils.preload import *
-from vcpeutils.vcpecommon import *
-
+from vcpeutils.csar_parser import *
+import requests
from robot.api import logger
from datetime import datetime
import urllib3
import sys
+from ONAPLibrary.PreloadSDNCKeywords import PreloadSDNCKeywords
class SoUtils:
self.region_name = None # set later
self.tenant_id = None # set later
self.logger = logger
- self.vcpecommon = VcpeCommon()
# SO urls, note: do NOT add a '/' at the end of the url
self.so_nbi_port = '8080'
self.aai_query_port = '8443'
self.aai_host = 'aai.onap'
+ # mr utls
+ self.mr_ip_addr = 'mr.onap'
+ self.mr_ip_port = '3904'
+
+ # sdnc urls
+ self.sdnc_ip_addr = 'sdnc.onap'
+ self.sdnc_preloading_port = '8282'
+ self.sdnc_endpoint = 'http://' + self.sdnc_ip_addr + ':' + self.sdnc_preloading_port
+ self.sdnc_preload_vnf_url = '/restconf/operations/VNF-API:preload-vnf-topology-operation'
# properties
self.homing_solution = 'sniro' # value is either 'sniro' or 'oof'
self.customer_location_used_by_oof = {
'vfmodule': 'vf'
}
+ # set the openstack cloud access credentials here
+ self.cloud = {
+ '--os-auth-url': 'http://10.12.25.2:5000',
+ '--os-username': 'kxi',
+ '--os-user-domain-id': 'default',
+ '--os-project-domain-id': 'default',
+ '--os-tenant-id': '09d8566ea45e43aa974cf447ed591d77',
+ '--os-region-name': 'RegionOne',
+ '--os-password': 'n3JhGMGuDzD8',
+ '--os-project-domain-name': 'Integration-SB-03',
+ '--os-identity-api-version': '3'
+ }
+
+ self.template_path = 'robot/assets/templates'
+ self.pub_key = 'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDKXDgoo3+WOqcUG8/5uUbk81+yczgwC4Y8ywTmuQqbNxlY1oQ0YxdMUqUnhitSXs5S/yRuAVOYHwGg2mCs20oAINrP+mxBI544AMIb9itPjCtgqtE2EWo6MmnFGbHB4Sx3XioE7F4VPsh7japsIwzOjbrQe+Mua1TGQ5d4nfEOQaaglXLLPFfuc7WbhbJbK6Q7rHqZfRcOwAMXgDoBqlyqKeiKwnumddo2RyNT8ljYmvB6buz7KnMinzo7qB0uktVT05FH9Rg0CTWH5norlG5qXgP2aukL0gk1ph8iAt7uYLf1ktp+LJI2gaF6L0/qli9EmVCSLr1uJ38Q8CBflhkh'
+ self.owning_entity_name = 'OE-Demonstration1'
+ self.project_name = 'Project-Demonstration'
+ self.owning_entity_id = '520cc603-a3c4-4ec2-9ef4-ca70facd79c0'
+ self.global_subscriber_id = 'Demonstration'
+ self.vgw_VfModuleModelInvariantUuid = '26d6a718-17b2-4ba8-8691-c44343b2ecd2'
+
def submit_create_req(self, req_json, req_type, service_instance_id=None, vnf_instance_id=None):
"""
POST {serverRoot}/serviceInstances/v4
def generate_service_request(self, instance_name, model):
req_details = {
'modelInfo': model,
- 'subscriberInfo': {'globalSubscriberId': self.vcpecommon.global_subscriber_id},
+ 'subscriberInfo': {'globalSubscriberId': self.global_subscriber_id},
'requestParameters': {
"userParams": [],
"subscriptionServiceType": "vCPE",
return {'requestDetails': req_details}
def add_project_info(self, req_details):
- req_details['project'] = {'projectName': self.vcpecommon.project_name}
+ req_details['project'] = {'projectName': self.project_name}
def add_owning_entity(self, req_details):
- req_details['owningEntity'] = {'owningEntityId': self.vcpecommon.owning_entity_id,
- 'owningEntityName': self.vcpecommon.owning_entity_name}
+ req_details['owningEntity'] = {'owningEntityId': self.owning_entity_id,
+ 'owningEntityName': self.owning_entity_name}
def generate_custom_service_request(self, instance_name, model, brg_mac):
brg_mac_enc = brg_mac.replace(':', '-')
req_details = {
'modelInfo': model,
'subscriberInfo': {'subscriberName': 'Kaneohe',
- 'globalSubscriberId': self.vcpecommon.global_subscriber_id},
+ 'globalSubscriberId': self.global_subscriber_id},
'cloudConfiguration': {"lcpCloudRegionId": self.region_name,
"tenantId": self.tenant_id},
'requestParameters': {
'name': 'VfModuleNames',
'value': [
{
- 'VfModuleModelInvariantUuid': self.vcpecommon.vgw_VfModuleModelInvariantUuid,
+ 'VfModuleModelInvariantUuid': self.vgw_VfModuleModelInvariantUuid,
'VfModuleName': 'VGW2BRG-{0}'.format(brg_mac_enc)
}
]
if not self.check_progress(req_id):
return None
- self.logger.info('Changing subnet name to ' + self.vcpecommon.network_name_to_subnet_name(network_name))
- self.vcpecommon.set_network_name(network_name)
+ self.logger.info('Changing subnet name to ' + self.network_name_to_subnet_name(network_name))
+ self.set_network_name(network_name)
subnet_name_changed = False
for i in range(20):
time.sleep(3)
- if self.vcpecommon.set_subnet_name(network_name):
+ if self.set_subnet_name(network_name):
subnet_name_changed = True
break
self.wait_for_aai('vnf', vnf_instance_id)
# SDNC Preload
+ preloader = PreloadSDNCKeywords()
+ vfmodule_name = '_'.join(['vf',
+ parser.vfmodule_models[0]['modelCustomizationName'].split('..')[0].lower(),
+ name_suffix])
+
+ extra_preload = {
+ 'pub_key': self.pub_key,
+ 'vnf_type': parser.vfmodule_models[0]['modelCustomizationName'],
+ 'generic_vnf_type': parser.vfmodule_models[0]['modelCustomizationName'],
+ 'service_type': svc_instance_id,
+ 'generic_vnf_name': vnf_model['modelCustomizationName'],
+ 'vnf_name': vfmodule_name,
+ 'mr_ip_addr': self.mr_ip_addr,
+ 'mr_ip_port': self.mr_ip_port,
+ 'sdnc_oam_ip': self.sdnc_ip_addr,
+ 'suffix': name_suffix,
+ 'oam_onap_net': 'oam_network_2No2',
+ 'oam_onap_subnet': 'oam_network_2No2',
+ 'public_net': 'external',
+ 'public_net_id': '971040b2-7059-49dc-b220-4fab50cb2ad4'
+ }
- preloader = Preload(self.vcpecommon)
- preloader.preload_vfmodule(vnf_template_file, svc_instance_id, parser.vnf_models[0], parser.vfmodule_models[0],
- preload_dict, name_suffix)
+ preload_dict.update(extra_preload)
+ preloader.preload_vfmodule(self.sdnc_endpoint, self.sdnc_preload_vnf_url, self.template_path, vnf_template_file,
+ preload_dict)
# create VF Module
if len(parser.vfmodule_models) == 1:
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
+
+ @staticmethod
+ def network_name_to_subnet_name(network_name):
+ """
+ :param network_name: example: vcpe_net_cpe_signal_201711281221
+ :return: vcpe_net_cpe_signal_subnet_201711281221
+ """
+ fields = network_name.split('_')
+ fields.insert(-1, 'subnet')
+ return '_'.join(fields)
+
+ def set_network_name(self, network_name):
+ param = ' '.join([k + ' ' + v for k, v in list(self.cloud.items())])
+ openstackcmd = 'openstack ' + param
+ cmd = ' '.join([openstackcmd, 'network set --name', network_name, 'ONAP-NW1'])
+ os.popen(cmd)
+
+ def set_subnet_name(self, network_name):
+ """
+ Example: network_name = vcpe_net_cpe_signal_201711281221
+ set subnet name to vcpe_net_cpe_signal_subnet_201711281221
+ :return:
+ """
+ param = ' '.join([k + ' ' + v for k, v in list(self.cloud.items())])
+ openstackcmd = 'openstack ' + param
+
+ # expected results: | subnets | subnet_id |
+ subnet_info = os.popen(openstackcmd + ' network show ' + network_name + ' |grep subnets').read().split('|')
+ if len(subnet_info) > 2 and subnet_info[1].strip() == 'subnets':
+ subnet_id = subnet_info[2].strip()
+ subnet_name = self.network_name_to_subnet_name(network_name)
+ cmd = ' '.join([openstackcmd, 'subnet set --name', subnet_name, subnet_id])
+ os.popen(cmd)
+ self.logger.info("Subnet name set to: " + subnet_name)
+ return True
+ else:
+ self.logger.error("Can't get subnet info from network name: " + network_name)
+ return False
+++ /dev/null
-#! /usr/bin/python
-
-from vcpeutils.csar_parser import *
-from robot.api import logger
-from past import builtins
-import requests
-
-
-class Preload:
- def __init__(self, vcpecommon):
- 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:
- logger.error('Error! Cannot find a value to replace ' + sz)
- return sz
-
- def generate_json(self, template_file, replace_dict):
- with open(template_file) as json_input:
- json_data = json.load(json_input)
- stk = [json_data]
- while len(stk) > 0:
- data = stk.pop()
- for k, v in list(data.items()):
- if type(v) is dict:
- stk.append(v)
- elif type(v) is list:
- stk.extend(v)
- elif type(v) is builtins.basestring:
- if self.vcpecommon.template_variable_symbol in v:
- data[k] = self.replace(v, replace_dict)
- else:
- logger.warn('Unexpected line in template: {}. Look for value {}'.format(template_file, v))
- return json_data
-
- def preload_network(self, template_file, network_role, subnet_start_ip, subnet_gateway, common_dict, name_suffix):
- """
- :param template_file:
- :param network_role: cpe_signal, cpe_public, brg_bng, bng_mux, mux_gw
- :param subnet_start_ip:
- :param subnet_gateway:
- :param common_dict:
- :param name_suffix: e.g. '201711201311'
- :return:
- """
- network_name = '_'.join([self.vcpecommon.instance_name_prefix['network'], network_role.lower(), name_suffix])
- subnet_name = self.vcpecommon.network_name_to_subnet_name(network_name)
- common_dict['${' + network_role+'_net}'] = network_name
- common_dict['${' + network_role+'_subnet}'] = subnet_name
- replace_dict = {'${network_role}': network_role,
- '${service_type}': 'vCPE',
- '${network_type}': 'Generic NeutronNet',
- '${network_name}': network_name,
- '${subnet_start_ip}': subnet_start_ip,
- '${subnet_gateway}': subnet_gateway
- }
- 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):
- logger.debug(json.dumps(replace_dict, indent=4, sort_keys=True))
- json_data = self.generate_json(template_file, replace_dict)
- 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:
- logger.debug(json.dumps(response, indent=4, sort_keys=True))
- logger.error('Preloading failed.')
- return False
- return True
-
- def preload_vgw(self, template_file, brg_mac, commont_dict, name_suffix):
- replace_dict = {'${brg_mac}': brg_mac,
- '${suffix}': name_suffix
- }
- replace_dict.update(commont_dict)
- 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):
- replace_dict = {'${brg_mac}': brg_mac,
- '${suffix}': name_suffix,
- '${vgw_vfmod_name_index}': vgw_vfmod_name_index
- }
- replace_dict.update(commont_dict)
- 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):
- """
- :param template_file:
- :param service_instance_id:
- :param vnf_model: parsing results from csar_parser
- :param vfmodule_model: parsing results from csar_parser
- :param common_dict:
- :param name_suffix:
- :return:
- """
-
- # examples:
- # vfmodule_model['modelCustomizationName']: "Vspinfra111601..base_vcpe_infra..module-0",
- # vnf_model['modelCustomizationName']: "vspinfra111601 0",
-
- vfmodule_name = '_'.join([self.vcpecommon.instance_name_prefix['vfmodule'],
- vfmodule_model['modelCustomizationName'].split('..')[0].lower(), name_suffix])
-
- # vnf_type and generic_vnf_type are identical
- replace_dict = {'${vnf_type}': vfmodule_model['modelCustomizationName'],
- '${generic_vnf_type}': vfmodule_model['modelCustomizationName'],
- '${service_type}': service_instance_id,
- '${generic_vnf_name}': vnf_model['modelCustomizationName'],
- '${vnf_name}': vfmodule_name,
- '${mr_ip_addr}': self.vcpecommon.mr_ip_addr,
- '${mr_ip_port}': self.vcpecommon.mr_ip_port,
- '${sdnc_oam_ip}': self.vcpecommon.sdnc_oam_ip,
- '${suffix}': name_suffix}
- replace_dict.update(common_dict)
- logger.info('Preloading VF Module ' + vfmodule_name)
- return self.preload(template_file, replace_dict, self.vcpecommon.sdnc_preload_vnf_url)
+++ /dev/null
-import logging
-import os
-
-
-class VcpeCommon:
- # set the openstack cloud access credentials here
- cloud = {
- '--os-auth-url': 'http://10.12.25.2:5000',
- '--os-username': 'kxi',
- '--os-user-domain-id': 'default',
- '--os-project-domain-id': 'default',
- '--os-tenant-id': '09d8566ea45e43aa974cf447ed591d77',
- '--os-region-name': 'RegionOne',
- '--os-password': 'n3JhGMGuDzD8',
- '--os-project-domain-name': 'Integration-SB-03',
- '--os-identity-api-version': '3'
- }
-
- #############################################################################################
-
- template_variable_symbol = '${'
- cpe_vm_prefix = 'zdcpe'
-
- dcae_ves_collector_name = 'dcae-bootstrap'
- global_subscriber_id = 'Demonstration'
- project_name = 'Project-Demonstration'
- owning_entity_id = '520cc603-a3c4-4ec2-9ef4-ca70facd79c0'
- owning_entity_name = 'OE-Demonstration1'
-
- def __init__(self, extra_host_names=None):
- rootlogger = logging.getLogger()
- handler = logging.StreamHandler()
- formatter = logging.Formatter('%(asctime)s %(levelname)s %(name)s.%(funcName)s(): %(message)s')
- handler.setFormatter(formatter)
- rootlogger.addHandler(handler)
- rootlogger.setLevel(logging.INFO)
-
- self.logger = logging.getLogger(__name__)
- self.logger.propagate = False
- self.logger.addHandler(handler)
- 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
- self.vgw_VfModuleModelInvariantUuid = '26d6a718-17b2-4ba8-8691-c44343b2ecd2'
- self.sdnc_preloading_port = '8282'
-
- 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 = {'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']
-
- #############################################################################################
- # SDNC urls
- self.sdnc_db_name = 'sdnctl'
- self.sdnc_db_user = 'sdnctl'
- self.sdnc_db_pass = 'gamma'
- self.sdnc_db_port = '32774'
- self.sdnc_headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
- self.sdnc_preload_network_url = 'http://' + self.hosts['sdnc'] + \
- ':' + self.sdnc_preloading_port + '/restconf/operations/VNF-API:preload-network-topology-operation'
- self.sdnc_preload_vnf_url = 'http://' + self.hosts['sdnc'] + \
- ':' + self.sdnc_preloading_port + '/restconf/operations/VNF-API:preload-vnf-topology-operation'
- self.sdnc_preload_gra_url = 'http://' + self.hosts['sdnc'] + \
- ':' + self.sdnc_preloading_port + '/restconf/operations/GENERIC-RESOURCE-API:preload-vf-module-topology-operation'
- self.sdnc_ar_cleanup_url = 'http://' + self.hosts['sdnc'] + ':' + self.sdnc_preloading_port + \
- '/restconf/config/GENERIC-RESOURCE-API:'
-
- 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'
-
- @staticmethod
- def network_name_to_subnet_name(network_name):
- """
- :param network_name: example: vcpe_net_cpe_signal_201711281221
- :return: vcpe_net_cpe_signal_subnet_201711281221
- """
- fields = network_name.split('_')
- fields.insert(-1, 'subnet')
- return '_'.join(fields)
-
- def set_network_name(self, network_name):
- param = ' '.join([k + ' ' + v for k, v in list(self.cloud.items())])
- openstackcmd = 'openstack ' + param
- cmd = ' '.join([openstackcmd, 'network set --name', network_name, 'ONAP-NW1'])
- os.popen(cmd)
-
- def set_subnet_name(self, network_name):
- """
- Example: network_name = vcpe_net_cpe_signal_201711281221
- set subnet name to vcpe_net_cpe_signal_subnet_201711281221
- :return:
- """
- param = ' '.join([k + ' ' + v for k, v in list(self.cloud.items())])
- openstackcmd = 'openstack ' + param
-
- # expected results: | subnets | subnet_id |
- subnet_info = os.popen(openstackcmd + ' network show ' + network_name + ' |grep subnets').read().split('|')
- if len(subnet_info) > 2 and subnet_info[1].strip() == 'subnets':
- subnet_id = subnet_info[2].strip()
- subnet_name = self.network_name_to_subnet_name(network_name)
- cmd = ' '.join([openstackcmd, 'subnet set --name', subnet_name, subnet_id])
- os.popen(cmd)
- self.logger.info("Subnet name set to: " + subnet_name)
- return True
- else:
- self.logger.error("Can't get subnet info from network name: " + network_name)
- return False
Code Review / testsuite / python-testing-utils.git / commitdiff