← Back to branch summary

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

« back to all changes in this revision

Viewing changes to test/engine/test_transaction.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_transaction.py
1148
1148
            return 'SERIALIZABLE'
1149
1149
        elif testing.against('postgresql'):
1150
1150
            return 'READ COMMITTED'
 
1151
        elif testing.against('mysql'):
 
1152
            return "REPEATABLE READ"
1151
1153
        else:
1152
1154
            assert False, "default isolation level not known"
1153
1155
 
1156
1158
            return 'READ UNCOMMITTED'
1157
1159
        elif testing.against('postgresql'):
1158
1160
            return 'SERIALIZABLE'
 
1161
        elif testing.against('mysql'):
 
1162
            return "SERIALIZABLE"
1159
1163
        else:
1160
1164
            assert False, "non default isolation level not known"
1161
1165
 
1162
1166
    def test_engine_param_stays(self):
1163
1167
 
1164
1168
        eng = testing_engine()
1165
 
        isolation_level = eng.dialect.get_isolation_level(eng.connect().connection)
 
1169
        isolation_level = eng.dialect.get_isolation_level(
 
1170
                                eng.connect().connection)
1166
1171
        level = self._non_default_isolation_level()
1167
1172
 
1168
1173
        ne_(isolation_level, level)
1169
1174
 
1170
1175
        eng = testing_engine(options=dict(isolation_level=level))
1171
1176
        eq_(
1172
 
            eng.dialect.get_isolation_level(eng.connect().connection),
 
1177
            eng.dialect.get_isolation_level(
 
1178
                                eng.connect().connection),
1173
1179
            level
1174
1180
        )
1175
1181
 
1190
1196
 
1191
1197
    def test_default_level(self):
1192
1198
        eng = testing_engine(options=dict())
1193
 
        isolation_level = eng.dialect.get_isolation_level(eng.connect().connection)
 
1199
        isolation_level = eng.dialect.get_isolation_level(
 
1200
                                        eng.connect().connection)
1194
1201
        eq_(isolation_level, self._default_isolation_level())
1195
1202
 
1196
1203
    def test_reset_level(self):
1197
1204
        eng = testing_engine(options=dict())
1198
1205
        conn = eng.connect()
1199
 
        eq_(eng.dialect.get_isolation_level(conn.connection), self._default_isolation_level())
 
1206
        eq_(
 
1207
            eng.dialect.get_isolation_level(conn.connection), 
 
1208
            self._default_isolation_level()
 
1209
        )
1200
1210
 
1201
 
        eng.dialect.set_isolation_level(conn.connection, self._non_default_isolation_level())
1202
 
        eq_(eng.dialect.get_isolation_level(conn.connection), self._non_default_isolation_level())
 
1211
        eng.dialect.set_isolation_level(
 
1212
                conn.connection, self._non_default_isolation_level()
 
1213
            )
 
1214
        eq_(
 
1215
            eng.dialect.get_isolation_level(conn.connection), 
 
1216
            self._non_default_isolation_level()
 
1217
        )
1203
1218
 
1204
1219
        eng.dialect.reset_isolation_level(conn.connection)
1205
 
        eq_(eng.dialect.get_isolation_level(conn.connection), self._default_isolation_level())
 
1220
        eq_(
 
1221
            eng.dialect.get_isolation_level(conn.connection), 
 
1222
            self._default_isolation_level()
 
1223
        )
1206
1224
 
1207
1225
        conn.close()
1208
1226
 
1209
1227
    def test_reset_level_with_setting(self):
1210
 
        eng = testing_engine(options=dict(isolation_level=self._non_default_isolation_level()))
 
1228
        eng = testing_engine(options=dict(
 
1229
                            isolation_level=
 
1230
                                self._non_default_isolation_level()))
1211
1231
        conn = eng.connect()
1212
 
        eq_(eng.dialect.get_isolation_level(conn.connection), self._non_default_isolation_level())
1213
 
 
1214
 
        eng.dialect.set_isolation_level(conn.connection, self._default_isolation_level())
1215
 
        eq_(eng.dialect.get_isolation_level(conn.connection), self._default_isolation_level())
1216
 
 
 
1232
        eq_(eng.dialect.get_isolation_level(conn.connection),
 
1233
            self._non_default_isolation_level())
 
1234
        eng.dialect.set_isolation_level(conn.connection,
 
1235
                self._default_isolation_level())
 
1236
        eq_(eng.dialect.get_isolation_level(conn.connection),
 
1237
            self._default_isolation_level())
1217
1238
        eng.dialect.reset_isolation_level(conn.connection)
1218
 
        eq_(eng.dialect.get_isolation_level(conn.connection), self._non_default_isolation_level())
1219
 
 
 
1239
        eq_(eng.dialect.get_isolation_level(conn.connection),
 
1240
            self._non_default_isolation_level())
1220
1241
        conn.close()
1221
1242
 
1222
1243
    def test_invalid_level(self):
1225
1246
            exc.ArgumentError, 
1226
1247
                "Invalid value '%s' for isolation_level. "
1227
1248
                "Valid isolation levels for %s are %s" % 
1228
 
                ("FOO", eng.dialect.name, ", ".join(eng.dialect._isolation_lookup)),
 
1249
                ("FOO", eng.dialect.name, 
 
1250
                ", ".join(eng.dialect._isolation_lookup)),
1229
1251
            eng.connect)
1230
1252
 
1231
1253
    def test_per_connection(self):
1232
1254
        from sqlalchemy.pool import QueuePool
1233
 
        eng = testing_engine(options=dict(poolclass=QueuePool, pool_size=2, max_overflow=0))
 
1255
        eng = testing_engine(options=dict(
 
1256
                                poolclass=QueuePool, 
 
1257
                                pool_size=2, max_overflow=0))
1234
1258
 
1235
1259
        c1 = eng.connect()
1236
 
        c1 = c1.execution_options(isolation_level=self._non_default_isolation_level())
1237
 
 
 
1260
        c1 = c1.execution_options(
 
1261
                    isolation_level=self._non_default_isolation_level()
 
1262
                )
1238
1263
        c2 = eng.connect()
1239
 
        eq_(eng.dialect.get_isolation_level(c1.connection), self._non_default_isolation_level())
1240
 
        eq_(eng.dialect.get_isolation_level(c2.connection), self._default_isolation_level())
1241
 
 
 
1264
        eq_(
 
1265
            eng.dialect.get_isolation_level(c1.connection),
 
1266
            self._non_default_isolation_level()
 
1267
        )
 
1268
        eq_(
 
1269
            eng.dialect.get_isolation_level(c2.connection),
 
1270
            self._default_isolation_level()
 
1271
        )
1242
1272
        c1.close()
1243
1273
        c2.close()
1244
1274
        c3 = eng.connect()
1245
 
        eq_(eng.dialect.get_isolation_level(c3.connection), self._default_isolation_level())
1246
 
 
 
1275
        eq_(
 
1276
            eng.dialect.get_isolation_level(c3.connection),
 
1277
            self._default_isolation_level()
 
1278
        )
1247
1279
        c4 = eng.connect()
1248
 
        eq_(eng.dialect.get_isolation_level(c4.connection), self._default_isolation_level())
 
1280
        eq_(
 
1281
            eng.dialect.get_isolation_level(c4.connection),
 
1282
            self._default_isolation_level()
 
1283
        )
1249
1284
 
1250
1285
        c3.close()
1251
1286
        c4.close()
1257
1292
            r"on Connection.execution_options\(\), or "
1258
1293
            r"per-engine using the isolation_level "
1259
1294
            r"argument to create_engine\(\).",
1260
 
            select([1]).execution_options, isolation_level=self._non_default_isolation_level()
 
1295
            select([1]).execution_options, 
 
1296
                    isolation_level=self._non_default_isolation_level()
1261
1297
        )
1262
1298
 
1263
1299
 
1269
1305
            r"To set engine-wide isolation level, "
1270
1306
            r"use the isolation_level argument to create_engine\(\).",
1271
1307
            create_engine,
1272
 
            testing.db.url, execution_options={'isolation_level':self._non_default_isolation_level}
 
1308
            testing.db.url, 
 
1309
                execution_options={'isolation_level':
 
1310
                            self._non_default_isolation_level}
1273
1311
        )