blob: 78bbc86bfd955d6491c689d90719e5da5109b20c [file] [log] [blame]
# Copyright 2020-2023 Open Networking Foundation (ONF) and the ONF Contributors
# delivered by ADTRAN, Inc.
#
# 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 __future__ import absolute_import
from __future__ import print_function
import inspect
import os
import operator
import requests
# global definition of keys (find in given 'inventory_data')
_NAME = 'name'
_CHILDREN = 'children'
_SENSOR_DATA = 'sensor_data'
_ROOT = 'root'
_INVENTORY = 'inventory'
_UUID = 'uuid'
def test(success):
if success is True:
return True
return False
def unique():
"""Returns the current filename and line number in our program."""
trace = str(os.path.basename(__file__) +
"[" + str(inspect.currentframe().f_back.f_lineno) + "]:")
return trace
# check if given paramter exist in inventory data, search recursive
def check_in_inventory_Component_data(inventory_data, name, element, value):
print(unique(), str(inventory_data), str(name), str(element), str(value))
if inventory_data.get(_NAME) == name and inventory_data.get(element) == value:
return True
for child in inventory_data[_CHILDREN]:
print(unique(), str(child))
if child.get(_NAME) == name and child.get(element) == value:
return True
if _SENSOR_DATA in child:
for sensor_data in child[_SENSOR_DATA]:
print(unique(), str(sensor_data))
if sensor_data.get(element) == value:
return True
if _CHILDREN in child:
result = check_in_inventory_Component_data(child, name, element, value)
if result is True:
return result
return False
# get uuid out of inventory data, search recursive
def get_uuid_from_inventory_Component_data(inventory_data, searchFor):
print(unique(), str(inventory_data), ', ', str(searchFor))
if inventory_data.get(_NAME) == searchFor:
return inventory_data.get(_UUID)
for child in inventory_data[_CHILDREN]:
print(unique(), str(child))
result = None
if child.get(_NAME) == searchFor:
print(unique(), str(child[_NAME]))
result = child.get(_UUID)
print(unique(), child.keys())
if result is None and _CHILDREN in child:
result = get_uuid_from_inventory_Component_data(child, searchFor)
if result is not None:
return result
return None
def get_uuid_from_Inventory_Element(inventory, searchFor):
for children in inventory[_INVENTORY][_ROOT][_CHILDREN]:
return get_uuid_from_inventory_Component_data(children, searchFor)
return None
def check_Inventory_Element(inventory, name, element, value):
for childrens in inventory[_INVENTORY][_ROOT][_CHILDREN]:
return check_in_inventory_Component_data(childrens, name, element, value)
return False
def getWord(line, number):
line_in_list = line.split()
if len(line_in_list) >= number-1:
return line_in_list[number-1]
return ""
def decode(data):
decoded_data = data
print(unique(), str(decoded_data))
# Compares two values using a given operator. The values are converted to float first so that
# numbers as strings are also accepted. Returns True or False.
# operator: ==, !=, <, <=, >, >=
# Example:
# | ${result} | Compare | 100 | > | 5 | # True |
def compare(value1, op, value2):
ops = {"==": operator.eq,
"!=": operator.ne,
"<": operator.lt,
"<=": operator.le,
">": operator.gt,
">=": operator.ge}
return ops[op](float(value1), float(value2))
# Validates two values using a given operator.
# The values are converted to float first so that numbers as strings are also accepted.
# Second value has to be a list in case of operator is 'in' or 'range'
# Returns True or False.
# operator: in, range, ==, !=, <, <=, >, >=
# Example:
# | ${result} | validate | 100 | > | 5 | # True |
# | ${result} | validate | 11 | in | ['11','264','329'] | # True |
# | ${result} | validate | 1 | range | ['0','1'] | # True |
def validate(value1, op, value2):
if op == "in":
return (float(value1) in [float(i) for i in value2])
if op == "range":
return ((compare(value1, ">=", value2[0])) and (compare(value1, "<=", value2[1])))
return compare(value1, op, value2)
def get_memory_consumptions(address, container, namespace="default"):
"""
Query Prometheus and generate instantaneous memory consumptions for given pods under test
:param address: string The address of the Prometheus instance to query
:container: string The pod name
:param namespace: string The pod namespace
:return: memory consumtion value
"""
container_mem_query = ('container_memory_working_set_bytes{namespace="%s",container="%s"}' %
(namespace, container))
mem_params = {
"query": container_mem_query,
}
r = requests.get("http://%s/api/v1/query" % address, mem_params)
container_cpu = r.json()["data"]["result"]
if len(container_cpu) > 0:
return container_cpu[0]["value"][1]
else:
return -1