# -*- coding: utf-8 -*
import os
import sys
import logging
import subprocess
from datetime import datetime
from time import sleep, perf_counter
from progress.bar import IncrementalBar
CICLE_TIME = 0.016
Q_ERROR = 0
Q_OK = 2
[docs]def readVars(vdb, vardbs, variable):
"""Read a variable from multiple vardbs
Parameters
----------
vdb : class VardbVarWrapper
instance of VardbVarWrapper
vardbs : list of strings
list of vardbs
variable : string
variable name to be read
Raises
------
Exception
Variable doesn't exist in vardb. \b
Error reading variable
Returns
-------
results : array
All vardb variable values
"""
result = []
for vardb in vardbs:
result[vardb] = vdb.readVar(vardb, variable)
return result
[docs]def forceVars(vdb, vardbs, variable, value):
"""Force a variable from multiple vardbs
Parameters
----------
vdb : class VardbVarWrapper
instance of VardbVarWrapper
vardbs : list of strings
list of vardbs
variable : string
variable name to be forced
value : int
value to be forced
Raises
------
Exception
Variable doesn't exist in vardb. \b
Error forcing variable
"""
for vardb in vardbs:
vdb.forceVar(vardb, variable, value)
[docs]def unforceVars(vdb, vardbs, variable):
"""Unforce a variable from multiple vardbs
Parameters
----------
vdb : class VardbVarWrapper
instance of VardbVarWrapper
vardbs : list of strings
list of vardbs
variable : string
variable name to be unforced
Raises
------
Exception
Variable doesn't exist in vardb. \b
Error unforcing variable
"""
for vardb in vardbs:
vdb.unforceVar(vardb, variable)
[docs]def forceAliases(vdb, aliases, value):
"""Force a variable from multiple vardbs
Parameters
----------
vdb : class VardbVarWrapper
instance of VardbVarWrapper
aliases : list of functions
list of aliases to be forced
value : int
value to be forced
Raises
------
Exception
Variable doesn't exist in vardb. \b
Error forcing variable
"""
if type(aliases) != list:
aliases = [aliases]
for alias in aliases:
objectVar = globals()[alias.encode("utf-8")]
vdb.forceVar(objectVar, value)
[docs]def waitVar(vdb, vardbs, variable, value, timeoutSec=10, dontCheckQ=False):
"""Wait until variable/s have a specific value. \b
Function will finish before timeout if all values are Ok.
Parameters
----------
vdb : class VardbVarWrapper
instance of VardbVarWrapper
vardbs : list of strings
list of vardbs
variable : string
variable name to be read
value : int
variable value to be waiting for
timeoutSec : int
Function timeout. Senconds to raise exception (default is 10)
Raises
------
Exception
vardb variable FAILED in case variable != value in any vardb after timeoutSec seconds
"""
start = perf_counter()
if type(vardbs) != list:
vardbs = [vardbs]
allVardbsOK = False
failingVardb = ""
while not allVardbsOK:
allVardbsOK = True
for vardb in vardbs:
readVal = (vdb.readVarQOk(vardb, variable) if not dontCheckQ else vdb.readVar(vardb, variable))
if value != readVal:
print(f"{vardb} waiting {variable} to be {value}")
allVardbsOK = False
failingVardb = vardb
break
sleep(CICLE_TIME)
elapsedTime = perf_counter() - start
print("ElapsedTime " + str(elapsedTime))
if elapsedTime > timeoutSec:
raise Exception(f"{failingVardb} {variable} FAILED in waitVar. Expected:({value}), Read:({readVal}) in {str(timeoutSec)}s")
[docs]def waitNotVar(vdb, vardbs, variable, value, timeoutSec=10, only1=False):
"""Wait until variable/s have a value different from value. \b
Function will finish before timeout if all values are different from value.
Parameters
----------
vdb : class VardbVarWrapper
instance of VardbVarWrapper
vardbs : list of strings
list of vardbs
variable : string
variable name to be read
value : int
variable value to be waiting for
timeoutSec : int
Function timeout. Senconds to raise exception (default is 10)
only1 : bool
If true the function will exit ok if any vardb value if different from value (default is False)
Raises
------
Exception
vardb variable FAILED in case variable == value in any vardb after timeoutSec seconds.
"""
start = perf_counter()
if type(vardbs) != list:
vardbs = [vardbs]
allVardbsOK = False
failingVardb = ""
while not allVardbsOK:
allVardbsOK = True
for vardb in vardbs:
if value == vdb.readVarQOk(vardb, variable):
print("%s waiting %s to be different from %s" % (vardb, variable, value))
allVardbsOK = False
failingVardb = vardb
break
# vardb differs from value, put it out
if only1:
allVardbsOK = True
break
sleep(CICLE_TIME)
elapsedTime = perf_counter() - start
print("ElapsedTime " + str(elapsedTime))
if elapsedTime > timeoutSec:
raise Exception(f"{failingVardb} {variable} FAILED in waitNotVar. Value({value}) have not changed in {str(timeoutSec)}s")
[docs]def checkVarDoesNotChange(vdb, vardbs, variable, value, timeoutSec=10, dontCheckQ=False):
"""Check if in timeoutSec variable keeps the value. \b
Function will finish before timeout if any value is different from value.
Parameters
----------
vdb : class VardbVarWrapper
instance of VardbVarWrapper
vardbs : list of strings
list of vardbs
variable : string
variable name to be read
value : int
variable value to be waiting for
timeoutSec : int
Seconds for waiting (default is 10)
dontCheckQ : bool
if True, it will not raise exception in case of variable quality is NOK (0) (default is False)
Raises
------
Exception
vardb variable FAILED in checkVarDoesNotChange. Expected(), Read().
"""
if type(vardbs) != list:
vardbs = [vardbs]
start = perf_counter()
allVardbsOK = True
while allVardbsOK:
for vardb in vardbs:
readVal = (vdb.readVarQOk(vardb, variable) if not dontCheckQ else vdb.readVar(vardb, variable))
if value == readVal:
print(f"Waiting {vardb} {variable} to NOT change value({variable})")
else:
raise Exception(f"{vardb} {variable} FAILED in checkVarDoesNotChange. Expected({value}), Read({readVal})")
sleep(CICLE_TIME)
elapsedTime = perf_counter() - start
print("ElapsedTime " + str(elapsedTime))
if elapsedTime > timeoutSec:
return
raise Exception(f"Error in checkVarDoesNotChange")
[docs]def checkVar(vdb, vardbs, variable, value, dontCheckQ=False, condition='=='):
"""Check if variable/s have a specific value. \b
Raise exception in case variable != value in any of the vardbs.
Parameters
----------
vdb : class VardbVarWrapper
instance of VardbVarWrapper
vardbs : list of strings
list of vardbs
variable : string
variable name to be read
value : int
variable value to be waiting for
dontCheckQ : bool
if True, it will not raise exception in case of variable quality is NOK (0) (default is False)
condition : string
if not None instead of checking (variable == value), it will check (variable condition value) (default is '==')
Raises
------
Exception
vardb variable FAILED in checkVar. Expected(), Read().
"""
if type(vardbs) != list:
vardbs = [vardbs]
for vardb in vardbs:
readVal = (vdb.readVarQOk(vardb, variable) if not dontCheckQ else vdb.readVar(vardb, variable))
if not eval(f'{readVal} {condition} {value}'):
raise Exception(f"{vardb} {variable} FAILED in checkVar. Expected({value}), Read({readVal}) Condition{condition}")
[docs]def initTest(programName, tester, log_file_path, description):
"""Initialize logger and print out something similar to mamut output file. \b
programName, tester, log_file_path can be override calling the from console params (Ex. RQM).
Parameters
----------
programName : string
name of the test (can be override with sys.argv[1])
tester : string
name of the tester (can be override with sys.argv[2])
log_file_path : string (can be override with sys.argv[3])
file path to write test logs
description : string
description of the test
Returns
-------
programName : string
name of the test (can be override with sys.argv[1])
tester : string
name of the tester (can be override with sys.argv[2])
log_file_path : string (can be override with sys.argv[3])
file path to write test logs
"""
n = len(sys.argv)
if n > 1:
programName = sys.argv[1]
if n > 2:
tester = sys.argv[2]
if n > 3:
log_file_path = sys.argv[3]
logging.basicConfig(filename=log_file_path, level=logging.NOTSET, format="%(message)s", filemode="w")
logging.info("--------------------------------------------------------------------------------")
logging.info("Test name: " + programName)
logging.info("Requirements:")
logging.info("Procedures:")
logging.info("Date: " + datetime.now().strftime("%Y-%m-%d-%H:%M:%S"))
logging.info("Place:")
logging.info("tester: " + tester)
logging.info("description: " + description)
logging.info("Traces:")
logging.info("--------------------------------------------------------------------------------")
logging.info(" ")
logging.info(" ")
logging.info("****************************************************")
logging.info("Main test execution started")
logging.info("****************************************************")
return (
programName,
tester,
log_file_path,
)
[docs]def finishTestOk(programName):
"""Print out something similar to mamut when test has PASSED OK
Parameters
----------
programName : string
name of the test
"""
logging.info(" ")
logging.info(" ")
logging.info("****************************************************")
logging.info("Main test execution finished")
logging.info("****************************************************")
logging.info("--------------------------------------------------------------------------------")
logging.info("Code Title Result Incident & Cause")
logging.info(programName + " PASSED")
logging.info("--------------------------------------------------------------------------------")
[docs]def finishTestNotOK(errorMessage, programName):
"""Print out something similar to mamut when test FAILS
Parameters
----------
errorMessage : string
Error message
programName : string
name of the test
"""
logging.info(" ")
logging.info(" ")
logging.info("****************************************************")
logging.info("Main test execution finished")
logging.info("****************************************************")
logging.info("--------------------------------------------------------------------------------")
logging.info("Code Title Result Incident & Cause")
logging.info(programName + " FAILED " + errorMessage)
logging.info("--------------------------------------------------------------------------------")
[docs]def stepOk(stepNumber):
"""Print out something similar to mamut when step has PASSED OK
Parameters
----------
stepNumber : string
number of the step
"""
logging.info(f"STEP {stepNumber}: PASSED")
[docs]def stepNOk(stepNumber, error=""):
"""Print out something similar to mamut when step has PASSED NOK
Parameters
----------
stepNumber : string
number of the step
error: string
error message to show (default is empty )
Raises
------
Exception
TEST STEP NOK: ERROR
"""
raise Exception(f"TEST STEP {stepNumber} NOK: ERROR {error}")
[docs]def wait(wait, message="Sleeping..."):
"""Prints a countdown progress bar while sleeping
Parameters
----------
wait : int
sleeping time in seconds
message : string
Custom message to show (default is 'Sleeping...' )
"""
bar = IncrementalBar(message, max=wait)
for i in range(wait, 0, -1):
bar.next()
sleep(1)
bar.finish()
[docs]def convertCompactRegToCsv(filePath, delete):
"""Extract CompactReg file into csv
Parameters
----------
filePath : string
Full file path of the CompactReg
delete : bool
If true remove generated metadata and the original file
Raises
------
Exception
Error cleaning compact file
Returns
-------
returnCode : int
return code of subprocess call
"""
dirPath = os.path.dirname(os.path.realpath(__file__))
filePath = os.path.abspath(filePath)
cwd = os.getcwd()
fileName = filePath.split(os.sep)[-1]
folder = os.path.dirname(filePath)
binaryPath = dirPath + "\\..\\bin\\RegToCsvConverter.exe" # TODO: fix this
os.chdir(folder)
returnCode = subprocess.call(binaryPath + ' REG_FILE="' + filePath + '" BASENAME=' + fileName, shell=True)
if delete:
# remove generated metadata and the original file
sleep(5)
try:
os.remove(filePath)
os.remove(folder + "\\" + fileName + "_Metadata.xml")
except Exception as e:
print("Error cleaning compact file: %s" % str(e))
os.chdir(cwd)
return returnCode