diff options
Diffstat (limited to 'test/orm/test_query.py')
| -rw-r--r-- | test/orm/test_query.py | 80 |
1 files changed, 60 insertions, 20 deletions
diff --git a/test/orm/test_query.py b/test/orm/test_query.py index db7e53a0e..a7818fa57 100644 --- a/test/orm/test_query.py +++ b/test/orm/test_query.py @@ -138,15 +138,31 @@ class GetTest(QueryTest): u2 = s.query(User).get(7) assert u is not u2 - def test_get_composite_pk(self): - s = create_session() + def test_get_composite_pk_no_result(self): + s = Session() assert s.query(CompositePk).get((100,100)) is None + + def test_get_composite_pk_result(self): + s = Session() one_two = s.query(CompositePk).get((1,2)) assert one_two.i == 1 assert one_two.j == 2 assert one_two.k == 3 + + def test_get_too_few_params(self): + s = Session() + q = s.query(CompositePk) + assert_raises(sa_exc.InvalidRequestError, q.get, 7) + + def test_get_too_few_params_tuple(self): + s = Session() q = s.query(CompositePk) - assert_raises(sa_exc.InvalidRequestError, q.get, 7) + assert_raises(sa_exc.InvalidRequestError, q.get, (7,)) + + def test_get_too_many_params(self): + s = Session() + q = s.query(CompositePk) + assert_raises(sa_exc.InvalidRequestError, q.get, (7, 10, 100)) def test_get_null_pk(self): """test that a mapping which can have None in a @@ -214,8 +230,9 @@ class GetTest(QueryTest): @testing.requires.unicode_connections def test_unicode(self): - """test that Query.get properly sets up the type for the bind parameter. using unicode would normally fail - on postgresql, mysql and oracle unless it is converted to an encoded string""" + """test that Query.get properly sets up the type for the bind + parameter. using unicode would normally fail on postgresql, mysql and + oracle unless it is converted to an encoded string""" metadata = MetaData(engines.utf8_engine()) table = Table('unicode_data', metadata, @@ -2825,34 +2842,51 @@ class InstancesTest(QueryTest, AssertsCompiledSQL): def test_from_alias(self): - query = users.select(users.c.id==7).union(users.select(users.c.id>7)).alias('ulist').outerjoin(addresses).select(use_labels=True,order_by=['ulist.id', addresses.c.id]) + query = users.select(users.c.id==7).\ + union(users.select(users.c.id>7)).\ + alias('ulist').\ + outerjoin(addresses).\ + select(use_labels=True, + order_by=['ulist.id', addresses.c.id]) sess =create_session() q = sess.query(User) def go(): - l = list(q.options(contains_alias('ulist'), contains_eager('addresses')).instances(query.execute())) + l = list(q.options(contains_alias('ulist'), + contains_eager('addresses')).\ + instances(query.execute())) assert self.static.user_address_result == l self.assert_sql_count(testing.db, go, 1) sess.expunge_all() def go(): - l = q.options(contains_alias('ulist'), contains_eager('addresses')).from_statement(query).all() + l = q.options(contains_alias('ulist'), + contains_eager('addresses')).\ + from_statement(query).all() assert self.static.user_address_result == l self.assert_sql_count(testing.db, go, 1) # better way. use select_from() def go(): - l = sess.query(User).select_from(query).options(contains_eager('addresses')).all() + l = sess.query(User).select_from(query).\ + options(contains_eager('addresses')).all() assert self.static.user_address_result == l self.assert_sql_count(testing.db, go, 1) - # same thing, but alias addresses, so that the adapter generated by select_from() is wrapped within + # same thing, but alias addresses, so that the adapter + # generated by select_from() is wrapped within # the adapter created by contains_eager() adalias = addresses.alias() - query = users.select(users.c.id==7).union(users.select(users.c.id>7)).alias('ulist').outerjoin(adalias).select(use_labels=True,order_by=['ulist.id', adalias.c.id]) + query = users.select(users.c.id==7).\ + union(users.select(users.c.id>7)).\ + alias('ulist').\ + outerjoin(adalias).\ + select(use_labels=True, + order_by=['ulist.id', adalias.c.id]) def go(): - l = sess.query(User).select_from(query).options(contains_eager('addresses', alias=adalias)).all() + l = sess.query(User).select_from(query).\ + options(contains_eager('addresses', alias=adalias)).all() assert self.static.user_address_result == l self.assert_sql_count(testing.db, go, 1) @@ -2860,13 +2894,19 @@ class InstancesTest(QueryTest, AssertsCompiledSQL): sess = create_session() # test that contains_eager suppresses the normal outer join rendering - q = sess.query(User).outerjoin(User.addresses).options(contains_eager(User.addresses)).order_by(User.id) - self.assert_compile(q.with_labels().statement, - "SELECT addresses.id AS addresses_id, addresses.user_id AS addresses_user_id, "\ - "addresses.email_address AS addresses_email_address, users.id AS users_id, "\ - "users.name AS users_name FROM users LEFT OUTER JOIN addresses "\ - "ON users.id = addresses.user_id ORDER BY users.id" - , dialect=default.DefaultDialect()) + q = sess.query(User).outerjoin(User.addresses).\ + options(contains_eager(User.addresses)).\ + order_by(User.id, addresses.c.id) + self.assert_compile(q.with_labels().statement, + 'SELECT addresses.id AS addresses_id, ' + 'addresses.user_id AS addresses_user_id, ' + 'addresses.email_address AS ' + 'addresses_email_address, users.id AS ' + 'users_id, users.name AS users_name FROM ' + 'users LEFT OUTER JOIN addresses ON ' + 'users.id = addresses.user_id ORDER BY ' + 'users.id, addresses.id', + dialect=default.DefaultDialect()) def go(): assert self.static.user_address_result == q.all() @@ -2874,7 +2914,7 @@ class InstancesTest(QueryTest, AssertsCompiledSQL): sess.expunge_all() adalias = addresses.alias() - q = sess.query(User).select_from(users.outerjoin(adalias)).options(contains_eager(User.addresses, alias=adalias)) + q = sess.query(User).select_from(users.outerjoin(adalias)).options(contains_eager(User.addresses, alias=adalias)).order_by(User.id, adalias.c.id) def go(): eq_(self.static.user_address_result, q.order_by(User.id).all()) self.assert_sql_count(testing.db, go, 1) |