GaudiTesting.BaseTest Namespace Reference

Classes
class	BaseTest
class	BasicOutputValidator
class	BlockSkipper
class	FilePreprocessor
class	FilePreprocessorSequence
class	JSONOutputValidator
class	LineSkipper
class	LineSorter
class	ReferenceFileValidator
class	RegexpReplacer
class	Result
class	SortGroupOfLines

Functions
def	_new_backslashreplace_errors (exc)
def	sanitize_for_xml (data)
def	dumpProcs (name)
def	kill_tree (ppid, sig)
def	ROOT6WorkAroundEnabled (id=None)
def	RationalizePath (p)
def	which (executable)
def	findTTreeSummaries (stdout)
def	cmpTreesDicts (reference, to_check, ignore=None)
def	getCmpFailingValues (reference, to_check, fail_path)
def	_parseTTreeSummary (lines, pos)
def	parseHistosSummary (lines, pos)
def	findHistosSummaries (stdout)
def	GetPlatform (self)
def	isWinPlatform (self)

Variables
	SKIP_RETURN_CODE
	OUTPUT_LIMIT
	maskPointers
	normalizeDate
	normalizeEOL
	__processLine__
	skipEmptyLines
	normalizeTestSuite
	lineSkipper
list	regexps
	normalizeExamples
	h_count_re

Function Documentation

_new_backslashreplace_errors()

def GaudiTesting.BaseTest._new_backslashreplace_errors (   exc )

private

Definition at line 32 of file BaseTest.py.

    def _new_backslashreplace_errors(exc):
        if isinstance(exc, UnicodeDecodeError):
            code = hex(ord(exc.object[exc.start]))
            return ("\\" + code[1:], exc.start + 1)
        else:
            return backslashreplace_errors(exc)
 

_parseTTreeSummary()

def GaudiTesting.BaseTest._parseTTreeSummary (   lines,   pos  )

private

Parse the TTree summary table in lines, starting from pos.
Returns a tuple with the dictionary with the digested informations and the
position of the first line after the summary.

Definition at line 1349 of file BaseTest.py.

def _parseTTreeSummary(lines, pos):
    """
    Parse the TTree summary table in lines, starting from pos.
    Returns a tuple with the dictionary with the digested informations and the
    position of the first line after the summary.
    """
    result = {}
    i = pos + 1  # first line is a sequence of '*'
    count = len(lines)
 
    def splitcols(l):
        return [f.strip() for f in l.strip("*\n").split(":", 2)]
 
    def parseblock(ll):
        r = {}
        delta_i = 0
        cols = splitcols(ll[0])
 
        if len(ll) == 3:
            # default one line name/title
            r["Name"], r["Title"] = cols[1:]
        elif len(ll) == 4:
            # in case title is moved to next line due to too long name
            delta_i = 1
            r["Name"] = cols[1]
            r["Title"] = ll[1].strip("*\n").split("|")[1].strip()
        else:
            assert False
 
        cols = splitcols(ll[1 + delta_i])
        r["Entries"] = int(cols[1])
 
        sizes = cols[2].split()
        r["Total size"] = int(sizes[2])
        if sizes[-1] == "memory":
            r["File size"] = 0
        else:
            r["File size"] = int(sizes[-1])
 
        cols = splitcols(ll[2 + delta_i])
        sizes = cols[2].split()
        if cols[0] == "Baskets":
            r["Baskets"] = int(cols[1])
            r["Basket size"] = int(sizes[2])
        r["Compression"] = float(sizes[-1])
 
        return r
 
    def nextblock(lines, i):
        delta_i = 1
        dots = re.compile(r"^\.+$")
        stars = re.compile(r"^\*+$")
        count = len(lines)
        while (
            i + delta_i < count
            and not dots.match(lines[i + delta_i][1:-1])
            and not stars.match(lines[i + delta_i])
        ):
            delta_i += 1
        return i + delta_i
 
    if i < (count - 3) and lines[i].startswith("*Tree"):
        i_nextblock = nextblock(lines, i)
        result = parseblock(lines[i:i_nextblock])
        result["Branches"] = {}
        i = i_nextblock + 1
        while i < (count - 3) and lines[i].startswith("*Br"):
            if i < (count - 2) and lines[i].startswith("*Branch "):
                # skip branch header
                i += 3
                continue
            i_nextblock = nextblock(lines, i)
            if i_nextblock >= count:
                break
            branch = parseblock(lines[i:i_nextblock])
            result["Branches"][branch["Name"]] = branch
            i = i_nextblock + 1
 
    return (result, i)
 
 

cmpTreesDicts()

def GaudiTesting.BaseTest.cmpTreesDicts	(		reference,
			to_check,
			ignore = None
	)

Check that all the keys in reference are in to_check too, with the same value.
If the value is a dict, the function is called recursively. to_check can
contain more keys than reference, that will not be tested.
The function returns at the first difference found.

Definition at line 1301 of file BaseTest.py.

def cmpTreesDicts(reference, to_check, ignore=None):
    """
    Check that all the keys in reference are in to_check too, with the same value.
    If the value is a dict, the function is called recursively. to_check can
    contain more keys than reference, that will not be tested.
    The function returns at the first difference found.
    """
    fail_keys = []
    # filter the keys in the reference dictionary
    if ignore:
        ignore_re = re.compile(ignore)
        keys = [key for key in reference if not ignore_re.match(key)]
    else:
        keys = reference.keys()
    # loop over the keys (not ignored) in the reference dictionary
    for k in keys:
        if k in to_check:  # the key must be in the dictionary to_check
            if isinstance(reference[k], dict) and isinstance(to_check[k], dict):
                # if both reference and to_check values are dictionaries,
                # recurse
                failed = fail_keys = cmpTreesDicts(reference[k], to_check[k], ignore)
            else:
                # compare the two values
                failed = to_check[k] != reference[k]
        else:  # handle missing keys in the dictionary to check (i.e. failure)
            to_check[k] = None
            failed = True
        if failed:
            fail_keys.insert(0, k)
            break  # exit from the loop at the first failure
    return fail_keys  # return the list of keys bringing to the different values
 
 

dumpProcs()

def GaudiTesting.BaseTest.dumpProcs

(

name

)

helper to debug GAUDI-1084, dump the list of processes

Definition at line 67 of file BaseTest.py.

def dumpProcs(name):
    """helper to debug GAUDI-1084, dump the list of processes"""
    from getpass import getuser
 
    if "WORKSPACE" in os.environ:
        p = Popen(["ps", "-fH", "-U", getuser()], stdout=PIPE)
        with open(os.path.join(os.environ["WORKSPACE"], name), "wb") as f:
            f.write(p.communicate()[0])
 
 

findHistosSummaries()

def GaudiTesting.BaseTest.findHistosSummaries

(

stdout

)

Scan stdout to find ROOT TTree summaries and digest them.

Definition at line 1498 of file BaseTest.py.

def findHistosSummaries(stdout):
    """
    Scan stdout to find ROOT TTree summaries and digest them.
    """
    outlines = stdout.splitlines()
    nlines = len(outlines) - 1
    summaries = {}
    global h_count_re
 
    pos = 0
    while pos < nlines:
        summ = {}
        # find first line of block:
        match = h_count_re.search(outlines[pos])
        while pos < nlines and not match:
            pos += 1
            match = h_count_re.search(outlines[pos])
        if match:
            summ, pos = parseHistosSummary(outlines, pos)
        summaries.update(summ)
    return summaries
 
 

findTTreeSummaries()

def GaudiTesting.BaseTest.findTTreeSummaries

(

stdout

)

Scan stdout to find ROOT TTree summaries and digest them.

Definition at line 1279 of file BaseTest.py.

def findTTreeSummaries(stdout):
    """
    Scan stdout to find ROOT TTree summaries and digest them.
    """
    stars = re.compile(r"^\*+$")
    outlines = stdout.splitlines()
    nlines = len(outlines)
    trees = {}
 
    i = 0
    while i < nlines:  # loop over the output
        # look for
        while i < nlines and not stars.match(outlines[i]):
            i += 1
        if i < nlines:
            tree, i = _parseTTreeSummary(outlines, i)
            if tree:
                trees[tree["Name"]] = tree
 
    return trees
 
 

getCmpFailingValues()

def GaudiTesting.BaseTest.getCmpFailingValues	(		reference,
			to_check,
			fail_path
	)

Definition at line 1334 of file BaseTest.py.

def getCmpFailingValues(reference, to_check, fail_path):
    c = to_check
    r = reference
    for k in fail_path:
        c = c.get(k, None)
        r = r.get(k, None)
        if c is None or r is None:
            break  # one of the dictionaries is not deep enough
    return (fail_path, r, c)
 
 
# signature of the print-out of the histograms

GetPlatform()

def GaudiTesting.BaseTest.GetPlatform

(

self

)

Return the platform Id defined in CMTCONFIG or SCRAM_ARCH.

Definition at line 1521 of file BaseTest.py.

def GetPlatform(self):
    """
    Return the platform Id defined in CMTCONFIG or SCRAM_ARCH.
    """
    arch = "None"
    # check architecture name
    if "BINARY_TAG" in os.environ:
        arch = os.environ["BINARY_TAG"]
    elif "CMTCONFIG" in os.environ:
        arch = os.environ["CMTCONFIG"]
    elif "SCRAM_ARCH" in os.environ:
        arch = os.environ["SCRAM_ARCH"]
    elif os.environ.get("ENV_CMAKE_BUILD_TYPE", "") in (
        "Debug",
        "FastDebug",
        "Developer",
    ):
        arch = "dummy-dbg"
    elif os.environ.get("ENV_CMAKE_BUILD_TYPE", "") in (
        "Release",
        "MinSizeRel",
        "RelWithDebInfo",
        "",
    ):  # RelWithDebInfo == -O2 -g -DNDEBUG
        arch = "dummy-opt"
    return arch
 
 

isWinPlatform()

def GaudiTesting.BaseTest.isWinPlatform

(

self

)

Return True if the current platform is Windows.

This function was needed because of the change in the CMTCONFIG format,
from win32_vc71_dbg to i686-winxp-vc9-dbg.

Definition at line 1549 of file BaseTest.py.

def isWinPlatform(self):
    """
    Return True if the current platform is Windows.
 
    This function was needed because of the change in the CMTCONFIG format,
    from win32_vc71_dbg to i686-winxp-vc9-dbg.
    """
    platform = GetPlatform(self)
    return "winxp" in platform or platform.startswith("win")
 
 

kill_tree()

def GaudiTesting.BaseTest.kill_tree	(		ppid,
			sig
	)

Send a signal to a process and all its child processes (starting from the
leaves).

Definition at line 77 of file BaseTest.py.

def kill_tree(ppid, sig):
    """
    Send a signal to a process and all its child processes (starting from the
    leaves).
    """
    log = logging.getLogger("kill_tree")
    ps_cmd = ["ps", "--no-headers", "-o", "pid", "--ppid", str(ppid)]
    # Note: start in a clean env to avoid a freeze with libasan.so
    # See https://sourceware.org/bugzilla/show_bug.cgi?id=27653
    get_children = Popen(ps_cmd, stdout=PIPE, stderr=PIPE, env={})
    children = map(int, get_children.communicate()[0].split())
    for child in children:
        kill_tree(child, sig)
    try:
        log.debug("killing process %d", ppid)
        os.kill(ppid, sig)
    except OSError as err:
        if err.errno != 3:  # No such process
            raise
        log.debug("no such process %d", ppid)
 
 
# -------------------------------------------------------------------------#
 
 

parseHistosSummary()

def GaudiTesting.BaseTest.parseHistosSummary	(		lines,
			pos
	)

Extract the histograms infos from the lines starting at pos.
Returns the position of the first line after the summary block.

Definition at line 1430 of file BaseTest.py.

def parseHistosSummary(lines, pos):
    """
    Extract the histograms infos from the lines starting at pos.
    Returns the position of the first line after the summary block.
    """
    global h_count_re
    h_table_head = re.compile(
        r'(?:INFO|SUCCESS)\s+(1D|2D|3D|1D profile|2D profile|3d profile) histograms in directory\s+"(\w*)"'
    )
    h_short_summ = re.compile(r"ID=([^\"]+)\s+\"([^\"]*)\"\s+(.*)")
 
    nlines = len(lines)
 
    # decode header
    m = h_count_re.search(lines[pos])
    name = m.group(1).strip()
    total = int(m.group(2))
    header = {}
    for k, v in [x.split("=") for x in m.group(3).split()]:
        header[k] = int(v)
    pos += 1
    header["Total"] = total
 
    summ = {}
    while pos < nlines:
        m = h_table_head.search(lines[pos])
        if m:
            t, d = m.groups(1)  # type and directory
            t = t.replace(" profile", "Prof")
            pos += 1
            if pos < nlines:
                l = lines[pos]
            else:
                l = ""
            cont = {}
            if l.startswith(" | ID"):
                # table format
                titles = [x.strip() for x in l.split("|")][1:]
                pos += 1
                while pos < nlines and lines[pos].startswith(" |"):
                    l = lines[pos]
                    values = [x.strip() for x in l.split("|")][1:]
                    hcont = {}
                    for i in range(len(titles)):
                        hcont[titles[i]] = values[i]
                    cont[hcont["ID"]] = hcont
                    pos += 1
            elif l.startswith(" ID="):
                while pos < nlines and lines[pos].startswith(" ID="):
                    values = [
                        x.strip() for x in h_short_summ.search(lines[pos]).groups()
                    ]
                    cont[values[0]] = values
                    pos += 1
            else:  # not interpreted
                raise RuntimeError("Cannot understand line %d: '%s'" % (pos, l))
            if d not in summ:
                summ[d] = {}
            summ[d][t] = cont
            summ[d]["header"] = header
        else:
            break
    if not summ:
        # If the full table is not present, we use only the header
        summ[name] = {"header": header}
    return summ, pos
 
 

RationalizePath()

def GaudiTesting.BaseTest.RationalizePath

(

p

)

Function used to normalize the used path

Definition at line 781 of file BaseTest.py.

def RationalizePath(p):
    """
    Function used to normalize the used path
    """
    p = os.path.expandvars(p)
 
    # handle the special case "path/to/file:some_suffix"
    suffix = ""
    if ":" in p:
        p, suffix = p.rsplit(":", 1)
        suffix = f":{suffix}"
 
    if os.path.exists(p):
        p = os.path.realpath(p)
    return p + suffix
 
 

ROOT6WorkAroundEnabled()

def GaudiTesting.BaseTest.ROOT6WorkAroundEnabled

(

id = None

)

Definition at line 773 of file BaseTest.py.

    def ROOT6WorkAroundEnabled(id=None):
        # dummy implementation
        return False
 
 
# --------------------------------- TOOLS ---------------------------------#
 
 

sanitize_for_xml()

def GaudiTesting.BaseTest.sanitize_for_xml

(

data

)

Take a string with invalid ASCII/UTF characters and quote them so that the
string can be used in an XML text.

>>> sanitize_for_xml('this is \x1b')
'this is [NON-XML-CHAR-0x1B]'

Definition at line 50 of file BaseTest.py.

def sanitize_for_xml(data):
    """
    Take a string with invalid ASCII/UTF characters and quote them so that the
    string can be used in an XML text.
 
    >>> sanitize_for_xml('this is \x1b')
    'this is [NON-XML-CHAR-0x1B]'
    """
    bad_chars = re.compile("[\x00-\x08\x0b\x0c\x0e-\x1f\ud800-\udfff\ufffe\uffff]")
 
    def quote(match):
        "helper function"
        return "".join("[NON-XML-CHAR-0x%2X]" % ord(c) for c in match.group())
 
    return bad_chars.sub(quote, data)
 
 

which()

def GaudiTesting.BaseTest.which

(

executable

)

Locates an executable in the executables path ($PATH) and returns the full
path to it.  An application is looked for with or without the '.exe' suffix.
If the executable cannot be found, None is returned

Definition at line 798 of file BaseTest.py.

def which(executable):
    """
    Locates an executable in the executables path ($PATH) and returns the full
    path to it.  An application is looked for with or without the '.exe' suffix.
    If the executable cannot be found, None is returned
    """
    if os.path.isabs(executable):
        if not os.path.isfile(executable):
            if executable.endswith(".exe"):
                if os.path.isfile(executable[:-4]):
                    return executable[:-4]
            else:
                executable = os.path.split(executable)[1]
        else:
            return executable
    for d in os.environ.get("PATH").split(os.pathsep):
        fullpath = os.path.join(d, executable)
        if os.path.isfile(fullpath):
            return fullpath
        elif executable.endswith(".exe") and os.path.isfile(fullpath[:-4]):
            return fullpath[:-4]
    return None
 
 
# -------------------------------------------------------------------------#
# ----------------------------- Result Classe -----------------------------#
# -------------------------------------------------------------------------#
 
 

Variable Documentation

__processLine__

GaudiTesting.BaseTest.__processLine__

private

Definition at line 1024 of file BaseTest.py.

h_count_re

GaudiTesting.BaseTest.h_count_re

Definition at line 1346 of file BaseTest.py.

lineSkipper

GaudiTesting.BaseTest.lineSkipper

Definition at line 1119 of file BaseTest.py.

maskPointers

GaudiTesting.BaseTest.maskPointers

Definition at line 1018 of file BaseTest.py.

normalizeDate

GaudiTesting.BaseTest.normalizeDate

Definition at line 1019 of file BaseTest.py.

normalizeEOL

GaudiTesting.BaseTest.normalizeEOL

Definition at line 1023 of file BaseTest.py.

normalizeExamples

GaudiTesting.BaseTest.normalizeExamples

Definition at line 1232 of file BaseTest.py.

normalizeTestSuite

GaudiTesting.BaseTest.normalizeTestSuite

Definition at line 1074 of file BaseTest.py.

OUTPUT_LIMIT

GaudiTesting.BaseTest.OUTPUT_LIMIT

Definition at line 47 of file BaseTest.py.

regexps

list GaudiTesting.BaseTest.regexps

Definition at line 1161 of file BaseTest.py.

SKIP_RETURN_CODE

GaudiTesting.BaseTest.SKIP_RETURN_CODE

Definition at line 44 of file BaseTest.py.

skipEmptyLines

GaudiTesting.BaseTest.skipEmptyLines

Definition at line 1026 of file BaseTest.py.