#!/usr/bin/python
"""
Cartesian configuration format file parser.
Filter syntax:
, means OR
.. means AND
. means IMMEDIATELY-FOLLOWED-BY
Example:
qcow2..Fedora.14, RHEL.6..raw..boot, smp2..qcow2..migrate..ide
means match all dicts whose names have:
(qcow2 AND (Fedora IMMEDIATELY-FOLLOWED-BY 14)) OR
((RHEL IMMEDIATELY-FOLLOWED-BY 6) AND raw AND boot) OR
(smp2 AND qcow2 AND migrate AND ide)
Note:
'qcow2..Fedora.14' is equivalent to 'Fedora.14..qcow2'.
'qcow2..Fedora.14' is not equivalent to 'qcow2..14.Fedora'.
'ide, scsi' is equivalent to 'scsi, ide'.
Filters can be used in 3 ways:
only <filter>
no <filter>
<filter>:
The last one starts a conditional block.
@copyright: Red Hat 2008-2011
"""
import re, os, sys, optparse, collections
class ParserError:
def __init__(self, msg, line=None, filename=None, linenum=None):
self.msg = msg
self.line = line
self.filename = filename
self.linenum = linenum
def __str__(self):
if self.line:
return "%s: %r (%s:%s)" % (self.msg, self.line,
self.filename, self.linenum)
else:
return "%s (%s:%s)" % (self.msg, self.filename, self.linenum)
num_failed_cases = 5
class Node(object):
def __init__(self):
self.name = []
self.dep = []
self.content = []
self.children = []
self.labels = set()
self.append_to_shortname = False
self.failed_cases = collections.deque()
def _match_adjacent(block, ctx, ctx_set):
# TODO: explain what this function does
if block[0] not in ctx_set:
return 0
if len(block) == 1:
return 1
if block[1] not in ctx_set:
return int(ctx[-1] == block[0])
k = 0
i = ctx.index(block[0])
while i < len(ctx):
if k > 0 and ctx[i] != block[k]:
i -= k - 1
k = 0
if ctx[i] == block[k]:
k += 1
if k >= len(block):
break
if block[k] not in ctx_set:
break
i += 1
return k
def _might_match_adjacent(block, ctx, ctx_set, descendant_labels):
matched = _match_adjacent(block, ctx, ctx_set)
for elem in block[matched:]:
if elem not in descendant_labels:
return False
return True
# Filter must inherit from object (otherwise type() won't work)
class Filter(object):
def __init__(self, s):
self.filter = []
for char in s:
if not (char.isalnum() or char.isspace() or char in ".,_-"):
raise ParserError("Illegal characters in filter")
for word in s.replace(",", " ").split():
word = [block.split(".") for block in word.split("..")]
for block in word:
for elem in block:
if not elem:
raise ParserError("Syntax error")
self.filter += [word]
def match(self, ctx, ctx_set):
for word in self.filter:
for block in word:
if _match_adjacent(block, ctx, ctx_set) != len(block):
break
else:
return True
return False
def might_match(self, ctx, ctx_set, descendant_labels):
for word in self.filter:
for block in word:
if not _might_match_adjacent(block, ctx, ctx_set,
descendant_labels):
break
else:
return True
return False
class NoOnlyFilter(Filter):
def __init__(self, line):
Filter.__init__(self, line.split(None, 1)[1])
self.line = line
class OnlyFilter(NoOnlyFilter):
def is_irrelevant(self, ctx, ctx_set, descendant_labels):
return self.match(ctx, ctx_set)
def requires_action(self, ctx, ctx_set, descendant_labels):
return not self.might_match(ctx, ctx_set, descendant_labels)
def might_pass(self, failed_ctx, failed_ctx_set, ctx, ctx_set,
descendant_labels):
for word in self.filter:
for block in word:
if (_match_adjacent(block, ctx, ctx_set) >
_match_adjacent(block, failed_ctx, failed_ctx_set)):
return self.might_match(ctx, ctx_set, descendant_labels)
return False
class NoFilter(NoOnlyFilter):
def is_irrelevant(self, ctx, ctx_set, descendant_labels):
return not self.might_match(ctx, ctx_set, descendant_labels)
def requires_action(self, ctx, ctx_set, descendant_labels):
return self.match(ctx, ctx_set)
def might_pass(self, failed_ctx, failed_ctx_set, ctx, ctx_set,
descendant_labels):
for word in self.filter:
for block in word:
if (_match_adjacent(block, ctx, ctx_set) <
_match_adjacent(block, failed_ctx, failed_ctx_set)):
return not self.match(ctx, ctx_set)
return False
class Condition(NoFilter):
def __init__(self, line):
Filter.__init__(self, line.rstrip(":"))
self.line = line
self.content = []
class NegativeCondition(OnlyFilter):
def __init__(self, line):
Filter.__init__(self, line.lstrip("!").rstrip(":"))
self.line = line
self.content = []
class Parser(object):
"""
Parse an input file or string that follows the Cartesian Config File format
and generate a list of dicts that will be later used as configuration
parameters by autotest tests that use that format.
@see: http://autotest.kernel.org/wiki/CartesianConfig
"""
def __init__(self, filename=None, debug=False):
"""
Initialize the parser and optionally parse a file.
@param filename: Path of the file to parse.
@param debug: Whether to turn on debugging output.
"""
self.node = Node()
self.debug = debug
if filename:
self.parse_file(filename)
def parse_file(self, filename):
"""
Parse a file.
@param filename: Path of the configuration file.
"""
self.node = self._parse(FileReader(filename), self.node)
def parse_string(self, s):
"""
Parse a string.
@param s: String to parse.
"""
self.node = self._parse(StrReader(s), self.node)
def get_dicts(self, node=None, ctx=[], content=[], shortname=[], dep=[]):
"""
Generate dictionaries from the code parsed so far. This should
be called after parsing something.
@return: A dict generator.
"""
def process_content(content, failed_filters):
# 1. Check that the filters in content are OK with the current
# context (ctx).
# 2. Move the parts of content that are still relevant into
# new_content and unpack conditional blocks if appropriate.
# For example, if an 'only' statement fully matches ctx, it
# becomes irrelevant and is not appended to new_content.
# If a conditional block fully matches, its contents are
# unpacked into new_content.
# 3. Move failed filters into failed_filters, so that next time we
# reach this node or one of its ancestors, we'll check those
# filters first.
for t in content:
filename, linenum, obj = t
if type(obj) is Op:
new_content.append(t)
continue
# obj is an OnlyFilter/NoFilter/Condition/NegativeCondition
if obj.requires_action(ctx, ctx_set, labels):
# This filter requires action now
if type(obj) is OnlyFilter or type(obj) is NoFilter:
self._debug(" filter did not pass: %r (%s:%s)",
obj.line, filename, linenum)
failed_filters.append(t)
return False
else:
self._debug(" conditional block matches: %r (%s:%s)",
obj.line, filename, linenum)
# Check and unpack the content inside this Condition
# object (note: the failed filters should go into
# new_internal_filters because we don't expect them to
# come from outside this node, even if the Condition
# itself was external)
if not process_content(obj.content,
new_internal_filters):
failed_filters.append(t)
return False
continue
elif obj.is_irrelevant(ctx, ctx_set, labels):
# This filter is no longer relevant and can be removed
continue
else:
# Keep the filter and check it again later
new_content.append(t)
return True
def might_pass(failed_ctx,
failed_ctx_set,
failed_external_filters,
failed_internal_filters):
for t in failed_external_filters:
if t not in content:
return True
filename, linenum, filter = t
if filter.might_pass(failed_ctx, failed_ctx_set, ctx, ctx_set,
labels):
return True
for t in failed_internal_filters:
filename, linenum, filter = t
if filter.might_pass(failed_ctx, failed_ctx_set, ctx, ctx_set,
labels):
return True
return False
def add_failed_case():
node.failed_cases.appendleft((ctx, ctx_set,
new_external_filters,
new_internal_filters))
if len(node.failed_cases) > num_failed_cases:
node.failed_cases.pop()
node = node or self.node
# Update dep
for d in node.dep:
dep = dep + [".".join(ctx + [d])]
# Update ctx
ctx = ctx + node.name
ctx_set = set(ctx)
labels = node.labels
# Get the current name
name = ".".join(ctx)
if node.name:
self._debug("checking out %r", name)
# Check previously failed filters
for i, failed_case in enumerate(node.failed_cases):
if not might_pass(*failed_case):
self._debug(" this subtree has failed before")
del node.failed_cases[i]
node.failed_cases.appendleft(failed_case)
return
# Check content and unpack it into new_content
new_content = []
new_external_filters = []
new_internal_filters = []
if (not process_content(node.content, new_internal_filters) or
not process_content(content, new_external_filters)):
add_failed_case()
return
# Update shortname
if node.append_to_shortname:
shortname = shortname + node.name
# Recurse into children
count = 0
for n in node.children:
for d in self.get_dicts(n, ctx, new_content, shortname, dep):
count += 1
yield d
# Reached leaf?
if not node.children:
self._debug(" reached leaf, returning it")
d = {"name": name, "dep": dep, "shortname": ".".join(shortname)}
for filename, linenum, op in new_content:
op.apply_to_dict(d)
yield d
# If this node did not produce any dicts, remember the failed filters
# of its descendants
elif not count:
new_external_filters = []
new_internal_filters = []
for n in node.children:
(failed_ctx,
failed_ctx_set,
failed_external_filters,
failed_internal_filters) = n.failed_cases[0]
for obj in failed_internal_filters:
if obj not in new_internal_filters:
new_internal_filters.append(obj)
for obj in failed_external_filters:
if obj in content:
if obj not in new_external_filters:
new_external_filters.append(obj)
else:
if obj not in new_internal_filters:
new_internal_filters.append(obj)
add_failed_case()
def _debug(self, s, *args):
if self.debug:
s = "DEBUG: %s" % s
print s % args
def _warn(self, s, *args):
s = "WARNING: %s" % s
print s % args
def _parse_variants(self, cr, node, prev_indent=-1):
"""
Read and parse lines from a FileReader object until a line with an
indent level lower than or equal to prev_indent is encountered.
@param cr: A FileReader/StrReader object.
@param node: A node to operate on.
@param prev_indent: The indent level of the "parent" block.
@return: A node object.
"""
node4 = Node()
while True:
line, indent, linenum = cr.get_next_line(prev_indent)
if not line:
break
name, dep = map(str.strip, line.lstrip("- ").split(":", 1))
for char in name:
if not (char.isalnum() or char in "@._-"):
raise ParserError("Illegal characters in variant name",
line, cr.filename, linenum)
for char in dep:
if not (char.isalnum() or char.isspace() or char in ".,_-"):
raise ParserError("Illegal characters in dependencies",
line, cr.filename, linenum)
node2 = Node()
node2.children = [node]
node2.labels = node.labels
node3 = self._parse(cr, node2, prev_indent=indent)
node3.name = name.lstrip("@").split(".")
node3.dep = dep.replace(",", " ").split()
node3.append_to_shortname = not name.startswith("@")
node4.children += [node3]
node4.labels.update(node3.labels)
node4.labels.update(node3.name)
return node4
def _parse(self, cr, node, prev_indent=-1):
"""
Read and parse lines from a StrReader object until a line with an
indent level lower than or equal to prev_indent is encountered.
@param cr: A FileReader/StrReader object.
@param node: A Node or a Condition object to operate on.
@param prev_indent: The indent level of the "parent" block.
@return: A node object.
"""
while True:
line, indent, linenum = cr.get_next_line(prev_indent)
if not line:
break
words = line.split(None, 1)
# Parse 'variants'
if line == "variants:":
# 'variants' is not allowed inside a conditional block
if (isinstance(node, Condition) or
isinstance(node, NegativeCondition)):
raise ParserError("'variants' is not allowed inside a "
"conditional block",
None, cr.filename, linenum)
node = self._parse_variants(cr, node, prev_indent=indent)
continue
# Parse 'include' statements
if words[0] == "include":
if len(words) < 2:
raise ParserError("Syntax error: missing parameter",
line, cr.filename, linenum)
filename = os.path.expanduser(words[1])
if isinstance(cr, FileReader) and not os.path.isabs(filename):
filename = os.path.join(os.path.dirname(cr.filename),
filename)
if not os.path.isfile(filename):
self._warn("%r (%s:%s): file doesn't exist or is not a "
"regular file", line, cr.filename, linenum)
continue
node = self._parse(FileReader(filename), node)
continue
# Parse 'only' and 'no' filters
if words[0] in ("only", "no"):
if len(words) < 2:
raise ParserError("Syntax error: missing parameter",
line, cr.filename, linenum)
try:
if words[0] == "only":
f = OnlyFilter(line)
elif words[0] == "no":
f = NoFilter(line)
except ParserError, e:
e.line = line
e.filename = cr.filename
e.linenum = linenum
raise
node.content += [(cr.filename, linenum, f)]
continue
# Look for operators
op_match = _ops_exp.search(line)
# Parse conditional blocks
if ":" in line:
index = line.index(":")
if not op_match or index < op_match.start():
index += 1
cr.set_next_line(line[index:], indent, linenum)
line = line[:index]
try:
if line.startswith("!"):
cond = NegativeCondition(line)
else:
cond = Condition(line)
except ParserError, e:
e.line = line
e.filename = cr.filename
e.linenum = linenum
raise
self._parse(cr, cond, prev_indent=indent)
node.content += [(cr.filename, linenum, cond)]
continue
# Parse regular operators
if not op_match:
raise ParserError("Syntax error", line, cr.filename, linenum)
node.content += [(cr.filename, linenum, Op(line, op_match))]
return node
# Assignment operators
_reserved_keys = set(("name", "shortname", "dep"))
def _op_set(d, key, value):
if key not in _reserved_keys:
d[key] = value
def _op_append(d, key, value):
if key not in _reserved_keys:
d[key] = d.get(key, "") + value
def _op_prepend(d, key, value):
if key not in _reserved_keys:
d[key] = value + d.get(key, "")
def _op_regex_set(d, exp, value):
exp = re.compile("%s$" % exp)
for key in d:
if key not in _reserved_keys and exp.match(key):
d[key] = value
def _op_regex_append(d, exp, value):
exp = re.compile("%s$" % exp)
for key in d:
if key not in _reserved_keys and exp.match(key):
d[key] += value
def _op_regex_prepend(d, exp, value):
exp = re.compile("%s$" % exp)
for key in d:
if key not in _reserved_keys and exp.match(key):
d[key] = value + d[key]
def _op_regex_del(d, empty, exp):
exp = re.compile("%s$" % exp)
for key in d.keys():
if key not in _reserved_keys and exp.match(key):
del d[key]
_ops = {"=": (r"\=", _op_set),
"+=": (r"\+\=", _op_append),
"<=": (r"\<\=", _op_prepend),
"?=": (r"\?\=", _op_regex_set),
"?+=": (r"\?\+\=", _op_regex_append),
"?<=": (r"\?\<\=", _op_regex_prepend),
"del": (r"^del\b", _op_regex_del)}
_ops_exp = re.compile("|".join([op[0] for op in _ops.values()]))
class Op(object):
def __init__(self, line, m):
self.func = _ops[m.group()][1]
self.key = line[:m.start()].strip()
value = line[m.end():].strip()
if value and (value[0] == value[-1] == '"' or
value[0] == value[-1] == "'"):
value = value[1:-1]
self.value = value
def apply_to_dict(self, d):
self.func(d, self.key, self.value)
# StrReader and FileReader
class StrReader(object):
"""
Preprocess an input string for easy reading.
"""
def __init__(self, s):
"""
Initialize the reader.
@param s: The string to parse.
"""
self.filename = "<string>"
self._lines = []
self._line_index = 0
self._stored_line = None
for linenum, line in enumerate(s.splitlines()):
line = line.rstrip().expandtabs()
stripped_line = line.lstrip()
indent = len(line) - len(stripped_line)
if (not stripped_line
or stripped_line.startswith("#")
or stripped_line.startswith("//")):
continue
self._lines.append((stripped_line, indent, linenum + 1))
def get_next_line(self, prev_indent):
"""
Get the next line in the current block.
@param prev_indent: The indentation level of the previous block.
@return: (line, indent, linenum), where indent is the line's
indentation level. If no line is available, (None, -1, -1) is
returned.
"""
if self._stored_line:
ret = self._stored_line
self._stored_line = None
return ret
if self._line_index >= len(self._lines):
return None, -1, -1
line, indent, linenum = self._lines[self._line_index]
if indent <= prev_indent:
return None, -1, -1
self._line_index += 1
return line, indent, linenum
def set_next_line(self, line, indent, linenum):
"""
Make the next call to get_next_line() return the given line instead of
the real next line.
"""
line = line.strip()
if line:
self._stored_line = line, indent, linenum
class FileReader(StrReader):
"""
Preprocess an input file for easy reading.
"""
def __init__(self, filename):
"""
Initialize the reader.
@parse filename: The name of the input file.
"""
StrReader.__init__(self, open(filename).read())
self.filename = filename
if __name__ == "__main__":
parser = optparse.OptionParser('usage: %prog [options] filename '
'[extra code] ...\n\nExample:\n\n '
'%prog tests.cfg "only my_set" "no qcow2"')
parser.add_option("-v", "--verbose", dest="debug", action="store_true",
help="include debug messages in console output")
parser.add_option("-f", "--fullname", dest="fullname", action="store_true",
help="show full dict names instead of short names")
parser.add_option("-c", "--contents", dest="contents", action="store_true",
help="show dict contents")
options, args = parser.parse_args()
if not args:
parser.error("filename required")
c = Parser(args[0], debug=options.debug)
for s in args[1:]:
c.parse_string(s)
for i, d in enumerate(c.get_dicts()):
if options.fullname:
print "dict %4d: %s" % (i + 1, d["name"])
else:
print "dict %4d: %s" % (i + 1, d["shortname"])
if options.contents:
keys = d.keys()
keys.sort()
for key in keys:
print " %s = %s" % (key, d[key])