Source code for autotest.client.bkr_xml

# Copyright (C) 2011 Jan Stancek <>
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
module to parse beaker xml recipe
__author__ = """Copyright Jan Stancek 2011"""

import logging
import os
from xml.dom import minidom

log = logging

[docs]def xml_attr(node, key, default=None): try: return str(node.attributes[key].value) except Exception: return default
[docs]def xml_get_nodes(node, tag): return [n for n in node.childNodes if n.nodeName == tag]
[docs]class Recipe(object): def __init__(self): self.tasks = [] = -1 self.job_id = -1 self.exclude_dir = []
[docs]class Task(object): """ Simple record to store task properties """ def __init__(self): = '' self.params = {} self.rpmName = '' self.rpmPath = '' self.timeout = '' self.exclude_dir = [] = -1 def __str__(self): return "- %s %s" % (, str(self.params)) def __repr__(self): return "%s %s" % (, str(self.params))
[docs] def get_param(self, key, default=None): if key in self.params: return self.params[key] else: return default
[docs]class BeakerXMLParser(object): """ Handles parsing of beaker job xml """ def __init__(self): = {}
[docs] def parse_from_file(self, file_name): log.debug('BeakerXMLParser - opening file: %s', file_name) f = open(os.path.expanduser(file_name), 'r') contents = f.close() log.debug('BeakerXMLParser - content read ok') return self.parse_xml(contents)
[docs] def parse_xml(self, xml): """ Returns dict, mapping hostname to recipe """ log.debug('Parsing recipes') log.debug("xml type is %s" % type(xml)) doc = minidom.parseString(xml) recipe_nodes = doc.getElementsByTagName('recipe') self.handle_recipes(recipe_nodes) log.debug('Parsing recipes ok') return
[docs] def handle_recipes(self, recipe_nodes): for recipe_node in recipe_nodes: self.handle_recipe(recipe_node)
[docs] def handle_recipe(self, recipe_node): hostname = recipe_node.getAttribute('system') recipe = Recipe() = recipe_node.getAttribute('id') recipe.job_id = recipe_node.getAttribute('job_id') log.debug('Parsing recipe with id: <%s>', #tasks = recipe.getElementsByTagName('task') task_nodes = xml_get_nodes(recipe_node, 'task') self.handle_tasks(recipe, task_nodes)[hostname] = recipe return True
[docs] def handle_tasks(self, recipe, task_nodes): for task_node in task_nodes: self.handle_task(recipe, task_node)
[docs] def handle_task(self, recipe, task_node): task = Task() = task_node.getAttribute('name') = task_node.getAttribute('id') task.timeout = task_node.getAttribute('max_time') or task_node.getAttribute('avg_time') task.status = task_node.getAttribute('status') log.debug('Parsing task with id: <%s>', recipe.tasks.append(task) params_tags = xml_get_nodes(task_node, 'params') if params_tags: param_nodes = params_tags[0].getElementsByTagName('param') self.handle_task_params(task, param_nodes) rpm_nodes = xml_get_nodes(task_node, 'rpm') task.rpmName = rpm_nodes[0].getAttribute('name') task.rpmPath = rpm_nodes[0].getAttribute('path')
[docs] def handle_task_params(self, task, param_nodes): for param_node in param_nodes: self.handle_task_param(task, param_node)
[docs] def handle_task_param(self, task, param_node): param_name = param_node.getAttribute('name') param_value = param_node.getAttribute('value') task.params[param_name] = param_value