from test.lib.testing import eq_, assert_raises, assert_raises_message import datetime from sqlalchemy.schema import CreateSequence, DropSequence from sqlalchemy.sql import select, text, literal_column import sqlalchemy as sa from test.lib import testing, engines from sqlalchemy import MetaData, Integer, String, ForeignKey, Boolean, exc,\ Sequence, func, literal from sqlalchemy.types import TypeDecorator from test.lib.schema import Table, Column from test.lib.testing import eq_ from test.sql import _base class DefaultTest(testing.TestBase): @classmethod def setup_class(cls): global t, f, f2, ts, currenttime, metadata, default_generator db = testing.db metadata = MetaData(db) default_generator = {'x':50} def mydefault(): default_generator['x'] += 1 return default_generator['x'] def myupdate_with_ctx(ctx): conn = ctx.connection return conn.execute(sa.select([sa.text('13')])).scalar() def mydefault_using_connection(ctx): conn = ctx.connection try: return conn.execute(sa.select([sa.text('12')])).scalar() finally: # ensure a "close()" on this connection does nothing, # since its a "branched" connection conn.close() use_function_defaults = testing.against('postgresql', 'mssql', 'maxdb') is_oracle = testing.against('oracle') # select "count(1)" returns different results on different DBs also # correct for "current_date" compatible as column default, value # differences currenttime = func.current_date(type_=sa.Date, bind=db) if is_oracle: ts = db.scalar(sa.select([func.trunc(func.sysdate(), sa.literal_column("'DAY'"), type_=sa.Date).label('today')])) assert isinstance(ts, datetime.date) and not isinstance(ts, datetime.datetime) f = sa.select([func.length('abcdef')], bind=db).scalar() f2 = sa.select([func.length('abcdefghijk')], bind=db).scalar() # TODO: engine propigation across nested functions not working currenttime = func.trunc(currenttime, sa.literal_column("'DAY'"), bind=db, type_=sa.Date) def1 = currenttime def2 = func.trunc(sa.text("sysdate"), sa.literal_column("'DAY'"), type_=sa.Date) deftype = sa.Date elif use_function_defaults: f = sa.select([func.length('abcdef')], bind=db).scalar() f2 = sa.select([func.length('abcdefghijk')], bind=db).scalar() def1 = currenttime deftype = sa.Date if testing.against('maxdb'): def2 = sa.text("curdate") elif testing.against('mssql'): def2 = sa.text("getdate()") else: def2 = sa.text("current_date") ts = db.func.current_date().scalar() else: f = len('abcdef') f2 = len('abcdefghijk') def1 = def2 = "3" ts = 3 deftype = Integer t = Table('default_test1', metadata, # python function Column('col1', Integer, primary_key=True, default=mydefault), # python literal Column('col2', String(20), default="imthedefault", onupdate="im the update"), # preexecute expression Column('col3', Integer, default=func.length('abcdef'), onupdate=func.length('abcdefghijk')), # SQL-side default from sql expression Column('col4', deftype, server_default=def1), # SQL-side default from literal expression Column('col5', deftype, server_default=def2), # preexecute + update timestamp Column('col6', sa.Date, default=currenttime, onupdate=currenttime), Column('boolcol1', sa.Boolean, default=True), Column('boolcol2', sa.Boolean, default=False), # python function which uses ExecutionContext Column('col7', Integer, default=mydefault_using_connection, onupdate=myupdate_with_ctx), # python builtin Column('col8', sa.Date, default=datetime.date.today, onupdate=datetime.date.today), # combo Column('col9', String(20), default='py', server_default='ddl')) t.create() @classmethod def teardown_class(cls): t.drop() def teardown(self): default_generator['x'] = 50 t.delete().execute() def test_bad_arg_signature(self): ex_msg = \ "ColumnDefault Python function takes zero or one positional arguments" def fn1(x, y): pass def fn2(x, y, z=3): pass class fn3(object): def __init__(self, x, y): pass class FN4(object): def __call__(self, x, y): pass fn4 = FN4() for fn in fn1, fn2, fn3, fn4: assert_raises_message(sa.exc.ArgumentError, ex_msg, sa.ColumnDefault, fn) def test_arg_signature(self): def fn1(): pass def fn2(): pass def fn3(x=1): pass def fn4(x=1, y=2, z=3): pass fn5 = list class fn6(object): def __init__(self, x): pass class fn6(object): def __init__(self, x, y=3): pass class FN7(object): def __call__(self, x): pass fn7 = FN7() class FN8(object): def __call__(self, x, y=3): pass fn8 = FN8() for fn in fn1, fn2, fn3, fn4, fn5, fn6, fn7, fn8: c = sa.ColumnDefault(fn) @testing.fails_on('firebird', 'Data type unknown') def test_standalone(self): c = testing.db.engine.contextual_connect() x = c.execute(t.c.col1.default) y = t.c.col2.default.execute() z = c.execute(t.c.col3.default) assert 50 <= x <= 57 eq_(y, 'imthedefault') eq_(z, f) eq_(f2, 11) def test_py_vs_server_default_detection(self): def has_(name, *wanted): slots = ['default', 'onupdate', 'server_default', 'server_onupdate'] col = tbl.c[name] for slot in wanted: slots.remove(slot) assert getattr(col, slot) is not None, getattr(col, slot) for slot in slots: assert getattr(col, slot) is None, getattr(col, slot) tbl = t has_('col1', 'default') has_('col2', 'default', 'onupdate') has_('col3', 'default', 'onupdate') has_('col4', 'server_default') has_('col5', 'server_default') has_('col6', 'default', 'onupdate') has_('boolcol1', 'default') has_('boolcol2', 'default') has_('col7', 'default', 'onupdate') has_('col8', 'default', 'onupdate') has_('col9', 'default', 'server_default') ColumnDefault, DefaultClause = sa.ColumnDefault, sa.DefaultClause t2 = Table('t2', MetaData(), Column('col1', Integer, Sequence('foo')), Column('col2', Integer, default=Sequence('foo'), server_default='y'), Column('col3', Integer, Sequence('foo'), server_default='x'), Column('col4', Integer, ColumnDefault('x'), DefaultClause('y')), Column('col4', Integer, ColumnDefault('x'), DefaultClause('y'), DefaultClause('y', for_update=True)), Column('col5', Integer, ColumnDefault('x'), DefaultClause('y'), onupdate='z'), Column('col6', Integer, ColumnDefault('x'), server_default='y', onupdate='z'), Column('col7', Integer, default='x', server_default='y', onupdate='z'), Column('col8', Integer, server_onupdate='u', default='x', server_default='y', onupdate='z')) tbl = t2 has_('col1', 'default') has_('col2', 'default', 'server_default') has_('col3', 'default', 'server_default') has_('col4', 'default', 'server_default', 'server_onupdate') has_('col5', 'default', 'server_default', 'onupdate') has_('col6', 'default', 'server_default', 'onupdate') has_('col7', 'default', 'server_default', 'onupdate') has_('col8', 'default', 'server_default', 'onupdate', 'server_onupdate') @testing.fails_on('firebird', 'Data type unknown') def test_insert(self): r = t.insert().execute() assert r.lastrow_has_defaults() eq_(set(r.context.postfetch_cols), set([t.c.col3, t.c.col5, t.c.col4, t.c.col6])) r = t.insert(inline=True).execute() assert r.lastrow_has_defaults() eq_(set(r.context.postfetch_cols), set([t.c.col3, t.c.col5, t.c.col4, t.c.col6])) t.insert().execute() ctexec = sa.select([currenttime.label('now')], bind=testing.db).scalar() l = t.select().order_by(t.c.col1).execute() today = datetime.date.today() eq_(l.fetchall(), [ (x, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today, 'py') for x in range(51, 54)]) t.insert().execute(col9=None) assert r.lastrow_has_defaults() eq_(set(r.context.postfetch_cols), set([t.c.col3, t.c.col5, t.c.col4, t.c.col6])) eq_(t.select(t.c.col1==54).execute().fetchall(), [(54, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today, None)]) @testing.fails_on('firebird', 'Data type unknown') def test_insertmany(self): # MySQL-Python 1.2.2 breaks functions in execute_many :( if (testing.against('mysql+mysqldb') and testing.db.dialect.dbapi.version_info[:3] == (1, 2, 2)): return r = t.insert().execute({}, {}, {}) ctexec = currenttime.scalar() l = t.select().execute() today = datetime.date.today() eq_(l.fetchall(), [(51, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today, 'py'), (52, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today, 'py'), (53, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today, 'py')]) def test_missing_many_param(self): assert_raises_message(exc.StatementError, "A value is required for bind parameter 'col7', in parameter group 1", t.insert().execute, {'col4':7, 'col7':12, 'col8':19}, {'col4':7, 'col8':19}, {'col4':7, 'col7':12, 'col8':19}, ) def test_insert_values(self): t.insert(values={'col3':50}).execute() l = t.select().execute() eq_(50, l.first()['col3']) @testing.fails_on('firebird', 'Data type unknown') def test_updatemany(self): # MySQL-Python 1.2.2 breaks functions in execute_many :( if (testing.against('mysql+mysqldb') and testing.db.dialect.dbapi.version_info[:3] == (1, 2, 2)): return t.insert().execute({}, {}, {}) t.update(t.c.col1==sa.bindparam('pkval')).execute( {'pkval':51,'col7':None, 'col8':None, 'boolcol1':False}) t.update(t.c.col1==sa.bindparam('pkval')).execute( {'pkval':51,}, {'pkval':52,}, {'pkval':53,}) l = t.select().execute() ctexec = currenttime.scalar() today = datetime.date.today() eq_(l.fetchall(), [(51, 'im the update', f2, ts, ts, ctexec, False, False, 13, today, 'py'), (52, 'im the update', f2, ts, ts, ctexec, True, False, 13, today, 'py'), (53, 'im the update', f2, ts, ts, ctexec, True, False, 13, today, 'py')]) @testing.fails_on('firebird', 'Data type unknown') def test_update(self): r = t.insert().execute() pk = r.inserted_primary_key[0] t.update(t.c.col1==pk).execute(col4=None, col5=None) ctexec = currenttime.scalar() l = t.select(t.c.col1==pk).execute() l = l.first() eq_(l, (pk, 'im the update', f2, None, None, ctexec, True, False, 13, datetime.date.today(), 'py')) eq_(11, f2) @testing.fails_on('firebird', 'Data type unknown') def test_update_values(self): r = t.insert().execute() pk = r.inserted_primary_key[0] t.update(t.c.col1==pk, values={'col3': 55}).execute() l = t.select(t.c.col1==pk).execute() l = l.first() eq_(55, l['col3']) class PKDefaultTest(_base.TablesTest): __requires__ = ('subqueries',) @classmethod def define_tables(cls, metadata): t2 = Table('t2', metadata, Column('nextid', Integer)) Table('t1', metadata, Column('id', Integer, primary_key=True, default=sa.select([func.max(t2.c.nextid)]).as_scalar()), Column('data', String(30))) @testing.requires.returning def test_with_implicit_returning(self): self._test(True) def test_regular(self): self._test(False) @testing.resolve_artifact_names def _test(self, returning): if not returning and not testing.db.dialect.implicit_returning: engine = testing.db else: engine = engines.testing_engine(options={'implicit_returning':returning}) engine.execute(t2.insert(), nextid=1) r = engine.execute(t1.insert(), data='hi') eq_([1], r.inserted_primary_key) engine.execute(t2.insert(), nextid=2) r = engine.execute(t1.insert(), data='there') eq_([2], r.inserted_primary_key) class PKIncrementTest(_base.TablesTest): run_define_tables = 'each' @classmethod def define_tables(cls, metadata): Table("aitable", metadata, Column('id', Integer, Sequence('ai_id_seq', optional=True), primary_key=True), Column('int1', Integer), Column('str1', String(20))) # TODO: add coverage for increment on a secondary column in a key @testing.fails_on('firebird', 'Data type unknown') @testing.resolve_artifact_names def _test_autoincrement(self, bind): ids = set() rs = bind.execute(aitable.insert(), int1=1) last = rs.inserted_primary_key[0] self.assert_(last) self.assert_(last not in ids) ids.add(last) rs = bind.execute(aitable.insert(), str1='row 2') last = rs.inserted_primary_key[0] self.assert_(last) self.assert_(last not in ids) ids.add(last) rs = bind.execute(aitable.insert(), int1=3, str1='row 3') last = rs.inserted_primary_key[0] self.assert_(last) self.assert_(last not in ids) ids.add(last) rs = bind.execute(aitable.insert(values={'int1':func.length('four')})) last = rs.inserted_primary_key[0] self.assert_(last) self.assert_(last not in ids) ids.add(last) eq_(ids, set([1,2,3,4])) eq_(list(bind.execute(aitable.select().order_by(aitable.c.id))), [(1, 1, None), (2, None, 'row 2'), (3, 3, 'row 3'), (4, 4, None)]) @testing.resolve_artifact_names def test_autoincrement_autocommit(self): self._test_autoincrement(testing.db) @testing.resolve_artifact_names def test_autoincrement_transaction(self): con = testing.db.connect() tx = con.begin() try: try: self._test_autoincrement(con) except: try: tx.rollback() except: pass raise else: tx.commit() finally: con.close() class EmptyInsertTest(testing.TestBase): @testing.exclude('sqlite', '<', (3, 3, 8), 'no empty insert support') @testing.fails_on('oracle', 'FIXME: unknown') def test_empty_insert(self): metadata = MetaData(testing.db) t1 = Table('t1', metadata, Column('is_true', Boolean, server_default=('1'))) metadata.create_all() try: result = t1.insert().execute() eq_(1, select([func.count(text('*'))], from_obj=t1).scalar()) eq_(True, t1.select().scalar()) finally: metadata.drop_all() class AutoIncrementTest(_base.TablesTest): __requires__ = ('identity',) run_define_tables = 'each' @classmethod def define_tables(cls, metadata): """Each test manipulates self.metadata individually.""" @testing.exclude('sqlite', '<', (3, 4), 'no database support') def test_autoincrement_single_col(self): single = Table('single', self.metadata, Column('id', Integer, primary_key=True)) single.create() r = single.insert().execute() id_ = r.inserted_primary_key[0] eq_(id_, 1) eq_(1, sa.select([func.count(sa.text('*'))], from_obj=single).scalar()) def test_autoincrement_fk(self): nodes = Table('nodes', self.metadata, Column('id', Integer, primary_key=True), Column('parent_id', Integer, ForeignKey('nodes.id')), Column('data', String(30))) nodes.create() r = nodes.insert().execute(data='foo') id_ = r.inserted_primary_key[0] nodes.insert().execute(data='bar', parent_id=id_) @testing.fails_on('sqlite', 'FIXME: unknown') def test_non_autoincrement(self): # sqlite INT primary keys can be non-unique! (only for ints) nonai = Table("nonaitest", self.metadata, Column('id', Integer, autoincrement=False, primary_key=True), Column('data', String(20))) nonai.create() try: # postgresql + mysql strict will fail on first row, # mysql in legacy mode fails on second row nonai.insert().execute(data='row 1') nonai.insert().execute(data='row 2') assert False except sa.exc.DBAPIError, e: assert True nonai.insert().execute(id=1, data='row 1') class SequenceTest(testing.TestBase, testing.AssertsCompiledSQL): @classmethod @testing.requires.sequences def setup_class(cls): global cartitems, sometable, metadata metadata = MetaData(testing.db) cartitems = Table("cartitems", metadata, Column("cart_id", Integer, Sequence('cart_id_seq'), primary_key=True), Column("description", String(40)), Column("createdate", sa.DateTime()) ) sometable = Table( 'Manager', metadata, Column('obj_id', Integer, Sequence('obj_id_seq'), ), Column('name', String(128)), Column('id', Integer, Sequence('Manager_id_seq', optional=True), primary_key=True), ) metadata.create_all() def test_compile(self): self.assert_compile( CreateSequence(Sequence('foo_seq')), "CREATE SEQUENCE foo_seq", use_default_dialect=True, ) self.assert_compile( CreateSequence(Sequence('foo_seq', start=5)), "CREATE SEQUENCE foo_seq START WITH 5", use_default_dialect=True, ) self.assert_compile( CreateSequence(Sequence('foo_seq', increment=2)), "CREATE SEQUENCE foo_seq INCREMENT BY 2", use_default_dialect=True, ) self.assert_compile( CreateSequence(Sequence('foo_seq', increment=2, start=5)), "CREATE SEQUENCE foo_seq INCREMENT BY 2 START WITH 5", use_default_dialect=True, ) self.assert_compile( DropSequence(Sequence('foo_seq')), "DROP SEQUENCE foo_seq", use_default_dialect=True, ) @testing.fails_on('firebird', 'no FB support for start/increment') @testing.requires.sequences def test_start_increment(self): for seq in ( Sequence('foo_seq'), Sequence('foo_seq', start=8), Sequence('foo_seq', increment=5)): seq.create(testing.db) try: values = [ testing.db.execute(seq) for i in range(3) ] start = seq.start or 1 inc = seq.increment or 1 assert values == list(xrange(start, start + inc * 3, inc)) finally: seq.drop(testing.db) @testing.requires.sequences def test_seq_nonpk(self): """test sequences fire off as defaults on non-pk columns""" engine = engines.testing_engine(options={'implicit_returning':False}) result = engine.execute(sometable.insert(), name="somename") assert set(result.postfetch_cols()) == set([sometable.c.obj_id]) result = engine.execute(sometable.insert(), name="someother") assert set(result.postfetch_cols()) == set([sometable.c.obj_id]) sometable.insert().execute( {'name':'name3'}, {'name':'name4'}) eq_(sometable.select().order_by(sometable.c.id).execute().fetchall(), [(1, "somename", 1), (2, "someother", 2), (3, "name3", 3), (4, "name4", 4)]) @testing.requires.sequences def test_sequence(self): cartitems.insert().execute(description='hi') cartitems.insert().execute(description='there') r = cartitems.insert().execute(description='lala') assert r.inserted_primary_key and r.inserted_primary_key[0] is not None id_ = r.inserted_primary_key[0] eq_(1, sa.select([func.count(cartitems.c.cart_id)], sa.and_(cartitems.c.description == 'lala', cartitems.c.cart_id == id_)).scalar()) cartitems.select().execute().fetchall() @testing.fails_on('maxdb', 'FIXME: unknown') # maxdb db-api seems to double-execute NEXTVAL internally somewhere, # throwing off the numbers for these tests... @testing.requires.sequences def test_implicit_sequence_exec(self): s = Sequence("my_sequence", metadata=MetaData(testing.db)) s.create() try: x = s.execute() eq_(x, 1) finally: s.drop() @testing.fails_on('maxdb', 'FIXME: unknown') @testing.requires.sequences def teststandalone_explicit(self): s = Sequence("my_sequence") s.create(bind=testing.db) try: x = s.execute(testing.db) eq_(x, 1) finally: s.drop(testing.db) @testing.requires.sequences def test_checkfirst(self): s = Sequence("my_sequence") s.create(testing.db, checkfirst=False) s.create(testing.db, checkfirst=True) s.drop(testing.db, checkfirst=False) s.drop(testing.db, checkfirst=True) @testing.fails_on('maxdb', 'FIXME: unknown') @testing.requires.sequences def teststandalone2(self): x = cartitems.c.cart_id.default.execute() self.assert_(1 <= x <= 4) @classmethod @testing.requires.sequences def teardown_class(cls): metadata.drop_all() class SpecialTypePKTest(testing.TestBase): """test process_result_value in conjunction with primary key columns. Also tests that "autoincrement" checks are against column.type._type_affinity, rather than the class of "type" itself. """ @classmethod def setup_class(cls): class MyInteger(TypeDecorator): impl = Integer def process_bind_param(self, value, dialect): if value is None: return None return int(value[4:]) def process_result_value(self, value, dialect): if value is None: return None return "INT_%d" % value cls.MyInteger = MyInteger @testing.provide_metadata def _run_test(self, *arg, **kw): implicit_returning = kw.pop('implicit_returning', True) kw['primary_key'] = True if kw.get('autoincrement', True): kw['test_needs_autoincrement'] = True t = Table('x', metadata, Column('y', self.MyInteger, *arg, **kw), Column('data', Integer), implicit_returning=implicit_returning ) t.create() r = t.insert().values(data=5).execute() # we don't pre-fetch 'server_default'. if 'server_default' in kw and (not testing.db.dialect.implicit_returning or not implicit_returning): eq_(r.inserted_primary_key, [None]) else: eq_(r.inserted_primary_key, ['INT_1']) r.close() eq_( t.select().execute().first(), ('INT_1', 5) ) def test_plain(self): # among other things, tests that autoincrement # is enabled. self._run_test() def test_literal_default_label(self): self._run_test(default=literal("INT_1", type_=self.MyInteger).label('foo')) def test_literal_default_no_label(self): self._run_test(default=literal("INT_1", type_=self.MyInteger)) def test_sequence(self): self._run_test(Sequence('foo_seq')) def test_server_default(self): self._run_test(server_default='1',) def test_server_default_no_autoincrement(self): self._run_test(server_default='1', autoincrement=False) def test_clause(self): stmt = select([literal("INT_1", type_=self.MyInteger)]).as_scalar() self._run_test(default=stmt) @testing.requires.returning def test_no_implicit_returning(self): self._run_test(implicit_returning=False) @testing.requires.returning def test_server_default_no_implicit_returning(self): self._run_test(server_default='1', autoincrement=False) class ServerDefaultsOnPKTest(testing.TestBase): @testing.provide_metadata def test_string_default_none_on_insert(self): """Test that without implicit returning, we return None for a string server default. That is, we don't want to attempt to pre-execute "server_default" generically - the user should use a Python side-default for a case like this. Testing that all backends do the same thing here. """ t = Table('x', metadata, Column('y', String(10), server_default='key_one', primary_key=True), Column('data', String(10)), implicit_returning=False ) metadata.create_all() r = t.insert().execute(data='data') eq_(r.inserted_primary_key, [None]) eq_( t.select().execute().fetchall(), [('key_one', 'data')] ) @testing.requires.returning @testing.provide_metadata def test_string_default_on_insert_with_returning(self): """With implicit_returning, we get a string PK default back no problem.""" t = Table('x', metadata, Column('y', String(10), server_default='key_one', primary_key=True), Column('data', String(10)) ) metadata.create_all() r = t.insert().execute(data='data') eq_(r.inserted_primary_key, ['key_one']) eq_( t.select().execute().fetchall(), [('key_one', 'data')] ) @testing.provide_metadata def test_int_default_none_on_insert(self): t = Table('x', metadata, Column('y', Integer, server_default='5', primary_key=True), Column('data', String(10)), implicit_returning=False ) assert t._autoincrement_column is None metadata.create_all() r = t.insert().execute(data='data') eq_(r.inserted_primary_key, [None]) if testing.against('sqlite'): eq_( t.select().execute().fetchall(), [(1, 'data')] ) else: eq_( t.select().execute().fetchall(), [(5, 'data')] ) @testing.provide_metadata def test_autoincrement_reflected_from_server_default(self): t = Table('x', metadata, Column('y', Integer, server_default='5', primary_key=True), Column('data', String(10)), implicit_returning=False ) assert t._autoincrement_column is None metadata.create_all() m2 = MetaData(metadata.bind) t2 = Table('x', m2, autoload=True, implicit_returning=False) assert t2._autoincrement_column is None @testing.provide_metadata def test_int_default_none_on_insert_reflected(self): t = Table('x', metadata, Column('y', Integer, server_default='5', primary_key=True), Column('data', String(10)), implicit_returning=False ) metadata.create_all() m2 = MetaData(metadata.bind) t2 = Table('x', m2, autoload=True, implicit_returning=False) r = t2.insert().execute(data='data') eq_(r.inserted_primary_key, [None]) if testing.against('sqlite'): eq_( t2.select().execute().fetchall(), [(1, 'data')] ) else: eq_( t2.select().execute().fetchall(), [(5, 'data')] ) @testing.requires.returning @testing.provide_metadata def test_int_default_on_insert_with_returning(self): t = Table('x', metadata, Column('y', Integer, server_default='5', primary_key=True), Column('data', String(10)) ) metadata.create_all() r = t.insert().execute(data='data') eq_(r.inserted_primary_key, [5]) eq_( t.select().execute().fetchall(), [(5, 'data')] )