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
|
#!/usr/bin/python
#
# Copyright 2010-2011 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
__metaclass__ = type
from bz2 import BZ2File
from optparse import (
OptionGroup,
OptionValueError,
)
from os import (
path,
walk,
)
from Queue import Empty
from urlparse import urlparse
from lpresults.scripts.application import ApplicationError
from lpresults.scripts.functor import Functor
from lpresults.scripts.mixins.launchpad import LaunchpadMixin
from lpresults.scripts.multiapplication import MultiApplication
from lpresults.scripts.option import DateOption
from lpresults.xunit.protocol import get_protocol
class Results2Launchpad(LaunchpadMixin, MultiApplication):
# Application defaults
usage = "Usage: %prog [OPTIONS] <PROJECT NAME> <SUBMISSION FILE>..."
def addOptions(self, parser):
"""See Application."""
super(Results2Launchpad, self).addOptions(parser)
group = OptionGroup(parser, "Results options")
group.option_class = DateOption
group.add_option("-d", "--date-created",
metavar="DATE",
type="date",
help="Date on which this test run was created.")
group.add_option("-r", "--recursive",
action="store_true",
help="Read all files under each directory, recursively")
group.add_option("-s", "--series",
help="The series to report results against")
parser.add_option_group(group)
def parseOptions(self, options, args):
"""See Application."""
super(Results2Launchpad, self).parseOptions(options, args)
# Check project name
if len(args) < 1:
raise OptionValueError("Must provide a project name")
# Check filename arguments
if len(args) < 2:
raise OptionValueError(
"Must provide a project name and a submission file")
self.project = args[0]
self.filenames = args[1:]
self.date_created = options.date_created
self.recursive = options.recursive
self.series = options.series
def processInput(self):
"""See MultiApplication."""
for filename in self.filenames:
if path.isfile(filename):
self.input_queue.put(filename)
elif self.recursive:
for root, directories, filenames in walk(filename):
for filename in filenames:
self.input_queue.put(path.join(root, filename))
def processMessages(self, index):
"""See MultiApplication."""
launchpad = self.getLaunchpad(index=index)
test_run_factory = Functor(
create_test_run, launchpad=launchpad,
date_created=self.date_created)
while not self.input_queue.empty():
try:
filename = self.input_queue.get(block=False)
self.handleFilename(test_run_factory, filename)
except Empty:
pass
def handleFilename(self, test_run_factory, filename):
if filename.endswith(".bz2"):
file = BZ2File(filename)
else:
file = open(filename)
try:
self.handleFile(test_run_factory, file)
finally:
file.close()
def handleFile(self, test_run_factory, file):
protocol = get_protocol(file)
if not protocol:
name = getattr(file, "name", "<stream>")
raise ApplicationError(
"Unsupported xunit protocol in file: %s" % name)
protocol.run(
test_run_factory, project=self.project, series=self.series)
def create_test_run(launchpad, project, series=None, date_created=None):
try:
# This will get a project or a distribution
target = launchpad.projects[project]
except KeyError:
raise ApplicationError(
"Unknown project name: %s" % project)
if series:
target = target.getSeries(name=series)
if not target:
raise ApplicationError(
"Unknown series name: %s" % series)
return target.createTestRun(date_created=date_created)
if __name__ == "__main__":
application = Results2Launchpad()
application.run()
|