← Back to branch summary

~ubuntu-branches/debian/jessie/sqlalchemy/jessie

« back to all changes in this revision

Viewing changes to test/engine/test_transaction.py

  • Committer: Package Import Robot
  • Author(s): Piotr Ożarowski, Jakub Wilk, Piotr Ożarowski
  • Date: 2013-07-06 20:53:52 UTC
  • mfrom: (1.4.23) (16.1.17 experimental)
  • Revision ID: package-import@ubuntu.com-20130706205352-ryppl1eto3illd79
Tags: 0.8.2-1
[ Jakub Wilk ]
* Use canonical URIs for Vcs-* fields.

[ Piotr Ożarowski ]
* New upstream release
* Upload to unstable
* Build depend on python3-all instead of -dev, extensions are not built for
  Python 3.X 

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/engine/test_transaction.py
1
 
from test.lib.testing import eq_, assert_raises, \
 
1
from sqlalchemy.testing import eq_, assert_raises, \
2
2
    assert_raises_message, ne_
3
3
import sys
4
4
import time
5
5
import threading
6
 
from test.lib.engines import testing_engine
 
6
from sqlalchemy.testing.engines import testing_engine
7
7
from sqlalchemy import create_engine, MetaData, INT, VARCHAR, Sequence, \
8
8
    select, Integer, String, func, text, exc
9
 
from test.lib.schema import Table
10
 
from test.lib.schema import Column
11
 
from test.lib import testing
12
 
from test.lib import fixtures
 
9
from sqlalchemy.testing.schema import Table
 
10
from sqlalchemy.testing.schema import Column
 
11
from sqlalchemy import testing
 
12
from sqlalchemy.testing import fixtures
13
13
 
14
14
 
15
15
users, metadata = None, None
29
29
        testing.db.execute(users.delete()).close()
30
30
 
31
31
    @classmethod
 
32
    @testing.crashes('mysql+cymysql', 'deadlock')
32
33
    def teardown_class(cls):
33
34
        users.drop(testing.db)
34
35
 
1008
1009
    def teardown_class(cls):
1009
1010
        counters.drop(testing.db)
1010
1011
 
1011
 
    def increment(
1012
 
        self,
1013
 
        count,
1014
 
        errors,
1015
 
        update_style=True,
1016
 
        delay=0.005,
1017
 
        ):
 
1012
    def increment(self, count, errors, update_style=True, delay=0.005):
1018
1013
        con = testing.db.connect()
1019
1014
        sel = counters.select(for_update=update_style,
1020
1015
                              whereclause=counters.c.counter_id == 1)
1065
1060
            threads.append(thrd)
1066
1061
        for thrd in threads:
1067
1062
            thrd.join()
1068
 
        for e in errors:
1069
 
            sys.stdout.write('Failure: %s\n' % e)
1070
 
        self.assert_(len(errors) == 0)
 
1063
        assert not errors
1071
1064
        sel = counters.select(whereclause=counters.c.counter_id == 1)
1072
1065
        final = db.execute(sel).first()
1073
 
        self.assert_(final['counter_value'] == iterations
1074
 
                     * thread_count)
1075
 
 
1076
 
    def overlap(
1077
 
        self,
1078
 
        ids,
1079
 
        errors,
1080
 
        update_style,
1081
 
        ):
 
1066
        eq_(final['counter_value'], iterations * thread_count)
 
1067
 
 
1068
    def overlap(self, ids, errors, update_style):
 
1069
 
1082
1070
        sel = counters.select(for_update=update_style,
1083
1071
                              whereclause=counters.c.counter_id.in_(ids))
1084
1072
        con = testing.db.connect()
1085
1073
        trans = con.begin()
1086
1074
        try:
1087
1075
            rows = con.execute(sel).fetchall()
1088
 
            time.sleep(0.25)
 
1076
            time.sleep(0.50)
1089
1077
            trans.commit()
1090
1078
        except Exception, e:
1091
1079
            trans.rollback()
1092
1080
            errors.append(e)
1093
1081
        con.close()
1094
1082
 
1095
 
    def _threaded_overlap(
1096
 
        self,
1097
 
        thread_count,
1098
 
        groups,
1099
 
        update_style=True,
1100
 
        pool=5,
1101
 
        ):
 
1083
    def _threaded_overlap(self, thread_count, groups, update_style=True, pool=5):
1102
1084
        db = testing.db
1103
1085
        for cid in range(pool - 1):
1104
1086
            db.execute(counters.insert(), counter_id=cid + 1,
1108
1090
            thrd = threading.Thread(target=self.overlap,
1109
1091
                                    args=(groups.pop(0), errors,
1110
1092
                                    update_style))
 
1093
            time.sleep(0.20)  # give the previous thread a chance to start
 
1094
                              # to ensure it gets a lock
1111
1095
            thrd.start()
1112
1096
            threads.append(thrd)
1113
1097
        for thrd in threads:
1123
1107
        """Simple SELECT FOR UPDATE conflict test"""
1124
1108
 
1125
1109
        errors = self._threaded_overlap(2, [(1, 2, 3), (3, 4, 5)])
1126
 
        for e in errors:
1127
 
            sys.stderr.write('Failure: %s\n' % e)
1128
 
        self.assert_(len(errors) == 0)
 
1110
        assert not errors
1129
1111
 
1130
1112
    @testing.crashes('mssql', 'FIXME: unknown')
1131
1113
    @testing.fails_on('mysql', 'No support for NOWAIT')
1138
1120
 
1139
1121
        errors = self._threaded_overlap(2, [(1, 2, 3), (3, 4, 5)],
1140
1122
                update_style='nowait')
1141
 
        self.assert_(len(errors) != 0)
 
1123
        assert errors
1142
1124
 
1143
1125
class IsolationLevelTest(fixtures.TestBase):
1144
1126
    __requires__ = ('isolation_level', 'ad_hoc_engines')