« back to all changes in this revision
Viewing changes to hwtest/question.py
- browse files at revision 40
- compare with another revision
- download diff
- download tarball
- view history from revision 40
- Committer: Marc Tardif
- Date: 2007-10-04 23:12:10 UTC
- Revision ID: marc.tardif@canonical.com-20071004231210-unxckndkgndxfdp6
Refactored questions to use templates and scripts which fixes bug #149195.
- files added:
- debian/hwtest.config
- examples
- questions
- scripts
- files removed:
- plugins/question_info.py
- files renamed:
- hwtest/template.py => hwtest/lib/template.py
- files modified:
- cli/hwtest-cli
added
removed
hwtest/question.py
1
import os
1
2
import re
3
import types
4
import logging
2
5
3
6
from hwtest.excluder import Excluder
4
7
from hwtest.iterator import Iterator
5
8
from hwtest.repeater import PreRepeater
6
9
from hwtest.resolver import Resolver
7
from hwtest.plugin import Plugin
8
10
9
11
from hwtest.answer import Answer, NO, SKIP
10
from hwtest.template import convert_string
12
from hwtest.lib.file import reader
11
13
12
from hwtest.report_helpers import createElement, createTypedElement
13
14
14
15
DESKTOP = 'desktop'
15
16
LAPTOP = 'laptop'
17
18
ALL_CATEGORIES = [DESKTOP, LAPTOP, SERVER]
18
19
19
20
21
class QuestionParser(object):
22
23
def __init__(self):
24
self.questions = []
25
26
def load_data(self, **data):
27
if "name" not in data:
28
raise Exception, \
29
"Question data does not contain a 'name': %s" % data
30
31
logging.info("Loading question data for: %s", data["name"])
32
33
if filter(lambda q: q["name"] == data["name"], self.questions):
34
raise Exception, \
35
"Question %s already has a question of the same name." \
36
% data["name"]
37
38
self.questions.append(data)
39
40
def load_path(self, path):
41
logging.info("Loading question from path: %s", path)
42
43
fd = file(path, "r")
44
for string in reader(fd):
45
data = {}
46
47
def save(field, value, extended, path):
48
if value and extended:
49
raise Exception, \
50
"Path %s has both a value and an extended value." % path
51
extended = extended.rstrip("\n")
52
if field:
53
if data.has_key(field):
54
raise Exception, \
55
"Path %s has a duplicate field '%s' with a new value '%s'." \
56
% (path, field, value)
57
data[field] = value or extended
58
59
string = string.strip("\n")
60
field = value = extended = ''
61
for line in string.split("\n"):
62
line.strip()
63
match = re.search(r"^([-_.A-Za-z0-9]*):\s?(.*)", line)
64
if match:
65
save(field, value, extended, path)
66
field = match.groups()[0].lower()
67
value = match.groups()[1].rstrip()
68
extended = ''
69
basefield = re.sub(r"-.+$", "", field)
70
continue
71
72
if re.search(r"^\s\.$", line):
73
extended += "\n\n"
74
continue
75
76
match = re.search(r"^\s(\s+.*)", line)
77
if match:
78
bit = match.groups()[0].rstrip()
79
if len(extended) and not re.search(r"[\n ]$", extended):
80
extended += "\n"
81
82
extended += bit + "\n"
83
continue
84
85
match = re.search(r"^\s(.*)", line)
86
if match:
87
bit = match.groups()[0].rstrip()
88
if len(extended) and not re.search(r"[\n ]$", extended):
89
extended += " "
90
91
extended += bit
92
continue
93
94
raise Exception, "Path %s parse error at: %s" \
95
% (path, line)
96
97
save(field, value, extended, path)
98
self.load_data(**data)
99
100
def load_directory(self, directory):
101
logging.info("Loading questions from directory: %s", directory)
102
for name in [name for name in os.listdir(directory)
103
if name.endswith(".txt")]:
104
path = os.path.join(directory, name)
105
self.load_path(path)
106
107
20
108
class QuestionManager(object):
109
21
110
def __init__(self):
22
self.questions = []
111
self._questions = []
23
112
24
def add(self, question):
25
self.questions.append(question)
113
def add_question(self, question):
114
self._questions.append(question)
26
115
27
116
def get_iterator(self):
28
117
def repeat_func(question, resolver):
29
118
answer = question.answer
30
119
if answer and (answer.status == NO or answer.status == SKIP):
31
120
for dependent in resolver.get_dependents(question):
32
dependent.create_answer(SKIP, auto=True)
121
dependent.set_answer(SKIP, auto=True)
33
122
34
123
def exclude_next_func(question):
35
124
return question.answer != None
42
131
return False
43
132
44
133
resolver = Resolver()
45
question_dict = dict((question.name, question) for question in self.questions)
46
for question in self.questions:
47
question_deps = [question_dict[dep] for dep in question.deps]
48
resolver.add(question, *question_deps)
134
question_dict = dict((question.name, question) for question in self._questions)
135
for question in self._questions:
136
question_depends = [question_dict[d] for d in question.depends]
137
resolver.add(question, *question_depends)
49
138
50
139
questions = resolver.get_dependents()
51
140
questions_iter = Iterator(questions)
59
148
60
149
class Question(object):
61
150
62
def __init__(self, name, desc, deps=[], cats=ALL_CATEGORIES, optional=False, command=None):
63
self.name = self.persist_name = name
64
self.desc = desc
65
self.deps = deps
66
self.cats = cats
67
self.optional = optional
68
self.command = command
151
required_fields = ["name", "description"]
152
optional_fields = {
153
"categories": ALL_CATEGORIES,
154
"depends": [],
155
"command": None,
156
"optional": False}
157
158
def __init__(self, **kwargs):
159
self.data = kwargs
69
160
self.answer = None
161
self._validate()
162
163
def _validate(self):
164
for field in self.data.keys():
165
if field not in self.required_fields + self.optional_fields.keys():
166
raise Exception, \
167
"Question data contains unknown field: %s" \
168
% field
169
170
for field in self.required_fields:
171
if not self.data.has_key(field):
172
raise Exception, \
173
"Question data does not contain a '%s': %s" \
174
% (field, data)
175
176
for field in self.optional_fields.keys():
177
if not self.data.has_key(field):
178
self.data[field] = self.optional_fields[field]
70
179
71
180
def __str__(self):
72
181
return self.name
73
74
@property
75
def description(self):
76
description = self.desc
77
if self.command:
78
result = self.command()
79
description = convert_string(self.desc, {'result': result})
80
81
return description
82
83
@property
84
def categories(self):
85
return self.cats
86
87
def create_answer(self, status, data='', auto=False):
182
183
def __getattr__(self, attr):
184
if attr in self.data:
185
return self.data[attr]
186
187
raise AttributeError, attr
188
189
def set_answer(self, status, data='', auto=False):
88
190
self.answer = Answer(self, status, data, auto)
89
return self.answer
90
91
92
class QuestionPlugin(Plugin):
93
94
run_priority = -500
95
96
questions = []
97
98
def gather(self):
99
report = self._manager.report
100
if not report.finalised:
101
for question in self.questions:
102
tests = getattr(report, 'tests', None)
103
if tests is None:
104
tests = createElement(report, 'tests', report.root)
105
report.tests = tests
106
test = createElement(report, 'test', tests)
107
createElement(report, 'suite', test, 'tool')
108
createElement(report, 'name', test, question.name)
109
createElement(report, 'description', test, question.description)
110
createElement(report, 'command', test)
111
createElement(report, 'architectures', test)
112
createTypedElement(report, 'categories', test, None,
113
question.categories, True, 'category')
114
createElement(report, 'optional', test, question.optional)
115
116
if question.answer:
117
result = createElement(report, 'result', test)
118
createElement(report, 'result_status', result,
119
question.answer.status)
120
createElement(report, 'result_data', result,
121
question.answer.data)
122
123
def run(self):
124
for question in self.questions:
125
self._manager.reactor.fire(("prompt", "add-question"), question)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1