56
56
create = ["CREATE TABLE test (foo TEXT UNIQUE, bar INT)"]
57
57
drop = ["DROP TABLE test"]
58
58
delete = ["DELETE FROM test"]
59
uri = "sqlite:///%s" % self.makeFile()
59
60
schema = ZSchema(create, drop, delete, patch_package)
60
uri = "sqlite:///%s" % self.makeFile()
61
self.resource = ZStormResourceManager({"test": (uri, schema)})
61
self.databases = [{"name": "test", "uri": uri, "schema": schema}]
62
self.resource = ZStormResourceManager(self.databases)
62
63
self.store = Store(create_database(uri))
64
65
def tearDown(self):
136
136
self.resource.clean(zstorm)
137
137
self.assertEqual([], list(self.store.execute("SELECT * FROM test")))
139
def test_clean_with_force_delete(self):
141
If L{ZStormResourceManager.force_delete} is C{True}, L{Schema.delete}
142
is always invoked upon test cleanup.
144
zstorm = self.resource.make([])
145
self.store.execute("INSERT INTO test (foo, bar) VALUES ('data', 123)")
147
self.resource.force_delete = True
148
self.resource.clean(zstorm)
149
self.assertEqual([], list(self.store.execute("SELECT * FROM test")))
151
def test_wb_clean_clears_alive_cache_before_abort(self):
153
L{ZStormResourceManager.clean} clears the alive cache before
154
aborting the transaction.
157
__storm_table__ = "test"
158
bar = Int(primary=True)
160
def __init__(self, bar):
163
zstorm = self.resource.make([])
164
store = zstorm.get("test")
167
real_invalidate = store.invalidate
169
def invalidate_proxy():
170
self.assertEqual(0, len(store._alive.values()))
172
store.invalidate = invalidate_proxy
174
self.resource.clean(zstorm)
176
def test_schema_uri(self):
178
It's possible to specify an alternate URI for applying the schema
179
and cleaning up tables after a test.
181
schema_uri = "sqlite:///%s" % self.makeFile()
182
self.databases[0]["schema-uri"] = schema_uri
183
zstorm = self.resource.make([])
184
store = zstorm.get("test")
185
schema_store = Store(create_database(schema_uri))
187
# The schema was applied using the alternate schema URI
188
statement = "SELECT name FROM sqlite_master WHERE name='patch'"
189
self.assertEqual([], list(store.execute(statement)))
190
self.assertEqual([("patch",)], list(schema_store.execute(statement)))
192
# The cleanup is performed with the alternate schema URI
194
schema_store.execute("INSERT INTO test (foo) VALUES ('data')")
195
schema_store.commit()
196
self.resource.clean(zstorm)
197
self.assertEqual([], list(schema_store.execute("SELECT * FROM test")))
199
def test_deprecated_database_format(self):
201
The old deprecated format of the 'database' constructor parameter is
205
uri = "sqlite:///%s" % self.makeFile()
206
schema = ZSchema([], [], [], patch_package)
207
resource = ZStormResourceManager({"test": (uri, schema)})
208
zstorm = resource.make([])
209
store = zstorm.get("test")
210
self.assertIsNot(None, store)