class TLTransactionTest(testbase.PersistTest):
def setUpAll(self):
global users, metadata, tlengine
- tlengine = create_engine(testbase.db_uri, strategy='threadlocal')
+ tlengine = create_engine(testbase.db_uri, strategy='threadlocal', **testbase.db_opts)
metadata = MetaData()
users = Table('query_users', metadata,
Column('user_id', INT, primary_key = True),
s.clear()
assert s.query(Address).selectone().address_id == a.address_id
assert s.query(User).selectfirst() is None
+
+ def test_fetchid(self):
+ # this is necessary to ensure the test fails on old versions of mssql
+ if hasattr(autoseq.columns['autoseq_id'], 'sequence'):
+ del autoseq.columns['autoseq_id'].sequence
+
+ mapper(Autoseq, autoseq)
+ s = create_session()
+ u = Autoseq()
+ s.save(u)
+ s.flush()
+ assert u.autoseq_id is not None
+ s.clear()
+
+
class OrphanDeletionTest(AssertMixin):
# Column('foo', Boolean, default=True)
)
+autoseq = Table('autoseq', metadata,
+ Column('autoseq_id', Integer, primary_key = True),
+ Column('name', String)
+)
+
def create():
metadata.create_all()
def drop():
def __repr__(self):
return "Keyword: %s/%s" % (repr(getattr(self, 'keyword_id', None)),repr(self.name))
+class Autoseq(object):
+ def __init__(self):
+ self.autoseq_id = None
user_result = [{'user_id' : 7}, {'user_id' : 8}, {'user_id' : 9}]
db = None
metadata = None
db_uri = None
+db_opts = {}
echo = True
# redefine sys.stdout so all those print statements go to the echo func
def parse_argv():
# we are using the unittest main runner, so we are just popping out the
# arguments we need instead of using our own getopt type of thing
- global db, db_uri, metadata
+ global db, db_uri, db_opts, metadata
DBTYPE = 'sqlite'
PROXY = False
if options.dburi:
db_uri = param = options.dburi
+ DBTYPE = db_uri[:db_uri.index(':')]
elif options.db:
DBTYPE = param = options.db
-
- opts = {}
+ if DBTYPE == 'mssql':
+ db_opts['auto_identity_insert'] = True
+
if (None == db_uri):
if DBTYPE == 'sqlite':
db_uri = 'sqlite:///:memory:'
db_uri = 'oracle://scott:tiger@127.0.0.1:1521'
elif DBTYPE == 'oracle8':
db_uri = 'oracle://scott:tiger@127.0.0.1:1521'
- opts = {'use_ansi':False, 'auto_setinputsizes':True}
+ db_opts = {'use_ansi':False, 'auto_setinputsizes':True}
elif DBTYPE == 'mssql':
db_uri = 'mssql://scott:tiger@SQUAWK\\SQLEXPRESS/test'
elif DBTYPE == 'firebird':
with_coverage = options.coverage
if options.enginestrategy is not None:
- opts['strategy'] = options.enginestrategy
+ db_opts['strategy'] = options.enginestrategy
if options.mockpool:
- db = engine.create_engine(db_uri, poolclass=pool.AssertionPool, **opts)
+ db = engine.create_engine(db_uri, poolclass=pool.AssertionPool, **db_opts)
else:
- db = engine.create_engine(db_uri, **opts)
+ db = engine.create_engine(db_uri, **db_opts)
db = EngineAssert(db)
import logging