1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
#
# This file is part of Checkbox.
#
# Copyright 2008 Canonical Ltd.
#
# Checkbox is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3,
# as published by the Free Software Foundation.
#
# Checkbox is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.
#
import copy
from checkbox.contrib.persist import Persist, MemoryBackend
from checkbox.job import UNINITIATED
from checkbox.lib.resolver import Resolver
from checkbox.plugin import Plugin
from checkbox.properties import String
from checkbox.user_interface import CONTINUE_ANSWER, RERUN_ANSWER
from gettext import gettext as _
class SuitesPrompt(Plugin):
deselect_warning = String(default="")
@property
def persist(self):
if self._persist is None:
self._persist = Persist(backend=MemoryBackend())
return self._persist.root_at("suites_prompt")
def register(self, manager):
super(SuitesPrompt, self).register(manager)
self._depends = {}
self._jobs = {}
self._statuses = {}
self._persist = None
self._recover = False
for (rt, rh) in [
("begin-persist", self.begin_persist),
("begin-recover", self.begin_recover),
("report-suite", self.report_suite),
("store-access", self.store_access)]:
self._manager.reactor.call_on(rt, rh)
for (rt, rh) in [
("prompt-gather", self.prompt_gather),
("report-suite", self.report_job),
("report-test", self.report_job)]:
self._manager.reactor.call_on(rt, rh, 100)
def begin_persist(self, persist):
self._persist = persist
def begin_recover(self, recover):
if recover in [CONTINUE_ANSWER, RERUN_ANSWER]:
self._recover = True
if not self._recover:
self.persist.remove("default")
def report_suite(self, suite):
suite.setdefault("type", "suite")
def store_access(self, store):
self.store = store
def report_job(self, job):
if job.get("type") == "suite":
attribute = "description"
else:
attribute = "name"
if attribute in job:
self._jobs[job["name"]] = job[attribute]
if "suite" in job:
self._depends[job["name"]] = [job["suite"]]
if job.get("type") == "test":
self._statuses[job["name"]] = job["status"]
def prompt_gather(self, interface):
# Resolve dependencies
interface.show_progress_start(_("Gathering information from your system..."))
resolver = Resolver()
for key in self._jobs.keys():
depends = self._depends.get(key, [])
resolver.add(key, *depends)
# Build options
options = {}
self._manager.reactor.fire("expose-msgstore")
offset = self.store.get_pending_offset()
self.store.set_pending_offset(0)
messages = self.store.get_pending_messages()
self.store.add_pending_offset(offset)
tests = dict([(m["name"], m) for m in messages
if m.get("type") in ("test", "metric")])
def walk_dependencies(job, all_dependencies):
for dependency in resolver.get_dependencies(job)[:-1]:
walk_dependencies(dependency, all_dependencies)
all_dependencies.append(job)
for job in resolver.get_dependents():
suboptions = options
dependencies = []
walk_dependencies(job, dependencies)
for dependency in dependencies:
if dependency in tests:
value = tests[dependency]["status"]
else:
value = self._statuses.get(dependency, {})
suboptions = suboptions.setdefault(self._jobs[dependency],
value)
# Build defaults
defaults = self.persist.get("default")
if defaults is None:
defaults = copy.deepcopy(options)
# Get results
interface.show_progress_stop()
defaults = interface.show_tree(
_("Choose tests to run on your system:"),
options, defaults, self.deselect_warning)
self.persist.set("default", defaults)
# Get tests to ignore
def get_ignore_jobs(options, results):
jobs = []
if isinstance(options, dict):
for k, v in options.items():
if v == UNINITIATED and k not in results:
jobs.append(k)
else:
jobs.extend(get_ignore_jobs(options[k],
results.get(k, {})))
return jobs
ignore_jobs = get_ignore_jobs(options, defaults)
self._manager.reactor.fire("ignore-jobs", ignore_jobs)
factory = SuitesPrompt
|