<commit_before>import testbase
from sqlalchemy import *
from sqlalchemy import exceptions, schema, util
from sqlalchemy.orm import mapper, create_session
from testlib import *
import datetime

class DefaultTest(PersistTest):

    def setUpAll(self):
        global t, f, f2, ts, currenttime, metadata, default_generator

        db = testbase.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(select([text('13')])).scalar()
        
        def mydefault_using_connection(ctx):
            conn = ctx.connection
            try:
                return conn.execute(select([text('12')])).scalar()
            finally:
                # ensure a "close()" on this connection does nothing,
                # since its a "branched" connection
                conn.close()
            
        use_function_defaults = db.engine.name == 'postgres' or db.engine.name == 'oracle'
        is_oracle = db.engine.name == '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_=Date, bind=db);
        if is_oracle:
            ts = db.func.trunc(func.sysdate(), literal_column("'DAY'")).scalar()
            f = select([func.length('abcdef')], bind=db).scalar()
            f2 = select([func.length('abcdefghijk')], bind=db).scalar()
            # TODO: engine propigation across nested functions not working
            currenttime = func.trunc(currenttime, literal_column("'DAY'"), bind=db)
            def1 = currenttime
            def2 = func.trunc(text("sysdate"), literal_column("'DAY'"))
            deftype = Date
        elif use_function_defaults:
            f = select([func.length('abcdef')], bind=db).scalar()
            f2 = select([func.length('abcdefghijk')], bind=db).scalar()
            def1 = currenttime
            def2 = text("current_date")
            deftype = Date
            ts = db.func.current_date().scalar()
        else:
            f = select([func.length('abcdef')], bind=db).scalar()
            f2 = select([func.length('abcdefghijk')], bind=db).scalar()
            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, PassiveDefault(def1)),
            
            # SQL-side default from literal expression
            Column('col5', deftype, PassiveDefault(def2)),
            
            # preexecute + update timestamp
            Column('col6', Date, default=currenttime, onupdate=currenttime),
            
            Column('boolcol1', Boolean, default=True),
            Column('boolcol2', Boolean, default=False),
            
            # python function which uses ExecutionContext
            Column('col7', Integer, default=mydefault_using_connection, onupdate=myupdate_with_ctx),
            
            # python builtin
            Column('col8', Date, default=datetime.date.today, onupdate=datetime.date.today)
        )
        t.create()

    def tearDownAll(self):
        t.drop()
    
    def tearDown(self):
        default_generator['x'] = 50
        t.delete().execute()
    
    def testargsignature(self):
        ex_msg = \
          "ColumnDefault Python function takes zero or one positional arguments"

        def fn1(x, y): pass
        def fn2(x, y, z=3): pass
        for fn in fn1, fn2:
            try:
                c = ColumnDefault(fn)
                assert False
            except exceptions.ArgumentError, e:
                assert str(e) == ex_msg

        def fn3(): pass
        def fn4(): pass
        def fn5(x=1): pass
        def fn6(x=1, y=2, z=3): pass
        fn7 = list

        for fn in fn3, fn4, fn5, fn6, fn7:
            c = ColumnDefault(fn)
            
    def teststandalone(self):
        c = testbase.db.engine.contextual_connect()
        x = c.execute(t.c.col1.default)
        y = t.c.col2.default.execute()
        z = c.execute(t.c.col3.default)
        self.assert_(50 <= x <= 57)
        self.assert_(y == 'imthedefault')
        self.assert_(z == f)
        self.assert_(f2==11)
        
    def testinsert(self):
        r = t.insert().execute()
        assert r.lastrow_has_defaults()
        assert util.Set(r.context.postfetch_cols()) == util.Set([t.c.col3, t.c.col5, t.c.col4, t.c.col6])

        r = t.insert(inline=True).execute()
        assert r.lastrow_has_defaults()
        assert util.Set(r.context.postfetch_cols()) == util.Set([t.c.col3, t.c.col5, t.c.col4, t.c.col6])
        
        t.insert().execute()
        t.insert().execute()

        ctexec = currenttime.scalar()
        l = t.select().execute()
        today = datetime.date.today()
        self.assert_(l.fetchall() == [
            (51, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today), 
            (52, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today), 
            (53, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today),
            (54, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today),
            ])

    def testinsertmany(self):
        r = t.insert().execute({}, {}, {})

        ctexec = currenttime.scalar()
        l = t.select().execute()
        today = datetime.date.today()
        self.assert_(l.fetchall() == [(51, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today), (52, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today), (53, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today)])

    def testinsertvalues(self):
        t.insert(values={'col3':50}).execute()
        l = t.select().execute()
        self.assert_(l.fetchone()['col3'] == 50)
        
    def testupdatemany(self):
        t.insert().execute({}, {}, {})

        t.update(t.c.col1==bindparam('pkval')).execute(
            {'pkval':51,'col7':None, 'col8':None, 'boolcol1':False},
        )
        
        
        t.update(t.c.col1==bindparam('pkval')).execute(
            {'pkval':51,},
            {'pkval':52,},
            {'pkval':53,},
        )<commit_msg>Avoid tickling the MySQL-python 1.2.2 executemany parsing bug on a couple tests.<commit_after>import testbase
from sqlalchemy import *
from sqlalchemy import exceptions, schema, util
from sqlalchemy.orm import mapper, create_session
from testlib import *
import datetime

class DefaultTest(PersistTest):

    def setUpAll(self):
        global t, f, f2, ts, currenttime, metadata, default_generator

        db = testbase.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(select([text('13')])).scalar()
        
        def mydefault_using_connection(ctx):
            conn = ctx.connection
            try:
                return conn.execute(select([text('12')])).scalar()
            finally:
                # ensure a "close()" on this connection does nothing,
                # since its a "branched" connection
                conn.close()
            
        use_function_defaults = db.engine.name == 'postgres' or db.engine.name == 'oracle'
        is_oracle = db.engine.name == '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_=Date, bind=db);
        if is_oracle:
            ts = db.func.trunc(func.sysdate(), literal_column("'DAY'")).scalar()
            f = select([func.length('abcdef')], bind=db).scalar()
            f2 = select([func.length('abcdefghijk')], bind=db).scalar()
            # TODO: engine propigation across nested functions not working
            currenttime = func.trunc(currenttime, literal_column("'DAY'"), bind=db)
            def1 = currenttime
            def2 = func.trunc(text("sysdate"), literal_column("'DAY'"))
            deftype = Date
        elif use_function_defaults:
            f = select([func.length('abcdef')], bind=db).scalar()
            f2 = select([func.length('abcdefghijk')], bind=db).scalar()
            def1 = currenttime
            def2 = text("current_date")
            deftype = Date
            ts = db.func.current_date().scalar()
        else:
            f = select([func.length('abcdef')], bind=db).scalar()
            f2 = select([func.length('abcdefghijk')], bind=db).scalar()
            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, PassiveDefault(def1)),
            
            # SQL-side default from literal expression
            Column('col5', deftype, PassiveDefault(def2)),
            
            # preexecute + update timestamp
            Column('col6', Date, default=currenttime, onupdate=currenttime),
            
            Column('boolcol1', Boolean, default=True),
            Column('boolcol2', Boolean, default=False),
            
            # python function which uses ExecutionContext
            Column('col7', Integer, default=mydefault_using_connection, onupdate=myupdate_with_ctx),
            
            # python builtin
            Column('col8', Date, default=datetime.date.today, onupdate=datetime.date.today)
        )
        t.create()

    def tearDownAll(self):
        t.drop()
    
    def tearDown(self):
        default_generator['x'] = 50
        t.delete().execute()
    
    def testargsignature(self):
        ex_msg = \
          "ColumnDefault Python function takes zero or one positional arguments"

        def fn1(x, y): pass
        def fn2(x, y, z=3): pass
        for fn in fn1, fn2:
            try:
                c = ColumnDefault(fn)
                assert False
            except exceptions.ArgumentError, e:
                assert str(e) == ex_msg

        def fn3(): pass
        def fn4(): pass
        def fn5(x=1): pass
        def fn6(x=1, y=2, z=3): pass
        fn7 = list

        for fn in fn3, fn4, fn5, fn6, fn7:
            c = ColumnDefault(fn)
            
    def teststandalone(self):
        c = testbase.db.engine.contextual_connect()
        x = c.execute(t.c.col1.default)
        y = t.c.col2.default.execute()
        z = c.execute(t.c.col3.default)
        self.assert_(50 <= x <= 57)
        self.assert_(y == 'imthedefault')
        self.assert_(z == f)
        self.assert_(f2==11)
        
    def testinsert(self):
        r = t.insert().execute()
        assert r.lastrow_has_defaults()
        assert util.Set(r.context.postfetch_cols()) == util.Set([t.c.col3, t.c.col5, t.c.col4, t.c.col6])

        r = t.insert(inline=True).execute()
        assert r.lastrow_has_defaults()
        assert util.Set(r.context.postfetch_cols()) == util.Set([t.c.col3, t.c.col5, t.c.col4, t.c.col6])
        
        t.insert().execute()
        t.insert().execute()

        ctexec = currenttime.scalar()
        l = t.select().execute()
        today = datetime.date.today()
        self.assert_(l.fetchall() == [
            (51, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today), 
            (52, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today), 
            (53, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today),
            (54, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today),
            ])

    def testinsertmany(self):
        # MySQL-Python 1.2.2 breaks functions in execute_many :(
        if (testbase.db.name == 'mysql' and
            testbase.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()
        self.assert_(l.fetchall() == [(51, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today), (52, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today), (53, 'imthedefault', f, ts, ts, ctexec, True, False, 12, today)])

    def testinsertvalues(self):
        t.insert(values={'col3':50}).execute()
        l = t.select().execute()
        self.assert_(l.fetchone()['col3'] == 50)
        
    def testupdatemany(self):
        # MySQL-Python 1.2.2 breaks functions in execute_many :(
        if (testbase.db.name == 'mysql' and
            testbase.db.dialect.dbapi.version_info[:3] == (1, 2, 2)):
            return

        t.insert().execute({}, {}, {})

        t.update(t.c.col1==bindparam('pkval')).execute(
            {'pkval':51,'col7':None, 'col8':None, 'boolcol1':False},
        )
        
        
        t.update(t.c.col1==bindparam('pkval')).execute(
            {'pkval':51,},
            {'pkval':52,},
            {'pkval':53,},
        )
{'fraction_non_alphanumeric': 0.13762581197801413, 'fraction_numerical': 0.01848811478335356, 'mean_word_length': 2.0129032258064514, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2601851', 'n_tokens_mistral': 4321, 'n_tokens_neox': 4128, 'n_words': 941}