Commit c0521579 authored by Marco Clemencic's avatar Marco Clemencic
Browse files

Use correct runtime environment for commands and tests.

Imported the Python code from the LHCb package EnvConfig.
parent e3ca916e
'''
Created on Jun 27, 2011
@author: mplajner
'''
import xmlModule
import os
import platform
from time import gmtime, strftime
import Variable
class Environment():
'''object to hold settings of environment'''
def __init__(self, loadFromSystem = True, useAsWriter=False, reportLevel = 1):
'''Initial variables to be pushed and setup
append switch between append and prepend for initial variables.
loadFromSystem causes variable`s system value to be loaded on first encounter.
If useAsWriter == True than every change to variables is recorded to XML file.
reportLevel sets the level of messaging.
**kwargs contain initial variables.
'''
self.report = xmlModule.Report(reportLevel)
if platform.system() != 'Linux':
self.sysSeparator = ';'
else:
self.sysSeparator = ':'
self.separator = ':'
self.posActions = ['append','prepend','set','unset','remove', 'remove-regexp', 'declare']
self.variables = {}
self.loadFromSystem = loadFromSystem
self.asWriter = useAsWriter
if useAsWriter:
self.writer = xmlModule.XMLFile()
self.startXMLinput()
def vars(self, strings=True):
'''returns dictionary of all variables optionally converted to string'''
if strings:
vars = self.variables.copy()
for item in vars:
vars[item] = self.var(item).value(True)
return vars
else:
return self.variables
def var(self, name):
'''Gets a single variable. If not available then tries to load from system.'''
if name in self.variables.keys():
return self.variables[name]
else:
self._loadFromSystem(name, '', 'append')
return self.variables[name]
def search(self, varName, expr, regExp = False):
'''Searches in a variable for a value.'''
return self.variables[varName].search(expr, regExp)
def append(self, name, value):
'''Appends to an existing variable.'''
if self.asWriter:
self._writeVarToXML(name, 'append', value)
else:
if name in self.variables.keys():
self.variables[name].append(value, self.separator, self.variables)
else:
self.declare(name, 'list', False)
self.append(name, value)
def prepend(self, name, value):
'''Prepends to an existing variable, or create a new one.'''
if self.asWriter:
self._writeVarToXML(name, 'prepend', value)
else:
if name in self.variables.keys():
self.variables[name].prepend(value, self.separator, self.variables)
else:
self.declare(name, 'list', False)
self.prepend(name, value)
def declare(self, name, type, local):
'''Creates an instance of new variable. It loads values from the OS if the variable is not local.'''
if self.asWriter:
self._writeVarToXML(name, 'declare', '', type, local)
if not isinstance(local, bool):
if str(local).lower() == 'true':
local = True
else:
local = False
if name in self.variables.keys():
if self.variables[name].local != local:
raise Variable.EnvironmentError(name, 'redeclaration')
else:
if type.lower() == "list":
if not isinstance(self.variables[name],Variable.List):
raise Variable.EnvironmentError(name, 'redeclaration')
else:
if not isinstance(self.variables[name],Variable.Scalar):
raise Variable.EnvironmentError(name, 'redeclaration')
if type.lower() == "list":
a = Variable.List(name, local, report=self.report)
else:
a = Variable.Scalar(name, local, report=self.report)
if self.loadFromSystem and not local:
if name in os.environ.keys():
a.set(os.environ[name], self.sysSeparator, environment=self.variables, resolve=False)
self.variables[name] = a
def set(self, name, value):
'''Sets a single variable - overrides any previous value!'''
name = str(name)
if self.asWriter:
self._writeVarToXML(name, 'set', value)
else:
if name in self.variables:
self.variables[name].set(value, self.separator, self.variables)
else:
self.declare(name, 'list', False)
self.set(name, value)
def searchFile(self, file, varName):
'''Searches for appearance of variable in a file.'''
XMLFile = xmlModule.XMLFile()
variable = XMLFile.variable(file, name=varName)
return variable
def unset(self, name, value=None):
'''Unsets a single variable to an empty value - overrides any previous value!'''
if self.asWriter:
self._writeVarToXML(name, 'unset', '')
else:
if name in self.variables:
self.variables[name].set([], self.separator)
else:
a = Variable.List(name, report=self.report)
self.variables[name] = a
def remove(self, name, value, regexp=False):
'''Remove value from variable.'''
if self.asWriter:
self._writeVarToXML(name, 'remove', value)
else:
if name in self.variables:
self.variables[name].remove(value, self.separator, regexp)
elif self.loadFromSystem:
self.declare(name, 'list', False)
def remove_regexp(self, name, value):
self.remove(name, value, True)
def loadXML(self, fileName = None, namespace = 'EnvSchema'):
'''Loads XML file for input variables.'''
XMLfile = xmlModule.XMLFile()
variables = XMLfile.variable(fileName, namespace = namespace)
i = 0
for variable in variables:
i += 1
if variable[1] not in self.posActions:
self.report.addError('Node '+str(i)+': No action taken with var "' + variable[0] + '". Probably wrong action argument: "'+variable[1]+'".')
elif variable[1] == 'declare':
self.declare(str(variable[0]), str(variable[2]), variable[3])
else:
if variable[1] == 'remove-regexp':
variable[1] = 'remove_regexp'
eval('self.'+variable[1]+'(str(variable[0]), str(variable[2]))')
def startXMLinput(self):
'''Renew writer for new input.'''
self.writer.resetWriter()
def finishXMLinput(self, outputFile = ''):
'''Finishes input of XML file and closes the file.'''
self.writer.writeToFile(outputFile)
def writeToFile(self, fileName, shell='sh'):
'''Creates an output file with a specified name to be used for setting variables by sourcing this file'''
f = open(fileName, 'w')
if shell == 'sh':
f.write('#!/bin/bash\n')
for variable in self.variables:
if not self[variable].local:
f.write('export ' +variable+'='+self[variable].value(True, self.sysSeparator)+'\n')
elif shell == 'csh':
f.write('#!/bin/csh\n')
for variable in self.variables:
if not self[variable].local:
f.write('setenv ' +variable+' '+self[variable].value(True, self.sysSeparator)+'\n')
else:
f.write('')
f.write('REM This is an enviroment settings file generated on '+strftime("%a, %d %b %Y %H:%M:%S\n", gmtime()))
for variable in self.variables:
if not self[variable].local:
f.write('set '+variable+'='+self[variable].value(True, self.sysSeparator)+'\n')
f.close()
def writeToXMLFile(self, fileName):
'''Writes the current state of environment to a XML file.
NOTE: There is no trace of actions taken, variables are written with a set action only.
'''
writer = xmlModule.XMLFile()
for var in self.variables.keys():
writer.writeVar(var, 'set', self.variables[var].value(True, self.separator))
writer.writeToFile(fileName)
def loadAllSystemVariables(self):
'''Loads all variables from the current system settings.'''
for val in os.environ.keys():
if not val in self.variables.keys():
self.declare(val, 'list', False)
def _concatenate(self, value):
'''Returns a variable string with separator separator from the values list'''
stri = ""
for it in value:
stri += it + self.separator
stri = stri[0:len(stri)-1]
return stri
def _writeVarToXML(self, name, action, value, type='list', local='false'):
'''Writes single variable to XML file.'''
if isinstance(value, list):
value = self._concatenate(value)
self.writer.writeVar(name, action, value, type, local)
def __getitem__(self, key):
return self.variables[key]
def __setitem__(self, key, value):
if value in self.variables.keys():
self.report.addWarn('Addition canceled because of duplicate entry. Var: "' + self.varName + '" value: "' + value + '".')
else:
self.append(key, value)
def __delitem__(self, key):
del self.variables[key]
def __iter__(self):
for i in self.variables:
yield i
def __contains__(self, item):
return item in self.variables.keys()
def __len__(self):
return len(self.variables.keys())
'''
Created on Jun 27, 2011
@author: mplajner
'''
import re
import os
import platform
class List():
'''
Class for manipulating with environment lists.
It holds its name and values represented by a list.
Some operations are done with separator, which is usually colon. For windows use semicolon.
'''
def __init__(self, name, local=False, report=None):
self.report = report
self.varName = name
self.local = local
self.val = []
def name(self):
'''Returns the name of the List.'''
return self.varName
def set(self, value, separator = ':', environment={}, resolve=True):
'''Sets the value of the List. Any previous value is overwritten.'''
if resolve:
value = self.resolveReferences(value, environment, separator)
else:
value = value.split(separator)
self.val = filter(None, value)
def unset(self, value, separator = ':', environment={}):
'''Sets the value of the List to empty. Any previous value is overwritten.'''
self.val = ''
def value(self, asString = False, separator = ':'):
'''Returns values of the List. Either as a list or string with desired separator.'''
if asString:
stri = self._concatenate(separator)
return stri.replace(']', ':')
else:
lis = self.val[:]
if platform.system() != 'Linux':
for item in lis:
item.replace(']',':')
return lis
def remove_regexp(self,value,separator = ':'):
self.remove(value, separator, True)
def remove(self, value, separator = ':', regexp=False):
'''Removes value(s) from List. If value is not found, removal is canceled.'''
if regexp:
value = self.search(value, True)
elif isinstance(value,str):
value = value.split(separator)
for i in range(len(value)):
val = value[i]
if val not in value:
self.report.addWarn('Value "'+val+'" not found in List: "'+self.varName+'". Removal canceled.')
while val in self.val:
self.val.remove(val)
def append(self,value, separator = ':', environment={}, warningOn=True):
'''Adds value(s) at the end of the list.'''
value = self.resolveReferences(value, environment, separator)
i = 0
while i < len(value):
if value[i] == '' :
value.remove(value[i])
continue
if value[i] in self.val:
if warningOn:
self.report.addWarn('Var: "'+self.varName+'" value: "' + value[i] + '". Addition canceled because of duplicate entry.')
value.remove(value[i])
else:
i += 1
self.val.extend(value)
def prepend(self,value, separator = ':', environment={}):
'''Adds value(s) at the beginning of the list.'''
'''resolve references and duplications'''
value = self.resolveReferences(value, environment, separator)
i = 0
while i < len(value):
if value[i] == '' :
value.remove(value[i])
continue
if value[i] in self.val:
self.report.addWarn('Var: "' + self.varName + '" value: "' + value[i] + '". Addition canceled because of duplicate entry.')
value.remove(value[i])
else:
i += 1
if isinstance(value,str):
self.val.insert(0, value)
else:
i = 0
for it in value:
self.val.insert(0+i, it)
i=i+1
def search(self, expr, regExp):
'''Searches in List`s values for a match
Use string value or set regExp to True.
In the first case search is done only for an exact match for one of List`s value ('^' and '$' added).
'''
if not regExp:
expr = '^' + expr + '$'
v = re.compile(expr)
res = []
for val in self.val:
if v.search(val):
res.append(val)
return res
def repl(self, s, d):
v = re.compile(r"\$([A-Za-z_][A-Za-z0-9_]*)|\$\(([A-Za-z_][A-Za-z0-9_]*)\)|\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
m = v.search(s)
if m:
s = s[:m.start()] + str(d[filter(None,m.groups())[0]]) + s[m.end():]
return self.repl(s,d)
else:
return s
def resolveReferences(self, value, environment, separator = ':'):
'''Resolves references to Lists'''
if isinstance(value, list):
str = ''
for val in value:
str = val + '' + separator
value = str[:len(str)-1]
value = self.repl(value, environment)
value = value.split(separator)
value = self._remDuplicates(value)
value = self._changeSlashes(value)
return value
def _changeSlashes(self, value):
'''Changes slashes depending on operating system.'''
i = 0
while i < len(value):
if len(value[i]) == 0:
del value[i]
continue
if value[i][0] == '[':
if platform.system() != 'Linux':
value[i] = value[i][1:]
else:
value[i] = value[i][3:]
value[i] = os.path.normpath(value[i])
i+=1
return value
def _concatenate(self, separator):
'''Returns a List string with separator "separator" from the values list'''
stri = ""
for it in self.val:
stri += it + separator
stri = stri[0:len(stri)-1]
return stri
def _remDuplicates(self, seq, idfun=None):
'''removes duplicated values from list'''
if idfun is None:
def idfun(x): return x
seen = {}
result = []
for item in seq:
marker = idfun(item)
if marker in seen:
self.report.addWarn('Var: "'+self.varName+'" value: "' + marker + '". Addition canceled because of duplicate entry.')
continue
seen[marker] = 1
result.append(item)
return result
def __getitem__(self, key):
return self.val[key]
def __setitem__(self, key, value):
if value in self.val:
self.report.addWarn('Var: "' + self.varName + '" value: "' + value + '". Addition canceled because of duplicate entry.')
else:
self.val.insert(key, value)
def __delitem__(self, key):
self.removeVal(self.val[key])
def __iter__(self):
for i in self.val:
yield i
def __contains__(self, item):
return item in self.val
def __len__(self):
return len(self.val)
def __str__(self):
return self._concatenate(':')
class Scalar():
'''Class for manipulating with environment scalars.'''
def __init__(self, name, local=False, report = None):
self.report = report
self.name = name
self.val = ''
self.local = local
def name(self):
'''Returns the name of the scalar.'''
return self.name
def set(self, value, separator = ':', environment={}, resolve = True):
'''Sets the value of the scalar. Any previous value is overwritten.'''
if resolve:
value = self.resolveReferences(value, environment)
self.val = value
self._changeSlashes()
if self.val == '.':
self.val = ""
def value(self, asString = False, separator = ':'):
'''Returns values of the scalar.'''
return self.val
def remove_regexp(self,value,separator = ':'):
self.remove(value, separator, True)
def remove(self, value, separator = ':', regexp=True):
'''Removes value(s) from the scalar. If value is not found, removal is canceled.'''
value = self.search(value)
for val in value:
self.val = self.val.replace(val,'')
def append(self,value, separator = ':', environment={}, warningOn=True):
'''Adds value(s) at the end of the scalar.'''
value = self.resolveReferences(value, environment)
self.val = self.val + value
self._changeSlashes()
def prepend(self,value,action='cancel', separator = ':', environment={}):
'''Adds value(s) at the beginning of the scalar.'''
value = self.resolveReferences(value, environment)
self.val = value + self.val
self._changeSlashes()
def repl(self, s, d):
v = re.compile(r"\$([A-Za-z_][A-Za-z0-9_]*)|\$\(([A-Za-z_][A-Za-z0-9_]*)\)|\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
m = v.search(s)
if m:
s = s[:m.start()] + str(d[filter(None,m.groups())[0]]) + s[m.end():]
return self.repl(s,d)
else:
return s
def resolveReferences(self, value, environment):
'''Resolve references inside the scalar.'''
value = self.repl(value, environment)
return value
def search(self, expr):
'''Searches in scalar`s values for a match'''
return re.findall(expr,self.val)
def _changeSlashes(self):
'''Changes slashes depending on operating system.'''
if self.val == '[':
if platform.system() != 'Linux':
self.val = self.val[1:]
else:
self.val = self.val[3:]
self.val = os.path.normpath(self.val)
def __str__(self):
return self.val
class EnvironmentError(Exception):
'''Class which defines errors for locals operations.'''
def __init__(self, value, code):
self.val = value
self.code = code
def __str__(self):
if self.code == 'undefined':
return 'Reference to undefined environment element: "'+self.val +'".'
elif self.code == 'ref2var':
return 'Reference to list from the middle of string.'
elif self.code == 'redeclaration':
return 'Wrong redeclaration of environment element "'+self.val+'".'