217 store.finish() |
217 store.finish() |
218 with self.admin_access.repo_cnx() as cnx: |
218 with self.admin_access.repo_cnx() as cnx: |
219 final_descr = self.get_db_descr(cnx) |
219 final_descr = self.get_db_descr(cnx) |
220 self.assertEqual(init_descr, final_descr) |
220 self.assertEqual(init_descr, final_descr) |
221 |
221 |
222 def test_on_commit_callback(self): |
|
223 counter = itertools.count() |
|
224 with self.admin_access.repo_cnx() as cnx: |
|
225 store = MassiveObjectStore(cnx, on_commit_callback=lambda: next(counter)) |
|
226 store.prepare_insert_entity('Location', name=u'toto') |
|
227 store.flush() |
|
228 store.commit() |
|
229 self.assertEqual(next(counter), 1) |
|
230 |
|
231 def test_on_rollback_callback(self): |
|
232 counter = itertools.count() |
|
233 with self.admin_access.repo_cnx() as cnx: |
|
234 store = MassiveObjectStore(cnx, on_rollback_callback=lambda *_: next(counter)) |
|
235 # oversized attribute |
|
236 store.prepare_insert_entity('Location', feature_class='toto') |
|
237 store.flush() |
|
238 self.assertEqual(next(counter), 1) |
|
239 |
|
240 def test_simple_insert(self): |
222 def test_simple_insert(self): |
241 with self.admin_access.repo_cnx() as cnx: |
223 with self.admin_access.repo_cnx() as cnx: |
242 store = MassiveObjectStore(cnx) |
224 store = MassiveObjectStore(cnx) |
243 self.push_geonames_data(self.datapath('geonames.csv'), store) |
225 self.push_geonames_data(self.datapath('geonames.csv'), store) |
244 store.flush() |
226 store.flush() |