# limitations under the License.
from ONAPLibrary.BaseAAIKeywords import BaseAAIKeywords
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
class AAI(HybridCore):
+# 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 robot.api.deco import keyword
import base64
# limitations under the License.
from ONAPLibrary.BaseCLAMPKeywords import BaseCLAMPKeywords
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
class CLAMP(HybridCore):
+# 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.
+
import dns.message
import dns.name
import dns.query
from hashlib import md5
from paramiko import RSAKey
from paramiko.ssh_exception import PasswordRequiredException
-
+from six import string_types
from ONAPLibrary.Utilities import Utilities
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, str) or isinstance(template_file, unicode):
+ if isinstance(template_file, string_types):
fin = open(template_file, 'r')
yamlobj = yaml.load(fin)
return yamlobj
"""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"""
contents = None
- if isinstance(template_file, str) or isinstance(template_file, unicode):
+ if isinstance(template_file, string_types):
fin = open(template_file, 'r')
yamlobj = yaml.load(fin)
fin.close()
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, str) or isinstance(template_file, unicode):
+ if isinstance(template_file, string_types):
fin = open(template_file, 'r')
yamlobj = yaml.load(fin)
fin.close()
# See the License for the specific language governing permissions and
# limitations under the License.
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
from ONAPLibrary.JSONKeywords import JSONKeywords
from ONAPLibrary.JSONPathKeywords import JSONPathKeywords
# 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 json
+import json
from robot.api.deco import keyword
from deepdiff import DeepDiff
+from six import string_types
class JSONKeywords(object):
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, str) or isinstance(left, unicode):
+ if isinstance(left, string_types):
left_json = json.loads(left)
else:
left_json = left
- if isinstance(right, str) or isinstance(right, unicode):
+ if isinstance(right, string_types):
right_json = json.loads(right)
else:
right_json = right
# 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 json
+import json
+from six import string_types
from robot.api.deco import keyword
from jsonpath_rw import parse
which is converted into string if needed and then compares them, returning the matches."""
jsonpath_expr = parse(expression)
- if isinstance(target, str) or isinstance(target, unicode):
+ if isinstance(target, string_types):
search_json = json.dumps(target)
else:
search_json = target
# See the License for the specific language governing permissions and
# limitations under the License.
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
from ONAPLibrary.KafkaKeywords import KafkaKeywords
partitions = []
for val in partition_set:
partitions.append(TopicPartition(str(topic_name), val))
- consumer.assign(partitions)
- last = consumer.end_offsets(partitions)
- offset = max(last.values())
+ consumer.assign(partitions)
+ last = consumer.end_offsets(partitions)
+ offset = max(last.values())
if set_offset_to_earliest:
consumer.seek_to_beginning()
else:
for tp in partitions:
- consumer.seek(tp, offset - 1)
+ consumer.seek(tp, offset - 1)
return consumer
# See the License for the specific language governing permissions and
# limitations under the License.
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
from ONAPLibrary.MUSICKeywords import MUSICKeywords
# limitations under the License.
from ONAPLibrary.SNIROKeywords import SNIROKeywords
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
class OOF(HybridCore):
# See the License for the specific language governing permissions and
# limitations under the License.
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
from ONAPLibrary.BaseOpenstackKeywords import BaseOpenstackKeywords
from ONAPLibrary.HEATKeywords import HEATKeywords
# See the License for the specific language governing permissions and
# limitations under the License.
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
from ONAPLibrary.PreloadDataKeywords import PreloadDataKeywords
# See the License for the specific language governing permissions and
# limitations under the License.
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
from ONAPLibrary.ProtobufKeywords import ProtobufKeywords
+# 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 robot.api import logger
+
def log_wrapped(func):
def _log_wrapped(*args, **kwargs):
if 'endpoint' in kwargs:
return _log_wrapped
+
def default_keywords(func):
def _default_keywords(*args, **kwargs):
dicts = _keyword_defaults(**kwargs)
return kwargs
return _default_keywords
-
# See the License for the specific language governing permissions and
# limitations under the License.
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
from ONAPLibrary.BaseSDCKeywords import BaseSDCKeywords
# limitations under the License.
from ONAPLibrary.PreloadSDNCKeywords import PreloadSDNCKeywords
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
from ONAPLibrary.BaseSDNCKeywords import BaseSDNCKeywords
# limitations under the License.
from ONAPLibrary.RequestSOKeywords import RequestSOKeywords
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
from ONAPLibrary.BaseSOKeywords import BaseSOKeywords
from ONAPLibrary.CloudConfigSOKeywords import CloudConfigSOKeywords
# See the License for the specific language governing permissions and
# limitations under the License.
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
from ONAPLibrary.ServiceMappingKeywords import ServiceMappingKeywords
-import socket, ssl
+# 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.
+
+import socket
+import ssl
from robot.api.deco import keyword
# See the License for the specific language governing permissions and
# limitations under the License.
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
from ONAPLibrary.TemplatingKeywords import TemplatingKeywords
# See the License for the specific language governing permissions and
# limitations under the License.
-from ONAPLibrary.robotlibcore import HybridCore
+from robotlibcore import HybridCore
from ONAPLibrary.DNSKeywords import DNSKeywords
from ONAPLibrary.SocketKeywords import SocketKeywords
from ONAPLibrary.UUIDKeywords import UUIDKeywords
+++ /dev/null
-# Copyright 2017- Robot Framework Foundation
-#
-# 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.
-
-"""Generic test library core for Robot Framework.
-
-Main usage is easing creating larger test libraries. For more information and
-examples see the project pages at
-https://github.com/robotframework/PythonLibCore
-"""
-
-import inspect
-import sys
-
-try:
- from robot.api.deco import keyword
-except ImportError: # Support RF < 2.9
- def keyword(name=None, tags=()):
- if callable(name):
- return keyword()(name)
- def decorator(func):
- func.robot_name = name
- func.robot_tags = tags
- return func
- return decorator
-
-
-PY2 = sys.version_info < (3,)
-
-__version__ = '1.0.1.dev1'
-
-
-class HybridCore(object):
-
- def __init__(self, library_components):
- self.keywords = {}
- self.attributes = {}
- self.add_library_components(library_components)
- self.add_library_components([self])
-
- def add_library_components(self, library_components):
- for component in library_components:
- for name, func in self._get_members(component):
- if callable(func) and hasattr(func, 'robot_name'):
- kw = getattr(component, name)
- kw_name = func.robot_name or name
- self.keywords[kw_name] = kw
- # Expose keywords as attributes both using original
- # method names as well as possible custom names.
- self.attributes[name] = self.attributes[kw_name] = kw
-
- def _get_members(self, component):
- if inspect.ismodule(component):
- return inspect.getmembers(component)
- if inspect.isclass(component):
- raise TypeError('Libraries must be modules or instances, got '
- 'class {!r} instead.'.format(component.__name__))
- if type(component) != component.__class__:
- raise TypeError('Libraries must be modules or new-style class '
- 'instances, got old-style class {!r} instead.'
- .format(component.__class__.__name__))
- return self._get_members_from_instance(component)
-
- def _get_members_from_instance(self, instance):
- # Avoid calling properties by getting members from class, not instance.
- cls = type(instance)
- for name in dir(instance):
- owner = cls if hasattr(cls, name) else instance
- yield name, getattr(owner, name)
-
- def __getattr__(self, name):
- if name in self.attributes:
- return self.attributes[name]
- raise AttributeError('{!r} object has no attribute {!r}'
- .format(type(self).__name__, name))
-
- def __dir__(self):
- if PY2:
- my_attrs = dir(type(self)) + list(self.__dict__)
- else:
- my_attrs = super().__dir__()
- return sorted(set(my_attrs) | set(self.attributes))
-
- def get_keyword_names(self):
- return sorted(self.keywords)
-
-
-class DynamicCore(HybridCore):
- _get_keyword_tags_supported = False # get_keyword_tags is new in RF 3.0.2
-
- def run_keyword(self, name, args, kwargs):
- return self.keywords[name](*args, **kwargs)
-
- def get_keyword_arguments(self, name):
- kw = self.keywords[name] if name != '__init__' else self.__init__
- args, defaults, varargs, kwargs = self._get_arg_spec(kw)
- args += ['{}={}'.format(name, value) for name, value in defaults]
- if varargs:
- args.append('*{}'.format(varargs))
- if kwargs:
- args.append('**{}'.format(kwargs))
- return args
-
- def _get_arg_spec(self, kw):
- if PY2:
- spec = inspect.getargspec(kw)
- keywords = spec.keywords
- else:
- spec = inspect.getfullargspec(kw)
- keywords = spec.varkw
- args = spec.args[1:] if inspect.ismethod(kw) else spec.args # drop self
- defaults = spec.defaults or ()
- nargs = len(args) - len(defaults)
- mandatory = args[:nargs]
- defaults = zip(args[nargs:], defaults)
- return mandatory, defaults, spec.varargs, keywords
-
- def get_keyword_tags(self, name):
- self._get_keyword_tags_supported = True
- return self.keywords[name].robot_tags
-
- def get_keyword_documentation(self, name):
- if name == '__intro__':
- return inspect.getdoc(self) or ''
- if name == '__init__':
- return inspect.getdoc(self.__init__) or ''
- kw = self.keywords[name]
- doc = inspect.getdoc(kw) or ''
- if kw.robot_tags and not self._get_keyword_tags_supported:
- tags = 'Tags: {}'.format(', '.join(kw.robot_tags))
- doc = '{}\n\n{}'.format(doc, tags) if doc else tags
- return doc
-
-
-class StaticCore(HybridCore):
-
- def __init__(self):
- HybridCore.__init__(self, [])
+++ /dev/null
-'''
-Created on Apr 7, 2017
-
-@author: jf9860
-'''
-from threading import Thread
-import subprocess
-import os
-from datetime import datetime
-import logging
-
-class RunEte(Thread):
- '''
- classdocs
- '''
- robot_test = ""
- robot_command = "runEteTag.sh"
- soaksubfolder = ""
- test_number =0
-
- def __init__(self, test_name, soaksubfolder, test_number):
- '''
- Constructor
- '''
- super(RunEte, self).__init__()
- self.robot_test = test_name
- self.soaksubfolder = soaksubfolder
- self.test_number = test_number
-
- def run(self):
- logging.info("{} ({}) started - {}".format(self.getName(), self.robot_test, str(datetime.now())))
- try:
- ''' Add the '/' here so that the shell doesn't require a subfolder... '''
- 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 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())))
+++ /dev/null
-'''
-Created on Apr 7, 2017
-
-@author: jf9860
-'''
-import json
-
-class TestConfig(object):
- '''
- The profile defines a cycle of tests. Each entry is defined as
- [<seconds to wait>, [<list of ete tags to run after the wait]],
- '''
- profile = [
- [0, ["health"]],
- ]
-
- duration=10
- cyclelength=60
-
- def __init__(self, duration=None, cyclelength=None, json=None):
- '''
- Constructor
- '''
- if (json != None):
- self.parseConfig(json)
- if (duration != None):
- self.duration = duration
- if (cyclelength != None):
- self.cyclelength = cyclelength
- running_time = 0
- for p in self.profile:
- secs = p[0]
- running_time = running_time + secs
- if (running_time < self.cyclelength):
- last = self.cyclelength - running_time
- self.profile.append([last, []])
-
- def parseConfig(self, fileName):
- with open(fileName) as data_file:
- config = json.load(data_file)
- self.profile = config["profile"]
- self.cyclelength = config["cyclelength"]
- self.duration = config["duration"]
-
-
- def to_string(self):
- pstring = 'Cycle length is {} seconds'.format(self.cyclelength)
- pstring = '{}\nDuration is {} seconds'.format(pstring, self.duration)
- running_time = 0
- for p in self.profile:
- secs = p[0]
- running_time = running_time + secs
- for ete in p[1]:
- pstring = "{0}\n{1:08d} : {2:08d} : {3}".format(pstring, secs, running_time, ete)
- if (len(p[1]) == 0):
- pstring = "{0}\n{1:08d} : {2:08d} : {3}".format(pstring, secs, running_time, "")
- return pstring
-
-
+++ /dev/null
-'''
-Created on Apr 7, 2017
-
-@author: jf9860
-'''
-import time
-import os
-from loadtest.RunEte import RunEte
-from loadtest.TestConfig import TestConfig
-import logging
-
-class TestController(object):
- '''
- classdocs
- '''
-
- threads = {}
- threadid = 0
- soaksubfolder = 'soak_' + str(os.getpid())
- test_number = 0
-
- def __init__(self, options):
- '''
- Constructor
- '''
- self.config = TestConfig(duration=options.duration, cyclelength=options.cyclelength, json=options.profile)
- logging.info(self.config.to_string())
-
- def execute(self):
- starttime = time.time()
- endtime = starttime + self.config.duration
- profileindex = 0
- currenttime = time.time()
- logging.info("{}:{}:{}".format(starttime, endtime, currenttime))
- while currenttime < endtime:
- if (profileindex >= len(self.config.profile)):
- profileindex = 0
- profile = self.config.profile[profileindex]
- sleeptime = profile[0]
- currenttime = time.time()
- if ((currenttime + sleeptime) < endtime):
- time.sleep(sleeptime)
- self.schedule(profile)
- profileindex = profileindex + 1
- currenttime = time.time()
- else:
- currenttime = endtime
-
- for threadname in self.threads:
- logging.info("TestController waiting on " + threadname)
- t = self.threads[threadname]
- t.join()
- logging.info("Soak test completed")
-
- def schedule(self, profile):
- self.remove_completed_threads()
- tests = profile[1]
- for test in tests:
- self.schedule_one(test)
-
- def schedule_one(self, test):
- self.test_number = self.test_number + 1
- self.threadid = self.threadid + 1
- threadname = "RunEte_" + str(self.threadid)
- ''' test for max threads '''
- t = RunEte(test, self.soaksubfolder, str(self.test_number))
- t.setName(threadname)
- t.start()
- self.threads[threadname] = t
-
-
- def remove_completed_threads(self):
- toremove = []
- for threadname in self.threads:
- t = self.threads[threadname]
- if (t.isAlive() == False):
- toremove.append(threadname)
- for threadname in toremove:
- logging.info("Removing " + threadname)
- del(self.threads[threadname])
\ No newline at end of file
+++ /dev/null
-#!/usr/bin/env python
-# encoding: utf-8
-'''
-loadtest.TestMain -- shortdesc
-
-loadtest.TestMain is a description
-
-It defines classes_and_methods
-
-@author: user_name
-
-@copyright: 2017 organization_name. All rights reserved.
-
-@license: license
-
-@contact: user_email
-@deffield updated: Updated
-'''
-
-import sys
-import os
-
-from optparse import OptionParser, Values
-
-from loadtest.TestController import TestController
-
-__all__ = []
-__version__ = 0.1
-__date__ = '2017-04-07'
-__updated__ = '2017-04-07'
-
-DEBUG = 1
-TESTRUN = 0
-PROFILE = 0
-import time
-import logging
-
-def main(argv=None):
- '''Command line options.'''
- program_name = os.path.basename(sys.argv[0])
- program_version = "v0.1"
- program_build_date = "%s" % __updated__
-
- program_version_string = '%%prog %s (%s)' % (program_version, program_build_date)
- #program_usage = '''usage: spam two eggs''' # optional - will be autogenerated by optparse
- program_longdesc = '''''' # optional - give further explanation about what the program does
- program_license = "Copyright 2017 user_name (organization_name) \
- Licensed under the Apache License 2.0\nhttp://www.apache.org/licenses/LICENSE-2.0"
-
- if argv is None:
- argv = sys.argv[1:]
- try:
- # setup option parser
- parser = OptionParser(version=program_version_string, epilog=program_longdesc, description=program_license)
- parser.add_option("-d", "--duration", dest="duration", help="duration of soak test in seconds [default: %default]", type=int)
- parser.add_option("-l", "--logfile", dest="logfile", help="Full path soak log file name")
- parser.add_option("-c", "--cyclelength", dest="cyclelength", help="Length of a single cycle through the config.\nMust be longer than a single iteration", type=int)
- parser.add_option("-p", "--profile", dest="profile", help="Filename of json profile file")
- parser.set_defaults(logfile="")
- (opts, args) = parser.parse_args(argv)
-
- if (opts.logfile != ""):
- logging.basicConfig(filename=opts.logfile, level=logging.DEBUG)
- else:
- logging.basicConfig(level=logging.DEBUG)
- controller = TestController(opts)
- controller.execute()
-
- except Exception as e:
- indent = len(program_name) * " "
- sys.stderr.write(program_name + ": " + repr(e) + "\n")
- sys.stderr.write(indent + " for help use --help")
- return 2
-
-
-if __name__ == "__main__":
- if DEBUG:
- print("debug")
- if TESTRUN:
- import doctest
- doctest.testmod()
- if PROFILE:
- import cProfile
- import pstats
- profile_filename = 'loadtest.TestMain_profile.txt'
- cProfile.run('main()', profile_filename)
- statsfile = open("profile_stats.txt", "wb")
- p = pstats.Stats(profile_filename, stream=statsfile)
- stats = p.strip_dirs().sort_stats('cumulative')
- stats.print_stats()
- statsfile.close()
- sys.exit(0)
- sys.exit(main())
\ No newline at end of file
'requests',
'robotframework',
'robotframework-requests',
+ 'robotlibcore-temp',
'six',
'urllib3'
], # what we need to run library
- packages=['loadtest', 'vcpeutils', 'ONAPLibrary'], # The name of your scripts package
+ packages=['vcpeutils', 'ONAPLibrary'], # The name of your scripts package
package_dir={
- 'loadtest': 'loadtest',
'vcpeutils': 'vcpeutils',
'ONAPLibrary': 'ONAPLibrary'
}, # The location of your scipts package
-#! /usr/bin/python
-
import time
-
-from vcpeutils.csar_parser import *
+import os
+import json
+from vcpeutils.csar_parser import CsarParser
from robot.api import logger
from datetime import datetime
import sys
sys.exit()
self.aai.wait_for_node_to_exist('generic-vnf', 'vnf-id', vnf_instance_id)
- # SDNC Preload
+ # SDNC Preload
preloader = PreloadSDNCKeywords()
vfmodule_name = '_'.join(['vf',
parser.vfmodule_models[0]['modelCustomizationName'].split('..')[0].lower(),
-#! /usr/bin/python
import os
import zipfile
import shutil
Code Review / testsuite / python-testing-utils.git / commitdiff