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
|
#
# Copyright © 2012 Intel Corporation
#
# Permission is hereby granted, free of charge, to any person obtaining a
# copy of this software and associated documentation files (the "Software"),
# to deal in the Software without restriction, including without limitation
# the rights to use, copy, modify, merge, publish, distribute, sublicense,
# and/or sell copies of the Software, and to permit persons to whom the
# Software is furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice (including the next
# paragraph) shall be included in all copies or substantial portions of the
# Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
# THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.
#
import apsw
import sys
import time
import os.path as path
import xdg.BaseDirectory as xdg
__all__ = ['ResultDatabase']
class ResultDatabase:
def __init__(self, config):
'''Load the results database, creating it if it doesn't exist.
Will exit the program if it fails.'''
try:
self.connection = apsw.Connection(config['options']['database'])
cursor = self.connection.cursor()
cursor.execute('''CREATE TABLE IF NOT EXISTS runs
(name TEXT PRIMARY KEY,
date INTEGER,
driver TEXT,
sysinfo TEXT)''')
cursor.execute('''CREATE TABLE IF NOT EXISTS results
(run_name TEXT,
test_name TEXT,
command TEXT,
result TEXT,
return_code INTEGER,
errors TEXT,
output TEXT)''')
except OSError as e:
print('error: could not open results database', str(e), '\n')
sys.exit(0)
def createRun(self, name, driver, sysinfo):
c = self.connection.cursor()
date = int(time.time())
try:
c.execute('INSERT INTO runs VALUES(?,?,?,?)',
(name, date, driver, sysinfo))
return True
except:
return False
def listRuns(self):
cursor = self.connection.cursor()
rs = cursor.execute('SELECT * FROM runs ORDER BY date DESC')
return [{'name': r[0], 'date': r[1], 'driver': r[2], 'sysinfo': r[3]} for r in rs]
def mostRecentRun(self):
cursor = self.connection.cursor()
rs = cursor.execute('SELECT name FROM runs ORDER BY date DESC LIMIT 1')
return rs[0][0]
def addPlaceholderResults(self, run, tests):
rows = [(run, test) for test in tests]
c = self.connection.cursor()
c.execute('BEGIN IMMEDIATE TRANSACTION')
c.executemany('INSERT INTO results(run_name, test_name) VALUES(?,?)', rows)
c.execute('END TRANSACTION')
def findPlaceholders(self, run):
c = self.connection.cursor()
xs = c.execute('SELECT test_name FROM results WHERE run_name = ? AND result IS NULL', [run])
return [x[0] for x in xs]
def writeResult(self, run, name, command, result):
c = self.connection.cursor()
c.execute('UPDATE results SET command = ?, result = ?, return_code = ?, errors = ?, output = ? WHERE run_name = ? AND test_name = ? AND result IS NULL', (' '.join(command), result['status'], result['exitcode'], result['err'], result['out'], run, name))
def getResults(self, run):
c = self.connection.cursor()
query = 'SELECT test_name, result FROM results WHERE run_name = ?'
xs = c.execute(query, [run])
return dict(xs)
|
