← Back to branch summary

~ubuntu-branches/debian/sid/sqlalchemy/sid

« back to all changes in this revision

Viewing changes to test/engine/test_execute.py

  • Committer: Package Import Robot
  • Author(s): Piotr Ożarowski
  • Date: 2012-03-15 21:05:49 UTC
  • mfrom: (1.4.19)
  • Revision ID: package-import@ubuntu.com-20120315210549-fiuynu6jue9keqlh
Tags: 0.7.6-1
* New upstream release
* debhelper's compatibility bumped to 7
* Standards-Version bumped to 3.9.3 (no changes needed)

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/engine/test_execute.py
41
41
    def teardown_class(cls):
42
42
        metadata.drop_all()
43
43
 
 
44
    @testing.fails_on("postgresql+pg8000", 
 
45
            "pg8000 still doesn't allow single % without params")
 
46
    def test_no_params_option(self):
 
47
        stmt = "SELECT '%'"
 
48
        if testing.against('oracle'):
 
49
            stmt += " FROM DUAL"
 
50
        conn = testing.db.connect()
 
51
        result = conn.\
 
52
                execution_options(no_parameters=True).\
 
53
                scalar(stmt)
 
54
        eq_(result, '%')
 
55
 
44
56
    @testing.fails_on_everything_except('firebird', 'maxdb',
45
57
                                        'sqlite', '+pyodbc',
46
58
                                        '+mxodbc', '+zxjdbc', 'mysql+oursql',
305
317
        assert eng.dialect.returns_unicode_strings in (True, False)
306
318
        eng.dispose()
307
319
 
 
320
class ConvenienceExecuteTest(fixtures.TablesTest):
 
321
    @classmethod
 
322
    def define_tables(cls, metadata):
 
323
        cls.table = Table('exec_test', metadata,
 
324
            Column('a', Integer),
 
325
            Column('b', Integer),
 
326
            test_needs_acid=True
 
327
        )
 
328
 
 
329
    def _trans_fn(self, is_transaction=False):
 
330
        def go(conn, x, value=None):
 
331
            if is_transaction:
 
332
                conn = conn.connection
 
333
            conn.execute(self.table.insert().values(a=x, b=value))
 
334
        return go
 
335
 
 
336
    def _trans_rollback_fn(self, is_transaction=False):
 
337
        def go(conn, x, value=None):
 
338
            if is_transaction:
 
339
                conn = conn.connection
 
340
            conn.execute(self.table.insert().values(a=x, b=value))
 
341
            raise Exception("breakage")
 
342
        return go
 
343
 
 
344
    def _assert_no_data(self):
 
345
        eq_(
 
346
            testing.db.scalar(self.table.count()), 0
 
347
        )
 
348
 
 
349
    def _assert_fn(self, x, value=None):
 
350
        eq_(
 
351
            testing.db.execute(self.table.select()).fetchall(),
 
352
            [(x, value)]
 
353
        )
 
354
 
 
355
    def test_transaction_engine_ctx_commit(self):
 
356
        fn = self._trans_fn()
 
357
        ctx = testing.db.begin()
 
358
        testing.run_as_contextmanager(ctx, fn, 5, value=8)
 
359
        self._assert_fn(5, value=8)
 
360
 
 
361
    def test_transaction_engine_ctx_rollback(self):
 
362
        fn = self._trans_rollback_fn()
 
363
        ctx = testing.db.begin()
 
364
        assert_raises_message(
 
365
            Exception, 
 
366
            "breakage",
 
367
            testing.run_as_contextmanager, ctx, fn, 5, value=8
 
368
        )
 
369
        self._assert_no_data()
 
370
 
 
371
    def test_transaction_tlocal_engine_ctx_commit(self):
 
372
        fn = self._trans_fn()
 
373
        engine = engines.testing_engine(options=dict(
 
374
                                strategy='threadlocal', 
 
375
                                pool=testing.db.pool))
 
376
        ctx = engine.begin()
 
377
        testing.run_as_contextmanager(ctx, fn, 5, value=8)
 
378
        self._assert_fn(5, value=8)
 
379
 
 
380
    def test_transaction_tlocal_engine_ctx_rollback(self):
 
381
        fn = self._trans_rollback_fn()
 
382
        engine = engines.testing_engine(options=dict(
 
383
                                strategy='threadlocal', 
 
384
                                pool=testing.db.pool))
 
385
        ctx = engine.begin()
 
386
        assert_raises_message(
 
387
            Exception, 
 
388
            "breakage",
 
389
            testing.run_as_contextmanager, ctx, fn, 5, value=8
 
390
        )
 
391
        self._assert_no_data()
 
392
 
 
393
    def test_transaction_connection_ctx_commit(self):
 
394
        fn = self._trans_fn(True)
 
395
        conn = testing.db.connect()
 
396
        ctx = conn.begin()
 
397
        testing.run_as_contextmanager(ctx, fn, 5, value=8)
 
398
        self._assert_fn(5, value=8)
 
399
 
 
400
    def test_transaction_connection_ctx_rollback(self):
 
401
        fn = self._trans_rollback_fn(True)
 
402
        conn = testing.db.connect()
 
403
        ctx = conn.begin()
 
404
        assert_raises_message(
 
405
            Exception, 
 
406
            "breakage",
 
407
            testing.run_as_contextmanager, ctx, fn, 5, value=8
 
408
        )
 
409
        self._assert_no_data()
 
410
 
 
411
    def test_connection_as_ctx(self):
 
412
        fn = self._trans_fn()
 
413
        ctx = testing.db.connect()
 
414
        testing.run_as_contextmanager(ctx, fn, 5, value=8)
 
415
        # autocommit is on
 
416
        self._assert_fn(5, value=8)
 
417
 
 
418
    @testing.fails_on('mysql+oursql', "oursql bug ?  getting wrong rowcount")
 
419
    def test_connect_as_ctx_noautocommit(self):
 
420
        fn = self._trans_fn()
 
421
        self._assert_no_data()
 
422
        ctx = testing.db.connect().execution_options(autocommit=False)
 
423
        testing.run_as_contextmanager(ctx, fn, 5, value=8)
 
424
        # autocommit is off
 
425
        self._assert_no_data()
 
426
 
 
427
    def test_transaction_engine_fn_commit(self):
 
428
        fn = self._trans_fn()
 
429
        testing.db.transaction(fn, 5, value=8)
 
430
        self._assert_fn(5, value=8)
 
431
 
 
432
    def test_transaction_engine_fn_rollback(self):
 
433
        fn = self._trans_rollback_fn()
 
434
        assert_raises_message(
 
435
            Exception, 
 
436
            "breakage",
 
437
            testing.db.transaction, fn, 5, value=8
 
438
        )
 
439
        self._assert_no_data()
 
440
 
 
441
    def test_transaction_connection_fn_commit(self):
 
442
        fn = self._trans_fn()
 
443
        conn = testing.db.connect()
 
444
        conn.transaction(fn, 5, value=8)
 
445
        self._assert_fn(5, value=8)
 
446
 
 
447
    def test_transaction_connection_fn_rollback(self):
 
448
        fn = self._trans_rollback_fn()
 
449
        conn = testing.db.connect()
 
450
        assert_raises(
 
451
            Exception, 
 
452
            conn.transaction, fn, 5, value=8
 
453
        )
 
454
        self._assert_no_data()
 
455
 
308
456
class CompiledCacheTest(fixtures.TestBase):
309
457
    @classmethod
310
458
    def setup_class(cls):