5
from testlib import config, sa, testing
6
from testlib.testing import resolve_artifact_names, adict
7
from testlib.compat import _function_named
5
import sqlalchemy as sa
6
from sqlalchemy.test import config, testing
7
from sqlalchemy.test.testing import resolve_artifact_names, adict
8
from sqlalchemy.util import function_named
10
11
_repr_stack = set()
58
59
self_key = sa.orm.attributes.instance_state(self).key
59
60
except sa.orm.exc.NO_STATE:
62
other_key = sa.orm.attributes.instance_state(other).key
63
except sa.orm.exc.NO_STATE:
66
if other_key is None and self_key is not None:
66
elif self_key is not None:
71
73
for attr in a.__dict__.keys():
72
74
if attr.startswith('_'):
94
96
class ORMTest(testing.TestBase, testing.AssertsExecutionResults):
95
97
__requires__ = ('subqueries',)
97
def tearDownAll(self):
100
def teardown_class(cls):
98
101
sa.orm.session.Session.close_all()
99
102
sa.orm.clear_mappers()
100
103
# TODO: ensure mapper registry is empty
101
104
# TODO: ensure instrumentation registry is empty
103
106
class MappedTest(ORMTest):
104
# 'once', 'foreach', None
107
# 'once', 'each', None
105
108
run_define_tables = 'once'
107
# 'once', 'foreach', None
110
# 'once', 'each', None
108
111
run_setup_classes = 'once'
110
# 'once', 'foreach', None
113
# 'once', 'each', None
111
114
run_setup_mappers = 'each'
113
# 'once', 'foreach', None
116
# 'once', 'each', None
114
117
run_inserts = 'each'
117
120
run_deletes = 'each'
124
127
other_artifacts = None
127
if self.run_setup_classes == 'each':
128
assert self.run_setup_mappers != 'once'
130
assert self.run_deletes in (None, 'each')
131
if self.run_inserts == 'once':
132
assert self.run_deletes is None
134
assert not hasattr(self, 'keep_mappers')
135
assert not hasattr(self, 'keep_data')
130
def setup_class(cls):
131
if cls.run_setup_classes == 'each':
132
assert cls.run_setup_mappers != 'once'
134
assert cls.run_deletes in (None, 'each')
135
if cls.run_inserts == 'once':
136
assert cls.run_deletes is None
138
assert not hasattr(cls, 'keep_mappers')
139
assert not hasattr(cls, 'keep_data')
138
141
if cls.tables is None:
139
142
cls.tables = adict()
140
143
if cls.classes is None:
142
145
if cls.other_artifacts is None:
143
146
cls.other_artifacts = adict()
145
if self.metadata is None:
146
setattr(type(self), 'metadata', sa.MetaData())
148
if self.metadata.bind is None:
149
self.metadata.bind = getattr(self, 'engine', config.db)
151
if self.run_define_tables:
152
self.define_tables(self.metadata)
153
self.metadata.create_all()
154
self.tables.update(self.metadata.tables)
156
if self.run_setup_classes:
157
baseline = subclasses(BasicEntity)
159
self._register_new_class_artifacts(baseline)
161
if self.run_setup_mappers:
162
baseline = subclasses(BasicEntity)
164
self._register_new_class_artifacts(baseline)
167
self._load_fixtures()
171
if self._sa_first_test:
148
if cls.metadata is None:
149
setattr(cls, 'metadata', sa.MetaData())
151
if cls.metadata.bind is None:
152
cls.metadata.bind = getattr(cls, 'engine', config.db)
154
if cls.run_define_tables == 'once':
155
cls.define_tables(cls.metadata)
156
cls.metadata.create_all()
157
cls.tables.update(cls.metadata.tables)
159
if cls.run_setup_classes == 'once':
160
baseline = subclasses(BasicEntity)
162
cls._register_new_class_artifacts(baseline)
164
if cls.run_setup_mappers == 'once':
165
baseline = subclasses(BasicEntity)
167
cls._register_new_class_artifacts(baseline)
169
if cls.run_inserts == 'once':
174
174
if self.run_define_tables == 'each':
175
175
self.tables.clear()
176
176
self.metadata.drop_all()
194
194
self._load_fixtures()
195
195
self.insert_data()
198
198
sa.orm.session.Session.close_all()
200
if self.run_setup_mappers == 'each':
200
# some tests create mappers in the test bodies
201
# and will define setup_mappers as None -
202
# clear mappers in any case
203
if self.run_setup_mappers != 'once':
201
204
sa.orm.clear_mappers()
203
206
# no need to run deletes if tables are recreated on setup
209
212
print >> sys.stderr, "Error emptying table %s: %r" % (
212
def tearDownAll(self):
213
for cls in self.classes.values():
214
self.unregister_class(cls)
215
ORMTest.tearDownAll(self)
216
self.metadata.drop_all()
217
self.metadata.bind = None
216
def teardown_class(cls):
217
for cl in cls.classes.values():
218
cls.unregister_class(cl)
219
ORMTest.teardown_class()
220
cls.metadata.drop_all()
221
cls.metadata.bind = None
219
def define_tables(self, metadata):
224
def define_tables(cls, metadata):
220
225
raise NotImplementedError()
222
def setup_classes(self):
225
def setup_mappers(self):
228
def setup_classes(cls):
232
def setup_mappers(cls):
231
def insert_data(self):
240
def insert_data(cls):
234
243
def sql_count_(self, count, fn):
256
265
if name[0].isupper:
257
266
delattr(cls, name)
258
267
del cls.classes[name]
260
def _load_fixtures(self):
270
def _load_fixtures(cls):
261
271
headers, rows = {}, {}
262
for table, data in self.fixtures().iteritems():
272
for table, data in cls.fixtures().iteritems():
263
273
if isinstance(table, basestring):
264
table = self.tables[table]
274
table = cls.tables[table]
265
275
headers[table] = data[0]
266
276
rows[table] = data[1:]
267
for table in self.metadata.sorted_tables:
277
for table in cls.metadata.sorted_tables:
268
278
if table not in headers:
270
280
table.bind.execute(
added
removed
test/orm/_base.py
