diff options
| author | Mike Bayer <mike_mp@zzzcomputing.com> | 2007-03-28 07:19:14 +0000 |
|---|---|---|
| committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2007-03-28 07:19:14 +0000 |
| commit | e0b638a704f7f0abf88d1a80a95cf052954e048c (patch) | |
| tree | 6b8717e491820322c48b5b271ec6b192fa8a86cd /test/sql | |
| parent | ccbcbda43e74a1d09d50aa2f8212b3cb9adafd23 (diff) | |
| download | sqlalchemy-e0b638a704f7f0abf88d1a80a95cf052954e048c.tar.gz | |
- column label and bind param "truncation" also generate
deterministic names now, based on their ordering within the
full statement being compiled. this means the same statement
will produce the same string across application restarts and
allowing DB query plan caching to work better.
- cleanup to sql.ClauseParameters since it was just falling
apart, API made more explicit
- many unit test tweaks to adjust for bind params not being
"pre" truncated, changes to ClauseParameters
Diffstat (limited to 'test/sql')
| -rw-r--r-- | test/sql/labels.py | 41 | ||||
| -rw-r--r-- | test/sql/select.py | 48 |
2 files changed, 56 insertions, 33 deletions
diff --git a/test/sql/labels.py b/test/sql/labels.py index 0302fee78..a2e899ed6 100644 --- a/test/sql/labels.py +++ b/test/sql/labels.py @@ -2,27 +2,34 @@ import testbase from sqlalchemy import * +# TODO: either create a mock dialect with named paramstyle and a short identifier length, +# or find a way to just use sqlite dialect and make those changes + class LongLabelsTest(testbase.PersistTest): def setUpAll(self): global metadata, table1 metadata = MetaData(engine=testbase.db) table1 = Table("some_large_named_table", metadata, - Column("this_is_the_primary_key_column", Integer, primary_key=True), + Column("this_is_the_primarykey_column", Integer, primary_key=True), Column("this_is_the_data_column", String(30)) ) metadata.create_all() - table1.insert().execute(**{"this_is_the_primary_key_column":1, "this_is_the_data_column":"data1"}) - table1.insert().execute(**{"this_is_the_primary_key_column":2, "this_is_the_data_column":"data2"}) - table1.insert().execute(**{"this_is_the_primary_key_column":3, "this_is_the_data_column":"data3"}) - table1.insert().execute(**{"this_is_the_primary_key_column":4, "this_is_the_data_column":"data4"}) + def tearDown(self): + table1.delete().execute() + def tearDownAll(self): metadata.drop_all() def test_result(self): + table1.insert().execute(**{"this_is_the_primarykey_column":1, "this_is_the_data_column":"data1"}) + table1.insert().execute(**{"this_is_the_primarykey_column":2, "this_is_the_data_column":"data2"}) + table1.insert().execute(**{"this_is_the_primarykey_column":3, "this_is_the_data_column":"data3"}) + table1.insert().execute(**{"this_is_the_primarykey_column":4, "this_is_the_data_column":"data4"}) + r = table1.select(use_labels=True).execute() result = [] for row in r: - result.append((row[table1.c.this_is_the_primary_key_column], row[table1.c.this_is_the_data_column])) + result.append((row[table1.c.this_is_the_primarykey_column], row[table1.c.this_is_the_data_column])) assert result == [ (1, "data1"), (2, "data2"), @@ -31,14 +38,30 @@ class LongLabelsTest(testbase.PersistTest): ] def test_colbinds(self): - r = table1.select(table1.c.this_is_the_primary_key_column == 4).execute() + table1.insert().execute(**{"this_is_the_primarykey_column":1, "this_is_the_data_column":"data1"}) + table1.insert().execute(**{"this_is_the_primarykey_column":2, "this_is_the_data_column":"data2"}) + table1.insert().execute(**{"this_is_the_primarykey_column":3, "this_is_the_data_column":"data3"}) + table1.insert().execute(**{"this_is_the_primarykey_column":4, "this_is_the_data_column":"data4"}) + + r = table1.select(table1.c.this_is_the_primarykey_column == 4).execute() assert r.fetchall() == [(4, "data4")] r = table1.select(or_( - table1.c.this_is_the_primary_key_column == 4, - table1.c.this_is_the_primary_key_column == 2 + table1.c.this_is_the_primarykey_column == 4, + table1.c.this_is_the_primarykey_column == 2 )).execute() assert r.fetchall() == [(2, "data2"), (4, "data4")] + + def test_insert_no_pk(self): + table1.insert().execute(**{"this_is_the_data_column":"data1"}) + table1.insert().execute(**{"this_is_the_data_column":"data2"}) + table1.insert().execute(**{"this_is_the_data_column":"data3"}) + table1.insert().execute(**{"this_is_the_data_column":"data4"}) + + def test_subquery(self): + q = table1.select(table1.c.this_is_the_primarykey_column == 4, use_labels=True) + x = select([q]) + print str(x) if __name__ == '__main__': testbase.main()
\ No newline at end of file diff --git a/test/sql/select.py b/test/sql/select.py index 5fcf88fd1..f71d5366b 100644 --- a/test/sql/select.py +++ b/test/sql/select.py @@ -59,9 +59,9 @@ class SQLTest(PersistTest): self.assert_(cc == result, "\n'" + cc + "'\n does not match \n'" + result + "'") if checkparams is not None: if isinstance(checkparams, list): - self.assert_(c.get_params().values() == checkparams, "params dont match ") + self.assert_(c.get_params().get_raw_list() == checkparams, "params dont match ") else: - self.assert_(c.get_params() == checkparams, "params dont match" + repr(c.get_params())) + self.assert_(c.get_params().get_original_dict() == checkparams, "params dont match" + repr(c.get_params())) class SelectTest(SQLTest): def testtableselect(self): @@ -201,7 +201,7 @@ sq.myothertable_othername AS sq_myothertable_othername FROM (" + sqstring + ") A order_by = ['dist', places.c.nm] ) - self.runtest(q,"SELECT places.id, places.nm, zips.zipcode, latlondist((SELECT zips.latitude AS latitude FROM zips WHERE zips.zipcode = :zips_zipco_1), (SELECT zips.longitude AS longitude FROM zips WHERE zips.zipcode = :zips_zipco_2)) AS dist FROM places, zips WHERE zips.zipcode = :zips_zipcode ORDER BY dist, places.nm") + self.runtest(q,"SELECT places.id, places.nm, zips.zipcode, latlondist((SELECT zips.latitude AS latitude FROM zips WHERE zips.zipcode = :zips_zipcode_1), (SELECT zips.longitude AS longitude FROM zips WHERE zips.zipcode = :zips_zipcode_2)) AS dist FROM places, zips WHERE zips.zipcode = :zips_zipcode ORDER BY dist, places.nm") zalias = zips.alias('main_zip') qlat = select([zips.c.latitude], zips.c.zipcode == zalias.c.zipcode, scalar=True) @@ -224,8 +224,8 @@ sq.myothertable_othername AS sq_myothertable_othername FROM (" + sqstring + ") A or_(table2.c.othername=='asdf', table2.c.othername == 'foo', table2.c.otherid == 9), "sysdate() = today()", )), - "SELECT mytable.myid, mytable.name, mytable.description FROM mytable, myothertable WHERE mytable.myid = :mytable_myid AND (myothertable.othername = :myothertable_othername OR myothertable.othername = :myothertable_otherna_1 OR myothertable.otherid = :myothertable_otherid) AND sysdate() = today()", - checkparams = {'myothertable_othername': 'asdf', 'myothertable_otherna_1':'foo', 'myothertable_otherid': 9, 'mytable_myid': 12} + "SELECT mytable.myid, mytable.name, mytable.description FROM mytable, myothertable WHERE mytable.myid = :mytable_myid AND (myothertable.othername = :myothertable_othername OR myothertable.othername = :myothertable_othername_1 OR myothertable.otherid = :myothertable_otherid) AND sysdate() = today()", + checkparams = {'myothertable_othername': 'asdf', 'myothertable_othername_1':'foo', 'myothertable_otherid': 9, 'mytable_myid': 12} ) def testoperators(self): @@ -235,7 +235,7 @@ sq.myothertable_othername AS sq_myothertable_othername FROM (" + sqstring + ") A ) self.runtest( - literal("a") + literal("b") * literal("c"), ":literal + (:liter_1 * :liter_2)" + literal("a") + literal("b") * literal("c"), ":literal + (:literal_1 * :literal_2)" ) # test the op() function, also that its results are further usable in expressions @@ -255,7 +255,7 @@ sq.myothertable_othername AS sq_myothertable_othername FROM (" + sqstring + ") A def testmultiparam(self): self.runtest( select(["*"], or_(table1.c.myid == 12, table1.c.myid=='asdf', table1.c.myid == 'foo')), - "SELECT * FROM mytable WHERE mytable.myid = :mytable_myid OR mytable.myid = :mytable_my_1 OR mytable.myid = :mytable_my_2" + "SELECT * FROM mytable WHERE mytable.myid = :mytable_myid OR mytable.myid = :mytable_myid_1 OR mytable.myid = :mytable_myid_2" ) def testorderby(self): @@ -419,7 +419,7 @@ FROM mytable, myothertable WHERE foo.id = foofoo(lala) AND datetime(foo) = Today def testliteral(self): self.runtest(select([literal("foo") + literal("bar")], from_obj=[table1]), - "SELECT :literal + :liter_1 FROM mytable") + "SELECT :literal + :literal_1 FROM mytable") def testcalculatedcolumns(self): value_tbl = table('values', @@ -449,7 +449,7 @@ FROM mytable, myothertable WHERE foo.id = foofoo(lala) AND datetime(foo) = Today """tests the generation of functions using the func keyword""" # test an expression with a function self.runtest(func.lala(3, 4, literal("five"), table1.c.myid) * table2.c.otherid, - "lala(:lala, :la_1, :literal, mytable.myid) * myothertable.otherid") + "lala(:lala, :lala_1, :literal, mytable.myid) * myothertable.otherid") # test it in a SELECT self.runtest(select([func.count(table1.c.myid)]), @@ -471,7 +471,7 @@ FROM mytable, myothertable WHERE foo.id = foofoo(lala) AND datetime(foo) = Today """test the EXTRACT function""" self.runtest(select([extract("month", table3.c.otherstuff)]), "SELECT extract(month FROM thirdtable.otherstuff) FROM thirdtable") - self.runtest(select([extract("day", func.to_date("03/20/2005", "MM/DD/YYYY"))]), "SELECT extract(day FROM to_date(:to_date, :to_da_1))") + self.runtest(select([extract("day", func.to_date("03/20/2005", "MM/DD/YYYY"))]), "SELECT extract(day FROM to_date(:to_date, :to_date_1))") def testjoin(self): self.runtest( @@ -526,7 +526,7 @@ mytable.description FROM myothertable JOIN mytable ON mytable.myid = myothertabl self.runtest(x, "SELECT mytable.myid, mytable.name, mytable.description \ FROM mytable WHERE mytable.myid = :mytable_myid UNION \ SELECT mytable.myid, mytable.name, mytable.description \ -FROM mytable WHERE mytable.myid = :mytable_my_1 ORDER BY mytable.myid") +FROM mytable WHERE mytable.myid = :mytable_myid_1 ORDER BY mytable.myid") self.runtest( union( @@ -636,17 +636,17 @@ myothertable.othername != :myothertable_othername AND EXISTS (select yay from fo ), ( select([table1], or_(table1.c.myid==bindparam('myid', unique=True), table2.c.otherid==bindparam('myid', unique=True))), - "SELECT mytable.myid, mytable.name, mytable.description FROM mytable, myothertable WHERE mytable.myid = :myid OR myothertable.otherid = :my_1", + "SELECT mytable.myid, mytable.name, mytable.description FROM mytable, myothertable WHERE mytable.myid = :myid OR myothertable.otherid = :myid_1", "SELECT mytable.myid, mytable.name, mytable.description FROM mytable, myothertable WHERE mytable.myid = ? OR myothertable.otherid = ?", - {'myid':None, 'my_1':None}, [None, None], - {'myid':5, 'my_1': 6}, {'myid':5, 'my_1':6}, [5,6] + {'myid':None, 'myid_1':None}, [None, None], + {'myid':5, 'myid_1': 6}, {'myid':5, 'myid_1':6}, [5,6] ), ( select([table1], or_(table1.c.myid==bindparam('myid', value=7, unique=True), table2.c.otherid==bindparam('myid', value=8, unique=True))), - "SELECT mytable.myid, mytable.name, mytable.description FROM mytable, myothertable WHERE mytable.myid = :myid OR myothertable.otherid = :my_1", + "SELECT mytable.myid, mytable.name, mytable.description FROM mytable, myothertable WHERE mytable.myid = :myid OR myothertable.otherid = :myid_1", "SELECT mytable.myid, mytable.name, mytable.description FROM mytable, myothertable WHERE mytable.myid = ? OR myothertable.otherid = ?", - {'myid':7, 'my_1':8}, [7,8], - {'myid':5, 'my_1':6}, {'myid':5, 'my_1':6}, [5,6] + {'myid':7, 'myid_1':8}, [7,8], + {'myid':5, 'myid_1':6}, {'myid':5, 'myid_1':6}, [5,6] ), ][2:3]: @@ -666,18 +666,18 @@ myothertable.othername != :myothertable_othername AND EXISTS (select yay from fo except exceptions.CompileError, err: assert str(err) == "Bind parameter 'mytable_myid' conflicts with unique bind parameter of the same name" - s = select([table1], or_(table1.c.myid==7, table1.c.myid==8, table1.c.myid==bindparam('mytable_my_1'))) + s = select([table1], or_(table1.c.myid==7, table1.c.myid==8, table1.c.myid==bindparam('mytable_myid_1'))) try: str(s) assert False except exceptions.CompileError, err: - assert str(err) == "Bind parameter 'mytable_my_1' conflicts with unique bind parameter of the same name" + assert str(err) == "Bind parameter 'mytable_myid_1' conflicts with unique bind parameter of the same name" # check that the bind params sent along with a compile() call # get preserved when the params are retreived later s = select([table1], table1.c.myid == bindparam('test')) c = s.compile(parameters = {'test' : 7}) - self.assert_(c.get_params() == {'test' : 7}) + self.assert_(c.get_params().get_original_dict() == {'test' : 7}) def testbindascol(self): t = table('foo', column('id')) @@ -688,7 +688,7 @@ myothertable.othername != :myothertable_othername AND EXISTS (select yay from fo def testin(self): self.runtest(select([table1], table1.c.myid.in_(1, 2, 3)), - "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:mytable_myid, :mytable_my_1, :mytable_my_2)") + "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:mytable_myid, :mytable_myid_1, :mytable_myid_2)") self.runtest(select([table1], table1.c.myid.in_(select([table2.c.otherid]))), "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (SELECT myothertable.otherid AS otherid FROM myothertable)") @@ -752,9 +752,9 @@ myothertable.othername != :myothertable_othername AND EXISTS (select yay from fo import datetime table = Table('dt', metadata, Column('date', Date)) - self.runtest(table.select(table.c.date.between(datetime.date(2006,6,1), datetime.date(2006,6,5))), "SELECT dt.date FROM dt WHERE dt.date BETWEEN :dt_date AND :dt_da_1", checkparams={'dt_date':datetime.date(2006,6,1), 'dt_da_1':datetime.date(2006,6,5)}) + self.runtest(table.select(table.c.date.between(datetime.date(2006,6,1), datetime.date(2006,6,5))), "SELECT dt.date FROM dt WHERE dt.date BETWEEN :dt_date AND :dt_date_1", checkparams={'dt_date':datetime.date(2006,6,1), 'dt_date_1':datetime.date(2006,6,5)}) - self.runtest(table.select(sql.between(table.c.date, datetime.date(2006,6,1), datetime.date(2006,6,5))), "SELECT dt.date FROM dt WHERE dt.date BETWEEN :literal AND :liter_1", checkparams={'literal':datetime.date(2006,6,1), 'liter_1':datetime.date(2006,6,5)}) + self.runtest(table.select(sql.between(table.c.date, datetime.date(2006,6,1), datetime.date(2006,6,5))), "SELECT dt.date FROM dt WHERE dt.date BETWEEN :literal AND :literal_1", checkparams={'literal':datetime.date(2006,6,1), 'literal_1':datetime.date(2006,6,5)}) class CRUDTest(SQLTest): def testinsert(self): @@ -803,7 +803,7 @@ class CRUDTest(SQLTest): values = { table1.c.name : table1.c.name + "lala", table1.c.myid : func.do_stuff(table1.c.myid, literal('hoho')) - }), "UPDATE mytable SET myid=do_stuff(mytable.myid, :liter_2), name=mytable.name + :mytable_name WHERE mytable.myid = hoho(:hoho) AND mytable.name = ((:literal + mytable.name) + :liter_1)") + }), "UPDATE mytable SET myid=do_stuff(mytable.myid, :literal_2), name=mytable.name + :mytable_name WHERE mytable.myid = hoho(:hoho) AND mytable.name = ((:literal + mytable.name) + :literal_1)") def testcorrelatedupdate(self): # test against a straight text subquery |
