--- /dev/null
+.tox/*
+python_openecomp_eteutils.egg-info/*
--- /dev/null
+ECOMP and OpenECOMP are trademarks and service marks of AT&T Intellectual Property.
+
+LICENSE.TXT file can be appended as follows:
+/*
+ * ============LICENSE_START==========================================
+ * ===================================================================
+ * Copyright © 2017 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.
+ * ============LICENSE_END============================================
+ *
+ * ECOMP and OpenECOMP are trademarks
+ * and service marks of AT&T Intellectual Property.
+ *
+ */
\ No newline at end of file
--- /dev/null
+OpenECOMP ETE Utils
+=======================
+
+
+
+Scripts written to be used during ete testing
+
+to install locally, checkout this repo and then run 'pip install -e .' in the root
\ No newline at end of file
--- /dev/null
+import dns.message
+import dns.name
+import dns.query
+
+class DNSUtils:
+ """ Utilities useful for DNS requests """
+
+ 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.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
--- /dev/null
+import json
+import yaml
+import StringIO
+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):
+ 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):
+ 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()
+ 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):
+ fin = open(template_file, 'r')
+ yamlobj = yaml.load(fin)
+ fin.close()
+ if 'parameters' in yamlobj:
+ fout = StringIO.StringIO()
+ 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 = {}
+ if isinstance(stack_info, dict):
+ s = stack_info['stack']
+ p = s['parameters']
+ d = copy.deepcopy(p)
+ d['id'] = s['id']
+ d['name'] = s['stack_name']
+ d['stack_status'] = s['stack_status']
+ 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]))
+ return printableFingerprint == fingerprint.__str__()
+ except PasswordRequiredException:
+ 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:
+ versions = server['addresses'][network_name]
+ for version in versions:
+ if version['version'] == ipversion:
+ ipaddr = version['addr']
+ break;
+ except ValueError:
+ return ipaddr
+ return ipaddr
\ No newline at end of file
--- /dev/null
+import urllib
+
+class HTTPUtils:
+ """HTTPUtils is common resource for simple http helper keywords."""
+
+ def url_encode_string(self, barestring):
+ """URL Encode String takes in a string and converts into 'percent-encoded' string"""
+ return urllib.quote_plus(barestring)
\ No newline at end of file
--- /dev/null
+import json
+
+from deepdiff import DeepDiff
+
+class JSONUtils:
+ """JSONUtils is common resource for simple json 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):
+ left_json = json.loads(left);
+ else:
+ left_json = left;
+ if isinstance(right, basestring):
+ right_json = json.loads(right);
+ else:
+ right_json = right;
+
+ ddiff = DeepDiff(left_json, right_json, ignore_order=True);
+ if ddiff == {}:
+ return True;
+ else:
+ return False;
+
+ def make_list_into_dict(self, listOfDicts, key):
+ """ Converts a list of dicts that contains a field that has a unique key into a dict of dicts """
+ d = {}
+ if isinstance(listOfDicts, list):
+ for thisDict in listOfDicts:
+ v = thisDict[key]
+ d[v] = thisDict
+ return d
+
+ def find_element_in_array(self, searchedArray, key, value):
+ """ Takes in an array and a key value, it will return the items in the array that has a key and value that matches what you pass in """
+ elements = [];
+ for item in searchedArray:
+ if key in item:
+ if item[key] == value:
+ elements.append(item);
+ return elements;
\ No newline at end of file
--- /dev/null
+from sys import platform
+
+class OSUtils:
+ """ Utilities useful for constructing OpenStack HEAT requests """
+
+ def get_normalized_os(self):
+ os = platform
+ if platform == "linux" or platform == "linux2":
+ os = 'linux64'
+ elif platform == "darwin":
+ os = 'mac64'
+ elif platform == "win32":
+ os = platform
+ return os
--- /dev/null
+from robot.libraries.BuiltIn import BuiltIn
+import robot.utils
+import json
+
+class OpenstackLibrary:
+ """OpenstackLibrary manages the connection state and service catalog of an openstack instance."""
+
+ ROBOT_LIBRARY_SCOPE = 'Global'
+
+
+ def __init__(self):
+ self._cache = robot.utils.ConnectionCache('No connections created')
+ self.builtin = BuiltIn()
+
+ def save_openstack_auth(self, alias, response):
+ """Save Openstack Auth takes in an openstack auth response and saves it to allow easy retrival of token and service catalog"""
+ self.builtin.log('Creating connection: %s' % alias, 'DEBUG')
+ self._cache.register(response, alias=alias)
+
+ def get_openstack_token(self, alias):
+ """Get Openstack auth token from the current alias"""
+ response = self._cache.switch(alias)
+ if isinstance(response, basestring):
+ jsonResponse = json.loads(response);
+ else:
+ jsonResponse = response;
+ return jsonResponse['access']['token']['id']
+
+ def get_openstack_catalog(self, alias):
+ """Get Openstack service catalog from the current alias"""
+ response = self._cache.switch(alias)
+ if isinstance(response, basestring):
+ jsonResponse = json.loads(response);
+ else:
+ jsonResponse = response;
+ return jsonResponse['access']['serviceCatalog']
+
+ def get_current_openstack_tenant(self, alias):
+ """Get Openstack tenant from the current alias"""
+ response = self._cache.switch(alias)
+ if isinstance(response, basestring):
+ jsonResponse = json.loads(response);
+ else:
+ jsonResponse = response;
+ return jsonResponse['access']['token']['tenant']
+
+ def get_current_openstack_tenant_id(self, alias):
+ """Get Openstack tenant id from the current alias"""
+ tenant = self.get_current_openstack_tenant(alias);
+ return tenant['id']
+
+ def get_openstack_regions(self, alias):
+ """Get all Openstack regions from the current alias"""
+ response = self._cache.switch(alias)
+ if isinstance(response, basestring):
+ jsonResponse = json.loads(response);
+ else:
+ jsonResponse = response;
+ regions = [];
+ for catalogEntry in jsonResponse['access']['serviceCatalog']:
+ listOfEndpoints = catalogEntry['endpoints'];
+ for endpoint in listOfEndpoints:
+ if 'region'in endpoint:
+ if endpoint['region'] not in regions:
+ regions.append(endpoint['region'])
+ return regions;
+
+ 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):
+ jsonResponse = json.loads(response);
+ else:
+ jsonResponse = response;
+ endPoint = None;
+ for catalogEntry in jsonResponse['access']['serviceCatalog']:
+ if self.__determine_match(catalogEntry['type'], servicetype):
+ listOfEndpoints = catalogEntry['endpoints'];
+ # filter out non matching regions if provided
+ listOfEndpoints[:] = [x for x in listOfEndpoints if self.__determine_match(x['region'], region)];
+ # filter out non matching tenants if provided
+ listOfEndpoints[:] = [y for y in listOfEndpoints if self.__determine_match(y['tenantId'], tenant_id)];
+ if len(listOfEndpoints) > 0:
+ endPoint = listOfEndpoints[0]['publicURL'];
+ if endPoint == None:
+ self.builtin.should_not_be_empty("", "Service Endpoint Url should not be empty")
+ return endPoint;
+
+ def __determine_match(self, listItem, item):
+ if item is None:
+ return True;
+ elif listItem == item:
+ return True;
+ else:
+ return False;
\ No newline at end of file
--- /dev/null
+
+class RequestsClientCert:
+ """RequestsClientCert allows adding a client cert to the Requests Robot Library."""
+
+ def add_client_cert(self, session, cert):
+ """Add Client Cert takes in a requests session object and a string path to the cert"""
+ session.cert = cert
\ No newline at end of file
--- /dev/null
+from string import Template
+
+class StringTemplater:
+ """StringTemplater is common resource for templating with strings."""
+
+ def template_string(self, template, values):
+ """Template String takes in a string and its values and converts it using the string.Template class"""
+ return Template(template).substitute(values)
\ No newline at end of file
--- /dev/null
+import uuid
+import time
+import datetime
+
+class UUID:
+ """UUID is a simple library that generates a uuid"""
+
+ def generate_UUID(self):
+ """generate a uuid"""
+ return uuid.uuid4()
+
+ 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
--- /dev/null
+[bdist_wheel]
+# 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
--- /dev/null
+from setuptools import setup
+
+setup(
+ name='python-openecomp-eteutils', # This is the name of your PyPI-package.
+ version='0.2', # Update the version number for new releases
+ description='Scripts written to be used during ete testing', # Info about script
+ install_requires=['dnspython','paramiko', 'pyyaml', 'robotframework', 'deepdiff'], # what we need
+ packages=['eteutils'], # The name of your scipts package
+ package_dir={'eteutils': 'eteutils'} # The location of your scipts package
+)
\ No newline at end of file
--- /dev/null
+# Tox (https://tox.readthedocs.io/) is a tool for running tests
+# in multiple virtualenvs. This configuration file will run the
+# test suite on all supported python versions. To use it, "pip install tox"
+# and then run "tox" from this directory.
+
+[tox]
+envlist = py27
+
+[testenv]
+commands = {envpython} setup.py test
+deps =
+
Code Review / testsuite / python-testing-utils.git / commitdiff