1
#!/usr/bin/env python2.6
10
import multiprocessing
12
gtk.gdk.threads_init()
13
gobject.threads_init()
16
class Application(object):
19
window.connect("destroy", self.quit)
21
vbox = gtk.VBox(False, 2)
24
button = gtk.Button("Run tests")
25
button.connect("clicked", self.button_callback)
26
vbox.pack_start(button, False, False)
28
scrolled = gtk.ScrolledWindow()
29
scrolled.set_policy(gtk.POLICY_AUTOMATIC, gtk.POLICY_AUTOMATIC)
32
files = self.list_of_files()
34
self.box = gtk.Table(len(files), 2)
36
headings = ("Filename", "Time spent", "Outcome")
37
for ii, heading in enumerate(headings):
39
label.set_markup("<b>%s</b>" % heading)
40
self.box.attach(label, ii, ii + 1, 0, 1)
43
for y, file in enumerate(files):
44
label = gtk.Label(file.split("/")[-1])
45
label.set_alignment(0, 0.5)
46
self.labels[file] = [label, gtk.Label("."), gtk.Label(".")]
47
for x, widget in enumerate(self.labels[file]):
48
self.box.attach(widget, x, x + 1, y + 1, y + 2)
50
scrolled.add_with_viewport(self.box)
52
vbox.pack_start(scrolled)
54
self.pipe, child_pipe = multiprocessing.Pipe()
55
self.process = TimeProcesses(self.box, files, child_pipe)
56
self.thread = threading.Thread(target=self.monitor_progress)
58
window.resize(600, 700)
67
for pattern in ("*.docx", "*.ppt", "*.pptx", "*.xls", "*.xlsx", "*.doc"):
68
files.extend(glob.glob("test/files/" + pattern))
71
def monitor_progress(self):
72
self.do_monitor_progress = True
73
while self.do_monitor_progress:
74
if self.pipe.poll(0.2):
75
recv = self.pipe.recv()
76
if recv[0] == "start":
77
gobject.idle_add(lambda recv=recv: self.modify_label(recv[1], ["<span color=\"red\">...</span>"] * 2))
78
elif recv[0] == "complete":
79
gobject.idle_add(lambda recv=recv: self.modify_label(recv[1], recv[2:4]))
80
elif recv[0] == "quit":
81
self.do_monitor_progress = False
84
def modify_label(self, file, markup_list):
85
labels = self.labels[file]
86
for ii, markup in enumerate(markup_list):
88
labels[ii + 1].set_markup(markup)
91
if not self.process is None:
93
if self.process.is_alive():
94
self.process.terminate()
96
print >> sys.stderr, str(ee)
97
self.do_monitor_progress = False
101
def button_callback(self, widget):
102
widget.set_sensitive(False)
109
class TimeProcesses(multiprocessing.Process):
110
def __init__(self, box, files, pipe):
111
multiprocessing.Process.__init__(self)
115
self.script = os.path.join(
116
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
120
if not os.path.isdir(os.path.join(os.path.dirname(self.script), "test/output")):
121
os.mkdir(os.path.join(os.path.dirname(self.script), "test/output"))
122
for ii, file in enumerate(self.files):
123
self.pipe.send(["start", file])
124
time, outcome = self.test_file(file, None)
125
self.pipe.send(["complete", file, time, outcome])
126
self.pipe.send(["quit"])
128
def test_file(self, file, label):
129
process = subprocess.Popen(
130
["/usr/bin/time","-f", "_%e_%D_%x_", self.script,
131
file, "test/output/%s.png" % file.split("/")[-1], "128"],
132
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
133
cwd=os.path.dirname(self.script))
135
for line in reversed(process.stderr.readlines()):
136
if len(line) > 2 and line[0] == "_" and line[-2] == "_":
137
params = line.split("_")
138
if int(params[2]) == 0 and \
139
os.path.exists(os.path.join(os.path.dirname(self.script), "test/output/%s.png" % file.split("/")[-1])):
141
elif int(params[2]) == 0:
142
outcome = "<span color=\"red\">Thumbnail not found</span>"
144
outcome = "<span color=\"red\">Failed (code:%s)</span>" % params[2]
145
return params[1] + " seconds", outcome
146
return "<span color=\"red\">Python logic error</span>", "<span color=\"red\">Python logic error</span>"
151
if __name__ == "__main__":