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
|
#!/usr/bin/env python2.3
# Copyright 2004 Canonical Ltd. All rights reserved.
"""Tests that get run automatically on a merge."""
import sys
import os, os.path
import popen2
import tabnanny
import checkarchtag
import checktitles
from StringIO import StringIO
from threading import Thread
import psycopg
class NonBlockingReader(Thread):
result = None
def __init__(self,file):
Thread.__init__(self)
self.file = file
def run(self):
self.result = self.file.read()
def read(self):
if self.result is None:
raise RuntimeError("read() called before run()")
return self.result
def readlines(self):
if self.result is None:
raise RuntimeError("readlines() called before run()")
return self.result.splitlines()
def main():
"""Call test.py with whatever arguments this script was run with.
If the tests ran ok (last line of stderr is 'OK<return>') then suppress
output and exit(0).
Otherwise, print output and exit(1).
"""
here = os.path.dirname(os.path.realpath(__file__))
if not checkarchtag.is_tree_good():
return 1
checktitles.summarise_directory("lib/canonical/launchpad/templates")
# Tabnanny
org_stdout = sys.stdout
sys.stdout = StringIO()
tabnanny.check(os.path.join(here, 'lib', 'canonical'))
tabnanny_results = sys.stdout.getvalue()
sys.stdout = org_stdout
if len(tabnanny_results) > 0:
print '---- tabnanny bitching ----'
print tabnanny_results
print '---- end tabnanny bitching ----'
return 1
# Ensure ++resource++ URL's are all absolute - this ensures they
# are cache friendly
results = os.popen(
"find lib/canonical -type f | xargs grep '[^/]++resource++'"
).readlines()
if results:
print '---- non-absolute ++resource++ URLs found ----'
print ''.join(results)
print '---- end non-absolute ++resource++ URLs found ----'
return 1
# Drop the template database if it exists - the Makefile does this
# too, but we can explicity check for errors here
con = psycopg.connect('dbname=template1')
cur = con.cursor()
cur.execute('end transaction; drop database launchpad_ftest_template')
cur.close()
con.close()
# Build the template database. Tests duplicate this.
here = os.path.dirname(os.path.realpath(__file__))
schema_dir = os.path.join(here, 'database', 'schema')
if os.system('cd %s; make test > /dev/null 2>&1' % schema_dir) != 0:
print 'Failed to create database'
return 1
# Sanity check the database. No point running tests if the
# bedrock is crumbling.
con = psycopg.connect('dbname=launchpad_ftest_template')
cur = con.cursor()
cur.execute('show search_path')
search_path = cur.fetchone()[0]
if search_path != '$user,public,ts2':
print 'Search path incorrect.'
print 'Add the following line to /etc/postgresql/postgresql.conf:'
print " search_path = '$user,public,ts2'"
return 1
cur.execute("""
select count(*) from person where displayname='Mark Shuttleworth'
""")
cnt = cur.fetchone()[0]
if cnt < 1:
print 'Sample data not loaded.'
return 1
cur.execute("""
select pg_encoding_to_char(encoding) as encoding from pg_database
where datname='launchpad_ftest_template'
""")
enc = cur.fetchone()[0]
if enc != 'UNICODE':
print 'Database encoding incorrectly set'
return 1
# Explicity close our connections - things will fail if we leave open
# connections.
cur.close()
del cur
con.close()
del con
print 'Running tests.'
proc = popen2.Popen3('cd %s; python test.py %s < /dev/null' %
(here, ' '.join(sys.argv[1:])), True)
stdin, out, err = proc.tochild, proc.fromchild, proc.childerr
# Use non-blocking reader threads to cope with differing expectations
# from the proess of when to consume data from out and error.
errthread = NonBlockingReader(err)
outthread = NonBlockingReader(out)
errthread.start()
outthread.start()
errthread.join()
outthread.join()
exitcode = proc.wait()
test_ok = (os.WIFEXITED(exitcode) and os.WEXITSTATUS(exitcode) == 0)
errlines = errthread.readlines()
dataout = outthread.read()
if test_ok:
print errlines[1]
return 0
else:
print '---- test stdout ----'
print dataout
print '---- end test stdout ----'
print '---- test stderr ----'
print '\n'.join(errlines)
print '---- end test stderr ----'
return 1
if __name__ == '__main__':
sys.exit(main())
|