-.tox/*
-python_openecomp_eteutils.egg-info/*
-# Created by https://www.gitignore.io/api/node,sonar,maven,eclipse,sonarqube,intellij+all
+# Created by https://www.gitignore.io/api/sonar,python,eclipse,sonarqube,pycharm+all,intellij+all
+# Edit at https://www.gitignore.io/?templates=sonar,python,eclipse,sonarqube,pycharm+all,intellij+all
### Eclipse ###
-
.metadata
bin/
tmp/
# Android studio 3.1+ serialized cache file
.idea/caches/build_file_checksums.ser
+# JetBrains templates
+**___jb_tmp___
+
### Intellij+all Patch ###
# Ignores the whole .idea folder and all .iml files
# See https://github.com/joeblau/gitignore.io/issues/186 and https://github.com/joeblau/gitignore.io/issues/360
.idea/misc.xml
*.ipr
-### Maven ###
-target/
-pom.xml.tag
-pom.xml.releaseBackup
-pom.xml.versionsBackup
-pom.xml.next
-release.properties
-dependency-reduced-pom.xml
-buildNumber.properties
-.mvn/timing.properties
-.mvn/wrapper/maven-wrapper.jar
-
-### Node ###
-# Logs
-logs
-*.log
-npm-debug.log*
-yarn-debug.log*
-yarn-error.log*
+# Sonarlint plugin
+.idea/sonarlint
-# Runtime data
-pids
-*.pid
-*.seed
-*.pid.lock
+### PyCharm+all ###
+# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio and WebStorm
+# Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839
-# Directory for instrumented libs generated by jscoverage/JSCover
-lib-cov
+# User-specific stuff
-# Coverage directory used by tools like istanbul
-coverage
+# Generated files
-# nyc test coverage
-.nyc_output
+# Sensitive or high-churn files
-# Grunt intermediate storage (https://gruntjs.com/creating-plugins#storing-task-files)
-.grunt
+# Gradle
-# Bower dependency directory (https://bower.io/)
-bower_components
+# Gradle and Maven with auto-import
+# When using Gradle or Maven with auto-import, you should exclude module files,
+# since they will be recreated, and may cause churn. Uncomment if using
+# auto-import.
+# .idea/modules.xml
+# .idea/*.iml
+# .idea/modules
-# node-waf configuration
-.lock-wscript
+# CMake
-# Compiled binary addons (https://nodejs.org/api/addons.html)
-build/Release
+# Mongo Explorer plugin
-# Dependency directories
-node_modules/
-jspm_packages/
+# File-based project format
-# TypeScript v1 declaration files
-typings/
+# IntelliJ
-# Optional npm cache directory
-.npm
+# mpeltonen/sbt-idea plugin
-# Optional eslint cache
-.eslintcache
+# JIRA plugin
-# Optional REPL history
-.node_repl_history
+# Cursive Clojure plugin
-# Output of 'npm pack'
-*.tgz
+# Crashlytics plugin (for Android Studio and IntelliJ)
-# Yarn Integrity file
-.yarn-integrity
+# Editor-based Rest Client
+
+# Android studio 3.1+ serialized cache file
+
+# JetBrains templates
+
+### PyCharm+all Patch ###
+# Ignores the whole .idea folder and all .iml files
+# See https://github.com/joeblau/gitignore.io/issues/186 and https://github.com/joeblau/gitignore.io/issues/360
+
+
+# Reason: https://github.com/joeblau/gitignore.io/issues/186#issuecomment-249601023
-# dotenv environment variables file
-.env
-# parcel-bundler cache (https://parceljs.org/)
+# Sonarlint plugin
+
+### Python ###
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+pip-wheel-metadata/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+# Usually these files are written by a python script from a template
+# before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
.cache
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+.pytest_cache/
-# next.js build output
-.next
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
-# nuxt.js build output
-.nuxt
+# Flask stuff:
+instance/
+.webassets-cache
-# vuepress build output
-.vuepress/dist
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
-# Serverless directories
-.serverless
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+.python-version
+
+# pipenv
+# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+# However, in case of collaboration, if having platform-specific dependencies or dependencies
+# having no cross-platform support, pipenv may install dependencies that don’t work, or not
+# install all needed dependencies.
+#Pipfile.lock
+
+# celery beat schedule file
+celerybeat-schedule
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
### Sonar ###
#Sonar generated dir
# SonarLint working directories, configuration files (including credentials)
.sonarlint/
-
-# End of https://www.gitignore.io/api/node,sonar,maven,eclipse,sonarqube,intellij+all
-.flattened-pom.xml
\ No newline at end of file
+# End of https://www.gitignore.io/api/sonar,python,eclipse,sonarqube,pycharm+all,intellij+all
def dns_request(self, domain, ns):
""" return the ip address of the given domain name from the given nameserver """
- request = dns.message.make_query(domain, dns.rdatatype.A);
- request.flags |= dns.flags.AD;
+ request = dns.message.make_query(domain, dns.rdatatype.A)
+ request.flags |= dns.flags.AD
request.find_rrset(request.additional, dns.name.root, 65535, dns.rdatatype.OPT, create=True, force_unique=True)
response = dns.query.udp(request, ns)
for answer in response.answer:
for item in answer.items:
- return item
\ No newline at end of file
+ return item
self.folder= ''
self.keyfile = keyfile
self.local_gather_data_sh = shell
- print "EteGatherDataListener instantiated"
+ print("EteGatherDataListener instantiated")
def end_test(self, name, attrs):
if attrs['status'] == 'PASS':
if (self.folder != ''):
return
self.folder = os.path.dirname(path)
- print(self.folder)
+ print((self.folder))
def close(self):
- print "EteGatherDataListener tests failed=" + str(self.tests_failed)
+ print("EteGatherDataListener tests failed=" + str(self.tests_failed))
if (self.tests_failed > 0):
self.gather_debug_data()
def gather_debug_data(self):
- for application in self.APPLICATIONS.keys():
+ for application in list(self.APPLICATIONS.keys()):
self.gather_application_data(application, self.APPLICATIONS.get(application))
def gather_application_data(self, application, ip):
import json
import yaml
-import StringIO
+import io
import copy
from hashlib import md5
from paramiko import RSAKey
from paramiko.ssh_exception import PasswordRequiredException
+
class HEATUtils:
""" Utilities useful for constructing OpenStack HEAT requests """
def get_yaml(self, template_file):
"""Template Yaml To Json reads a YAML Heat template file returns a JSON string that can be used included in an Openstack Add Stack Request"""
- if isinstance(template_file, basestring):
+ if isinstance(template_file, str):
fin = open(template_file, 'r')
yamlobj = yaml.load(fin)
return yamlobj
return None
-
+
def template_yaml_to_json(self, template_file):
"""Template Yaml To Json reads a YAML Heat template file returns a JSON string that can be used included in an Openstack Add Stack Request"""
- if isinstance(template_file, basestring):
+ if isinstance(template_file, str):
fin = open(template_file, 'r')
yamlobj = yaml.load(fin)
fin.close()
if 'heat_template_version' in yamlobj:
datetime = yamlobj['heat_template_version']
yamlobj['heat_template_version'] = str(datetime)
- fout = StringIO.StringIO()
+ fout = io.BytesIO()
json.dump(yamlobj, fout)
contents = fout.getvalue()
fout.close()
return contents
-
+
def env_yaml_to_json(self, template_file):
"""Env Yaml To JSon reads a YAML Heat env file and returns a JSON string that can be used included in an Openstack Add Stack Request"""
- if isinstance(template_file, basestring):
+ if isinstance(template_file, str):
fin = open(template_file, 'r')
yamlobj = yaml.load(fin)
fin.close()
if 'parameters' in yamlobj:
- fout = StringIO.StringIO()
+ fout = io.BytesIO()
json.dump(yamlobj['parameters'], fout)
contents = fout.getvalue()
fout.close()
return contents
return None
-
+
def stack_info_parse(self, stack_info):
""" returns a flattened version of the Openstack Find Stack results """
d = {}
d['id'] = s['id']
d['name'] = s['stack_name']
d['stack_status'] = s['stack_status']
- return d
-
-
+ return d
+
def match_fingerprint(self, pvt_file, pw, fingerprint):
try:
sshKey = RSAKey.from_private_key_file(pvt_file, pw)
keybytes = md5(sshKey.asbytes()).hexdigest()
- printableFingerprint = ':'.join(a+b for a,b in zip(keybytes[::2], keybytes[1::2]))
+ printableFingerprint = ':'.join(a + b for a, b in zip(keybytes[::2], keybytes[1::2]))
return printableFingerprint == fingerprint.__str__()
except PasswordRequiredException:
- return False
-
+ return False
+
def match_private_key_file_to_keypair(self, files, keypair):
for keyfile in files:
if (self.match_fingerprint(keyfile, None, keypair['keypair']['fingerprint'])):
return keyfile
return None
-
+
def get_openstack_server_ip(self, server, network_name="public", ipversion=4):
ipaddr = None
try:
+from six.moves.urllib.parse import urlparse
+
import urllib
import urllib3
-import urlparse
+
class HTTPUtils:
"""HTTPUtils is common resource for simple http helper keywords."""
def json_equals(self, left, right):
"""JSON Equals takes in two strings or json objects, converts them into json if needed and then compares them, returning if they are equal or not."""
- if isinstance(left, basestring):
+ if isinstance(left, str):
left_json = json.loads(left);
else:
left_json = left;
- if isinstance(right, basestring):
+ if isinstance(right, str):
right_json = json.loads(right);
else:
right_json = right;
def get_openstack_token(self, alias):
"""Get Openstack auth token from the current alias"""
response = self._cache.switch(alias)
- if isinstance(response, basestring):
+ if isinstance(response, str):
jsonResponse = json.loads(response);
else:
jsonResponse = response;
def get_openstack_catalog(self, alias):
"""Get Openstack service catalog from the current alias"""
response = self._cache.switch(alias)
- if isinstance(response, basestring):
+ if isinstance(response, str):
jsonResponse = json.loads(response);
else:
jsonResponse = response;
def get_current_openstack_tenant(self, alias):
"""Get Openstack tenant from the current alias"""
response = self._cache.switch(alias)
- if isinstance(response, basestring):
+ if isinstance(response, str):
jsonResponse = json.loads(response);
else:
jsonResponse = response;
def get_openstack_regions(self, alias):
"""Get all Openstack regions from the current alias"""
response = self._cache.switch(alias)
- if isinstance(response, basestring):
+ if isinstance(response, str):
jsonResponse = json.loads(response);
else:
jsonResponse = response;
def get_openstack_service_url(self, alias, servicetype, region = None, tenant_id = None):
"""Get Openstack service catalog from the current alias"""
response = self._cache.switch(alias)
- if isinstance(response, basestring):
+ if isinstance(response, str):
jsonResponse = json.loads(response);
else:
jsonResponse = response;
# Only provide tenant id when authorizing without qualifying with tenant id
# WindRiver does not return the tenantId on the endpoint in this case.
if tenant_id is not None:
- listOfEndpoints[:] = [y for y in listOfEndpoints if self.__determine_match(y['tenantId'], tenant_id)];
+ listOfEndpoints[:] = [y for y in listOfEndpoints if self.__determine_match(y['tenantId'], tenant_id)];
if jsonResponse['keystone_api_version'] == 'v3':
listOfEndpoints[:] = [z for z in listOfEndpoints if self.__determine_match(z['interface'], 'public')];
if len(listOfEndpoints) > 0:
-import json
from string import Template
class StringTemplater:
def apply_template(self, template_location, values):
"""returns a string that is the jinja template in template_location filled in via the dictionary in values """
- print
template = self.jinja_env.get_template(template_location)
- return template.render(values)
\ No newline at end of file
+ return template.render(values)
def generate_MilliTimestamp_UUID(self):
"""generate a millisecond timestamp uuid"""
then = datetime.datetime.now()
- return int(time.mktime(then.timetuple())*1e3 + then.microsecond/1e3)
\ No newline at end of file
+ return int(time.mktime(then.timetuple())*1e3 + then.microsecond/1e3)
env = dict(os.environ, SOAKSUBFOLDER=self.soaksubfolder + "/")
output = subprocess.check_output(["bash", self.robot_command, self.robot_test, self.test_number], shell=False, env=env)
logging.info("{} ({}) {}".format(self.getName(), self.robot_test, output))
- except Exception, e:
+ except Exception as e:
logging.error("{} ({}) Unexpected error {}".format(self.getName(), self.robot_test, repr(e)))
logging.info("{} ({}) ended - {}".format(self.getName(), self.robot_test, str(datetime.now())))
controller = TestController(opts)
controller.execute()
- except Exception, e:
+ except Exception as e:
indent = len(program_name) * " "
sys.stderr.write(program_name + ": " + repr(e) + "\n")
sys.stderr.write(indent + " for help use --help")
if __name__ == "__main__":
if DEBUG:
- print "debug"
+ print("debug")
if TESTRUN:
import doctest
doctest.testmod()
# This flag says that the code is written to work on both Python 2 and Python
# 3. If at all possible, it is good practice to do this. If you cannot, you
# will need to generate wheels for each Python version that you support.
-universal=0
\ No newline at end of file
+universal=1
\ No newline at end of file
setup(
name='robotframework-onap', # This is the name of your PyPI-package.
- keywords=("utils", "robotframework", "testing", "onap"),
+ keywords=["utils", "robotframework", "testing", "onap"],
version='0.4', # Update the version number for new releases
license="Apache 2.0",
description='Scripts written to be used during robot framework testing', # Info about script
'paramiko',
'pyyaml',
'robotframework',
- 'deepdiff>=2.5,<3.3',
- 'Jinja2'
+ 'deepdiff',
+ 'Jinja2',
+ 'urllib3',
+ 'six',
+ 'requests'
], # what we need
packages=['eteutils', 'loadtest', 'vcpeutils'], # The name of your scripts package
package_dir={'eteutils': 'eteutils', 'loadtest': 'loadtest', 'vcpeutils':'vcpeutils'}, # The location of your scipts package
'Development Status :: 4 - Beta',
'Intended Audience :: Developers',
'Programming Language :: Python :: 2.7',
+ 'Programming Language :: Python :: 3.7',
'Environment :: Plugins',
'Framework :: Robot Framework',
'Framework :: Robot Framework :: Library',
# and then run "tox" from this directory.
[tox]
-envlist = py27
+envlist = py27, py37
[testenv]
commands = {envpython} setup.py test
#! /usr/bin/python
-import sys
-import logging
-import requests
-import json
-from datetime import datetime
-#import progressbar
import time
-import csar_parser
-import preload
-from vcpecommon import *
+
+from .csar_parser import *
+from .preload import *
+from .vcpecommon import *
from robot.api import logger
return {'requestDetails': req_details}
def create_custom_service(self, csar_file, brg_mac, name_suffix=None):
- parser = csar_parser.CsarParser()
+ parser = CsarParser()
if not parser.parse_csar(csar_file):
return False
self.tenant_id=tenant_id
self.logger.info('\n----------------------------------------------------------------------------------')
self.logger.info('Start to create entire service defined in csar: {0}'.format(csar_file))
- parser = csar_parser.CsarParser()
+ parser = CsarParser()
self.logger.info('Parsing csar ...')
if not parser.parse_csar(csar_file):
self.logger.error('Cannot parse csar: {0}'.format(csar_file))
# SDNC Preload
- preloader = preload.Preload(self.vcpecommon)
+ 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)
},
"""
node_dic = svc_template['topology_template']['node_templates']
- for node_name, v in node_dic.items():
+ for node_name, v in list(node_dic.items()):
model = {
'modelInvariantId': v['metadata']['invariantUUID'],
'modelVersionId': v['metadata']['UUID'],
},
"""
node_dic = svc_template['topology_template']['groups']
- for node_name, v in node_dic.items():
+ for node_name, v in list(node_dic.items()):
if v['type'].startswith('org.openecomp.groups.VfModule'):
model = {
'modelType': 'vfModule',
def print_models(self):
print('---------Service Model----------')
- print(json.dumps(self.svc_model, indent=2, sort_keys=True))
+ print((json.dumps(self.svc_model, indent=2, sort_keys=True)))
print('---------Network Model(s)----------')
for model in self.net_models:
- print(json.dumps(model, indent=2, sort_keys=True))
+ print((json.dumps(model, indent=2, sort_keys=True)))
print('---------VNF Model(s)----------')
for model in self.vnf_models:
- print(json.dumps(model, indent=2, sort_keys=True))
+ print((json.dumps(model, indent=2, sort_keys=True)))
print('---------VF Module Model(s)----------')
for model in self.vfmodule_models:
- print(json.dumps(model, indent=2, sort_keys=True))
+ print((json.dumps(model, indent=2, sort_keys=True)))
def test(self):
self.parse_csar('csar/service-Vcpesvcinfra111601-csar.csar')
#! /usr/bin/python
-import requests
-import json
-import sys
from datetime import datetime
-from vcpecommon import *
-import csar_parser
-#import logging
+from .vcpecommon import *
+from .csar_parser import *
from robot.api import logger
import base64
self.vcpecommon = vcpecommon
def replace(self, sz, replace_dict):
- for old_string, new_string in replace_dict.items():
+ 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)
stk = [json_data]
while len(stk) > 0:
data = stk.pop()
- for k, v in data.items():
+ 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 str or type(v) is unicode:
+ elif type(v) is str or type(v) is str:
if self.vcpecommon.template_variable_symbol in v:
data[k] = self.replace(v, replace_dict)
else:
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 self.vcpecommon.common_preload_config.items()}
- for network, v in self.vcpecommon.preload_network_config.items():
+ 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):
vcpecommon = VcpeCommon()
preloader = Preload(vcpecommon)
- network_dict = {'${' + k + '}': v for k, v in self.vcpecommon.common_preload_config.items()}
+ 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 self.vcpecommon.preload_network_config.items():
+ 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((json.dumps(network_dict, indent=4, sort_keys=True)))
print('---------------------------------------------------------------')
keys = ['infra', 'bng', 'gmux', 'brg']
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 = csar_parser.CsarParser()
+ parser = CsarParser()
parser.parse_csar(csar_file)
service_instance_id = 'test112233'
preloader.preload_vfmodule(template_file, service_instance_id, parser.vnf_models[0],
template_sniro_request = self.vcpecommon.find_file('sniro_request', 'json', 'preload_templates')
vcperescust_csar = self.vcpecommon.find_file('rescust', 'csar', 'csar')
- parser = csar_parser.CsarParser()
+ parser = CsarParser()
parser.parse_csar(vcperescust_csar)
tunnelxconn_ar_name = None
brg_ar_name = None
import ipaddress
import requests
-import commands
-import time
-from novaclient import client as openstackclient
-from netaddr import IPAddress, IPNetwork
class VcpeCommon:
#############################################################################################
return '_'.join(fields)
def set_network_name(self, network_name):
- param = ' '.join([k + ' ' + v for k, v in self.cloud.items()])
+ 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)
set subnet name to vcpe_net_cpe_signal_subnet_201711281221
:return:
"""
- param = ' '.join([k + ' ' + v for k, v in self.cloud.items()])
+ param = ' '.join([k + ' ' + v for k, v in list(self.cloud.items())])
openstackcmd = 'openstack ' + param
# expected results: | subnets | subnet_id |
:param sz: a string
:return: the first IP address matching the network, e.g. 10.5.12.3
"""
- network = ipaddress.ip_network(unicode('{0}/{1}'.format(net_addr, net_addr_len)), strict=False)
+ network = ipaddress.ip_network(str('{0}/{1}'.format(net_addr, net_addr_len)), strict=False)
ip_list = re.findall(r'[0-9]+(?:\.[0-9]+){3}', sz)
for ip in ip_list:
- this_net = ipaddress.ip_network(unicode('{0}/{1}'.format(ip, net_addr_len)), strict=False)
+ this_net = ipaddress.ip_network(str('{0}/{1}'.format(ip, net_addr_len)), strict=False)
if this_net == network:
return str(ip)
return None
Code Review / testsuite / python-testing-utils.git / commitdiff