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
|
from git_taskrepo.testinfo import SemiStrictParser, ParserError
import string
import os
import commands
# Keys from testinfo to populate with
keys = ("runfor",
"types",
"bugs",
)
class TaskRepoException(Exception):
pass
class TRX(TaskRepoException):
pass
class TRX_Parse(TRX):
pass
class TRX_TestInfo(TRX):
pass
def only_ascii(s):
return filter(lambda x: x in string.printable, s)
def _delete_taskrepo(taskrepo, taskname):
with taskrepo:
cur = taskrepo.cursor()
cur.execute("SELECT id FROM tasks WHERE name=?", (taskname,))
result = cur.fetchone()
if result:
taskid = result[0]
for key in keys:
# Clear old values
cur.execute("DELETE FROM %s WHERE task_id=?" % key, (taskid,))
cur.execute("DELETE FROM tasks WHERE id=?", (taskid,))
def _update_taskrepo(taskrepo, taskname, testinfo):
with taskrepo:
cur = taskrepo.cursor()
# Do we already have an entry for this task?
cur.execute("SELECT id FROM tasks WHERE name=?", (taskname,))
result = cur.fetchone()
if result:
taskid = result[0]
else:
cur.execute("INSERT INTO tasks(name) VALUES (?)", (taskname,))
taskid = cur.lastrowid;
# Update description and owner
cur.execute("UPDATE tasks SET description = ?, owner = ? WHERE id = ? ",
(only_ascii(testinfo.test_description),
only_ascii(testinfo.owner),
taskid))
for key in keys:
# Clear old values
cur.execute("DELETE FROM %s WHERE task_id=?" % key, (taskid,))
# Populate with new values
for value in getattr(testinfo, key):
cur.execute("INSERT INTO %s VALUES (?,?)" % key, (taskid, value))
def update_taskrepo(repo, taskrepo, taskpath):
taskname = taskpath.split("%s/" % repo.working_tree_dir)[1]
if os.path.isfile(os.path.join(taskpath, "Makefile")):
os.system("make -ki -C %s testinfo.desc >/dev/null 2>&1" % taskpath)
if os.path.isfile(os.path.join(taskpath, "testinfo.desc")):
try:
testinfo = parse_testinfo(os.path.join(taskpath, "testinfo.desc"))
except ParserError:
raise
_update_taskrepo(taskrepo, taskname, testinfo)
else:
_delete_taskrepo(taskrepo, taskname)
raise TRX_TestInfo('No testinfo.desc')
def parse_testinfo(filename):
p = SemiStrictParser(True)
p.parse(open(filename).readlines())
return p.info
|
