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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
|
import os
from sqlalchemy import *
from sqlalchemy.ext.proxy import ProxyEngine
from testbase import PersistTest
import testbase
#
# Define an engine, table and mapper at the module level, to show that the
# table and mapper can be used with different real engines in multiple threads
#
class ProxyTestBase(PersistTest):
def setUpAll(self):
global users, User, module_engine, module_metadata
module_engine = ProxyEngine(echo=testbase.echo)
module_metadata = MetaData()
users = Table('users', module_metadata,
Column('user_id', Integer, primary_key=True),
Column('user_name', String(16)),
Column('password', String(20))
)
class User(object):
pass
User.mapper = mapper(User, users)
def tearDownAll(self):
clear_mappers()
class ConstructTest(ProxyTestBase):
"""tests that we can build SQL constructs without engine-specific parameters, particulary
oid_column, being needed, as the proxy engine is usually not connected yet."""
def test_join(self):
engine = ProxyEngine()
t = Table('table1', engine,
Column('col1', Integer, primary_key=True))
t2 = Table('table2', engine,
Column('col2', Integer, ForeignKey('table1.col1')))
j = join(t, t2)
class ProxyEngineTest1(ProxyTestBase):
def test_engine_connect(self):
# connect to a real engine
module_engine.connect(testbase.db_uri)
module_metadata.create_all(module_engine)
session = create_session(bind_to=module_engine)
try:
user = User()
user.user_name='fred'
user.password='*'
session.save(user)
session.flush()
query = session.query(User)
# select
sqluser = query.select_by(user_name='fred')[0]
assert sqluser.user_name == 'fred'
# modify
sqluser.user_name = 'fred jones'
# flush - saves everything that changed
session.flush()
allusers = [ user.user_name for user in query.select() ]
assert allusers == ['fred jones']
finally:
module_metadata.drop_all(module_engine)
class ThreadProxyTest(ProxyTestBase):
def tearDownAll(self):
os.remove('threadtesta.db')
os.remove('threadtestb.db')
def test_multi_thread(self):
from threading import Thread
from Queue import Queue
# start 2 threads with different connection params
# and perform simultaneous operations, showing that the
# 2 threads don't share a connection
qa = Queue()
qb = Queue()
def run(db_uri, uname, queue):
def test():
try:
module_engine.connect(db_uri)
module_metadata.create_all(module_engine)
try:
session = create_session(bind_to=module_engine)
query = session.query(User)
all = list(query.select())
assert all == []
u = User()
u.user_name = uname
u.password = 'whatever'
session.save(u)
session.flush()
names = [u.user_name for u in query.select()]
assert names == [uname]
finally:
module_metadata.drop_all(module_engine)
except Exception, e:
import traceback
traceback.print_exc()
queue.put(e)
else:
queue.put(False)
return test
# NOTE: I'm not sure how to give the test runner the option to
# override these uris, or how to safely clear them after test runs
a = Thread(target=run('sqlite:///threadtesta.db', 'jim', qa))
b = Thread(target=run('sqlite:///threadtestb.db', 'joe', qb))
a.start()
b.start()
# block and wait for the threads to push their results
res = qa.get(True)
if res != False:
raise res
res = qb.get(True)
if res != False:
raise res
class ProxyEngineTest2(ProxyTestBase):
def test_table_singleton_a(self):
"""set up for table singleton check
"""
#
# For this 'test', create a proxy engine instance, connect it
# to a real engine, and make it do some work
#
engine = ProxyEngine()
cats = Table('cats', engine,
Column('cat_id', Integer, primary_key=True),
Column('cat_name', String))
engine.connect(testbase.db_uri)
cats.create(engine)
cats.drop(engine)
ProxyEngineTest2.cats_table_a = cats
assert isinstance(cats, Table)
def test_table_singleton_b(self):
"""check that a table on a 2nd proxy engine instance gets 2nd table
instance
"""
#
# Now create a new proxy engine instance and attach the same
# table as the first test. This should result in 2 table instances,
# since different proxy engine instances can't attach to the
# same table instance
#
engine = ProxyEngine()
cats = Table('cats', engine,
Column('cat_id', Integer, primary_key=True),
Column('cat_name', String))
assert id(cats) != id(ProxyEngineTest2.cats_table_a)
# the real test -- if we're still using the old engine reference,
# this will fail because the old reference's local storage will
# not have the default attributes
engine.connect(testbase.db_uri)
cats.create(engine)
cats.drop(engine)
if __name__ == "__main__":
testbase.main()
|
