diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/engine/test_reflection.py | 42 | ||||
| -rw-r--r-- | test/engine/test_transaction.py | 1 | ||||
| -rw-r--r-- | test/ext/test_declarative.py | 47 | ||||
| -rw-r--r-- | test/orm/test_collection.py | 23 | ||||
| -rw-r--r-- | test/orm/test_dynamic.py | 14 | ||||
| -rw-r--r-- | test/orm/test_eager_relations.py | 29 | ||||
| -rw-r--r-- | test/orm/test_query.py | 16 | ||||
| -rw-r--r-- | test/orm/test_subquery_relations.py | 25 | ||||
| -rw-r--r-- | test/sql/test_case_statement.py | 22 | ||||
| -rw-r--r-- | test/sql/test_query.py | 39 | ||||
| -rw-r--r-- | test/sql/test_selectable.py | 30 |
11 files changed, 259 insertions, 29 deletions
diff --git a/test/engine/test_reflection.py b/test/engine/test_reflection.py index 1cbc9ab59..a82f1ec52 100644 --- a/test/engine/test_reflection.py +++ b/test/engine/test_reflection.py @@ -326,7 +326,44 @@ class ReflectionTest(TestBase, ComparesTables): assert len(a4.constraints) == 2 finally: meta.drop_all() + + @testing.provide_metadata + def test_override_composite_fk(self): + """Test double-remove of composite foreign key, when replaced.""" + + a = Table('a', + metadata, + Column('x', sa.Integer, primary_key=True), + Column('y', sa.Integer, primary_key=True), + ) + + b = Table('b', + metadata, + Column('x', sa.Integer, primary_key=True), + Column('y', sa.Integer, primary_key=True), + sa.ForeignKeyConstraint(['x', 'y'], ['a.x', 'a.y']) + ) + + metadata.create_all() + meta2 = MetaData() + + c1 = Column('x', sa.Integer, primary_key=True) + c2 = Column('y', sa.Integer, primary_key=True) + f1 = sa.ForeignKeyConstraint(['x', 'y'], ['a.x', 'a.y']) + b1 = Table('b', + meta2, c1, c2, f1, + autoload=True, + autoload_with=testing.db + ) + + assert b1.c.x is c1 + assert b1.c.y is c2 + assert f1 in b1.constraints + assert len(b1.constraints) == 2 + + + def test_override_keys(self): """test that columns can be overridden with a 'key', and that ForeignKey targeting during reflection still works.""" @@ -510,8 +547,9 @@ class ReflectionTest(TestBase, ComparesTables): ) assert_raises_message(sa.exc.InvalidRequestError, - "Could not find table 'pkgs' with which " - "to generate a foreign key", + "Foreign key assocated with column 'slots.pkg_id' " + "could not find table 'pkgs' with which to generate " + "a foreign key to target column 'pkg_id'", metadata.create_all) def test_composite_pks(self): diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py index bc0985bec..e7e2fe1b8 100644 --- a/test/engine/test_transaction.py +++ b/test/engine/test_transaction.py @@ -868,6 +868,7 @@ class TLTransactionTest(TestBase): assert r2.connection.closed assert tlengine.closed + @testing.crashes('oracle+cx_oracle', 'intermittent failures on the buildbot') def test_dispose(self): eng = create_engine(testing.db.url, strategy='threadlocal') result = eng.execute(select([1])) diff --git a/test/ext/test_declarative.py b/test/ext/test_declarative.py index 4da826d38..71e31233b 100644 --- a/test/ext/test_declarative.py +++ b/test/ext/test_declarative.py @@ -690,6 +690,40 @@ class DeclarativeTest(DeclarativeTestBase): eq_(sess.query(User).all(), [User(name='u1', address_count=2, addresses=[Address(email='one'), Address(email='two')])]) + def test_useless_classproperty(self): + class Address(Base, ComparableEntity): + + __tablename__ = 'addresses' + id = Column('id', Integer, primary_key=True, + test_needs_autoincrement=True) + email = Column('email', String(50)) + user_id = Column('user_id', Integer, ForeignKey('users.id')) + + class User(Base, ComparableEntity): + + __tablename__ = 'users' + id = Column('id', Integer, primary_key=True, + test_needs_autoincrement=True) + name = Column('name', String(50)) + addresses = relationship('Address', backref='user') + + @classproperty + def address_count(cls): + # this doesn't really gain us anything. but if + # one is used, lets have it function as expected... + return sa.orm.column_property(sa.select([sa.func.count(Address.id)]). + where(Address.user_id == cls.id)) + + Base.metadata.create_all() + u1 = User(name='u1', addresses=[Address(email='one'), + Address(email='two')]) + sess = create_session() + sess.add(u1) + sess.flush() + sess.expunge_all() + eq_(sess.query(User).all(), [User(name='u1', address_count=2, + addresses=[Address(email='one'), Address(email='two')])]) + def test_column(self): class User(Base, ComparableEntity): @@ -2585,17 +2619,18 @@ class DeclarativeMixinTest(DeclarativeTestBase): def test_table_in_model_and_different_named_column_in_mixin(self): class ColumnMixin: - tada = Column(Integer) - + + def go(): class Model(Base, ColumnMixin): - __table__ = Table('foo', Base.metadata, Column('data', - Integer), Column('id', Integer, - primary_key=True)) - + __table__ = Table('foo', Base.metadata, + Column('data',Integer), + Column('id', Integer,primary_key=True)) + foo = relationship("Dest") + assert_raises_message(sa.exc.ArgumentError, "Can't add additional column 'tada' when " "specifying __table__", go) diff --git a/test/orm/test_collection.py b/test/orm/test_collection.py index 7dcc5c56f..405829f74 100644 --- a/test/orm/test_collection.py +++ b/test/orm/test_collection.py @@ -7,12 +7,12 @@ from sqlalchemy.orm.collections import collection import sqlalchemy as sa from sqlalchemy.test import testing -from sqlalchemy import Integer, String, ForeignKey +from sqlalchemy import Integer, String, ForeignKey, text from sqlalchemy.test.schema import Table, Column from sqlalchemy import util, exc as sa_exc from sqlalchemy.orm import create_session, mapper, relationship, attributes from test.orm import _base -from sqlalchemy.test.testing import eq_, assert_raises +from sqlalchemy.test.testing import eq_, assert_raises, assert_raises_message class Canary(sa.orm.interfaces.AttributeExtension): def __init__(self): @@ -1561,6 +1561,25 @@ class DictHelpersTest(_base.MappedTest): eq_(Bar.foos.property.collection_class().keyfunc(Foo(id=3)), 3) eq_(Bar.foos2.property.collection_class().keyfunc(Foo(id=3, bar_id=12)), (3, 12)) + + @testing.resolve_artifact_names + def test_column_mapped_assertions(self): + assert_raises_message( + sa_exc.ArgumentError, + "Column-based expression object expected; got: 'a'", + collections.column_mapped_collection, "a", + ) + assert_raises_message( + sa_exc.ArgumentError, + "Column-based expression object expected; got", + collections.column_mapped_collection, text("a"), + ) + assert_raises_message( + sa_exc.ArgumentError, + "Column-based expression object expected; got", + collections.column_mapped_collection, text("a"), + ) + @testing.resolve_artifact_names def test_column_mapped_collection(self): diff --git a/test/orm/test_dynamic.py b/test/orm/test_dynamic.py index b2fac6005..5d822fa3d 100644 --- a/test/orm/test_dynamic.py +++ b/test/orm/test_dynamic.py @@ -6,7 +6,7 @@ from sqlalchemy import Integer, String, ForeignKey, desc, select, func from sqlalchemy.test.schema import Table, Column from sqlalchemy.orm import mapper, relationship, create_session, Query, attributes from sqlalchemy.orm.dynamic import AppenderMixin -from sqlalchemy.test.testing import eq_, AssertsCompiledSQL +from sqlalchemy.test.testing import eq_, AssertsCompiledSQL, assert_raises_message from sqlalchemy.util import function_named from test.orm import _base, _fixtures @@ -123,6 +123,18 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL): self.assert_sql_count(testing.db, go, 2) @testing.resolve_artifact_names + def test_no_populate(self): + mapper(User, users, properties={ + 'addresses':dynamic_loader(mapper(Address, addresses)) + }) + u1 = User() + assert_raises_message( + NotImplementedError, + "Dynamic attributes don't support collection population.", + attributes.set_committed_value, u1, 'addresses', [] + ) + + @testing.resolve_artifact_names def test_m2m(self): mapper(Order, orders, properties={ 'items':relationship(Item, secondary=order_items, lazy="dynamic", diff --git a/test/orm/test_eager_relations.py b/test/orm/test_eager_relations.py index de397f47b..b96014a57 100644 --- a/test/orm/test_eager_relations.py +++ b/test/orm/test_eager_relations.py @@ -1,13 +1,17 @@ -"""basic tests of eager loaded attributes""" +"""tests of joined-eager loaded attributes""" from sqlalchemy.test.testing import eq_, is_, is_not_ import sqlalchemy as sa from sqlalchemy.test import testing -from sqlalchemy.orm import joinedload, deferred, undefer, joinedload_all, backref -from sqlalchemy import Integer, String, Date, ForeignKey, and_, select, func +from sqlalchemy.orm import joinedload, deferred, undefer, \ + joinedload_all, backref +from sqlalchemy import Integer, String, Date, ForeignKey, and_, select, \ + func from sqlalchemy.test.schema import Table, Column -from sqlalchemy.orm import mapper, relationship, create_session, lazyload, aliased -from sqlalchemy.test.testing import eq_, assert_raises +from sqlalchemy.orm import mapper, relationship, create_session, \ + lazyload, aliased +from sqlalchemy.test.testing import eq_, assert_raises, \ + assert_raises_message from sqlalchemy.test.assertsql import CompiledSQL from test.orm import _base, _fixtures from sqlalchemy.util import OrderedDict as odict @@ -283,7 +287,20 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): ) self.assert_sql_count(testing.db, go, count) - + @testing.resolve_artifact_names + def test_disable_dynamic(self): + """test no joined option on a dynamic.""" + + mapper(User, users, properties={ + 'addresses':relationship(Address, lazy="dynamic") + }) + mapper(Address, addresses) + sess = create_session() + assert_raises_message( + sa.exc.InvalidRequestError, + "User.addresses' does not support object population - eager loading cannot be applied.", + sess.query(User).options(joinedload(User.addresses)).first, + ) @testing.resolve_artifact_names def test_many_to_many(self): diff --git a/test/orm/test_query.py b/test/orm/test_query.py index ebe72b565..cc2f046cd 100644 --- a/test/orm/test_query.py +++ b/test/orm/test_query.py @@ -1132,6 +1132,22 @@ class SetOpsTest(QueryTest, AssertsCompiledSQL): (User(id=10, name=u'chuck'), u'y') ] ) + + c1, c2 = column('c1'), column('c2') + q1 = s.query(User, c1.label('foo'), c1.label('bar')) + q2 = s.query(User, c1.label('foo'), c2.label('bar')) + q3 = q1.union(q2) + self.assert_compile( + q3, + "SELECT anon_1.users_id AS anon_1_users_id, " + "anon_1.users_name AS anon_1_users_name, " + "anon_1.foo AS anon_1_foo, anon_1.bar AS anon_1_bar " + "FROM (SELECT users.id AS users_id, users.name AS users_name, " + "c1 AS foo, c1 AS bar FROM users UNION SELECT users.id AS " + "users_id, users.name AS users_name, c1 AS foo, c2 AS bar " + "FROM users) AS anon_1", + use_default_dialect=True + ) @testing.fails_on('mysql', "mysql doesn't support intersect") def test_intersect(self): diff --git a/test/orm/test_subquery_relations.py b/test/orm/test_subquery_relations.py index 827630c3d..71f87a726 100644 --- a/test/orm/test_subquery_relations.py +++ b/test/orm/test_subquery_relations.py @@ -3,10 +3,10 @@ from sqlalchemy.test import testing from sqlalchemy.test.schema import Table, Column from sqlalchemy import Integer, String, ForeignKey, bindparam from sqlalchemy.orm import backref, subqueryload, subqueryload_all, \ - mapper, relationship, clear_mappers,\ - create_session, lazyload, aliased, joinedload,\ - deferred, undefer -from sqlalchemy.test.testing import eq_, assert_raises + mapper, relationship, clear_mappers, create_session, lazyload, \ + aliased, joinedload, deferred, undefer +from sqlalchemy.test.testing import eq_, assert_raises, \ + assert_raises_message from sqlalchemy.test.assertsql import CompiledSQL from test.orm import _base, _fixtures import sqlalchemy as sa @@ -80,6 +80,23 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): self.assert_sql_count(testing.db, go, 2) + @testing.resolve_artifact_names + def test_disable_dynamic(self): + """test no subquery option on a dynamic.""" + + mapper(User, users, properties={ + 'addresses':relationship(Address, lazy="dynamic") + }) + mapper(Address, addresses) + sess = create_session() + + # previously this would not raise, but would emit + # the query needlessly and put the result nowhere. + assert_raises_message( + sa.exc.InvalidRequestError, + "User.addresses' does not support object population - eager loading cannot be applied.", + sess.query(User).options(subqueryload(User.addresses)).first, + ) @testing.resolve_artifact_names def test_many_to_many(self): diff --git a/test/sql/test_case_statement.py b/test/sql/test_case_statement.py index 3f3abe7e1..645822fa7 100644 --- a/test/sql/test_case_statement.py +++ b/test/sql/test_case_statement.py @@ -1,4 +1,4 @@ -from sqlalchemy.test.testing import assert_raises, assert_raises_message +from sqlalchemy.test.testing import assert_raises, assert_raises_message, eq_ import sys from sqlalchemy import * from sqlalchemy.test import * @@ -99,7 +99,25 @@ class CaseTest(TestBase, AssertsCompiledSQL): self.assert_compile(case([("x", "y")], value=t.c.col1), "CASE test.col1 WHEN :param_1 THEN :param_2 END") self.assert_compile(case([(t.c.col1==7, "y")], else_="z"), "CASE WHEN (test.col1 = :col1_1) THEN :param_1 ELSE :param_2 END") - + + def test_text_doesnt_explode(self): + + for s in [ + select([case([(info_table.c.info == 'pk_4_data', + text("'yes'"))], else_=text("'no'" + ))]).order_by(info_table.c.info), + + select([case([(info_table.c.info == 'pk_4_data', + literal_column("'yes'"))], else_=literal_column("'no'" + ))]).order_by(info_table.c.info), + + ]: + eq_(s.execute().fetchall(), [ + (u'no', ), (u'no', ), (u'no', ), (u'yes', ), + (u'no', ), (u'no', ), + ]) + + @testing.fails_on('firebird', 'FIXME: unknown') @testing.fails_on('maxdb', 'FIXME: unknown') diff --git a/test/sql/test_query.py b/test/sql/test_query.py index e8f9d118b..0a496906d 100644 --- a/test/sql/test_query.py +++ b/test/sql/test_query.py @@ -16,15 +16,19 @@ class QueryTest(TestBase): users = Table('query_users', metadata, Column('user_id', INT, primary_key=True, test_needs_autoincrement=True), Column('user_name', VARCHAR(20)), + test_needs_acid=True ) addresses = Table('query_addresses', metadata, Column('address_id', Integer, primary_key=True, test_needs_autoincrement=True), Column('user_id', Integer, ForeignKey('query_users.user_id')), - Column('address', String(30))) + Column('address', String(30)), + test_needs_acid=True + ) users2 = Table('u2', metadata, Column('user_id', INT, primary_key = True), Column('user_name', VARCHAR(20)), + test_needs_acid=True ) metadata.create_all() @@ -615,6 +619,39 @@ class QueryTest(TestBase): eq_(r[users.c.user_name], 'jack') eq_(r.user_name, 'jack') + def test_graceful_fetch_on_non_rows(self): + """test that calling fetchone() etc. on a result that doesn't + return rows fails gracefully. + + """ + + # these proxies don't work with no cursor.description present. + # so they don't apply to this test at the moment. + # base.FullyBufferedResultProxy, + # base.BufferedRowResultProxy, + # base.BufferedColumnResultProxy + + conn = testing.db.connect() + for meth in ('fetchone', 'fetchall', 'first', 'scalar', 'fetchmany'): + trans = conn.begin() + result = conn.execute(users.insert(), user_id=1) + assert_raises_message( + exc.ResourceClosedError, + "This result object does not return rows. " + "It has been closed automatically.", + getattr(result, meth), + ) + trans.rollback() + + def test_fetchone_til_end(self): + result = testing.db.execute("select * from query_users") + eq_(result.fetchone(), None) + assert_raises_message( + exc.ResourceClosedError, + "This result object is closed.", + result.fetchone + ) + def test_result_case_sensitivity(self): """test name normalization for result sets.""" diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py index 062ed5f1b..5bebbe05f 100644 --- a/test/sql/test_selectable.py +++ b/test/sql/test_selectable.py @@ -28,20 +28,40 @@ table2 = Table('table2', metadata, class SelectableTest(TestBase, AssertsExecutionResults): - def test_distance_on_labels(self): - + def test_indirect_correspondence_on_labels(self): + # this test depends upon 'distance' to + # get the right result + # same column three times s = select([table1.c.col1.label('c2'), table1.c.col1, table1.c.col1.label('c1')]) - # didnt do this yet...col.label().make_proxy() has same - # "distance" as col.make_proxy() so far assert - # s.corresponding_column(table1.c.col1) is s.c.col1 + # this tests the same thing as + # test_direct_correspondence_on_labels below - + # that the presence of label() affects the 'distance' + assert s.corresponding_column(table1.c.col1) is s.c.col1 assert s.corresponding_column(s.c.col1) is s.c.col1 assert s.corresponding_column(s.c.c1) is s.c.c1 + def test_direct_correspondence_on_labels(self): + # this test depends on labels being part + # of the proxy set to get the right result + + l1, l2 = table1.c.col1.label('foo'), table1.c.col1.label('bar') + sel = select([l1, l2]) + + sel2 = sel.alias() + assert sel2.corresponding_column(l1) is sel2.c.foo + assert sel2.corresponding_column(l2) is sel2.c.bar + + sel2 = select([table1.c.col1.label('foo'), table1.c.col2.label('bar')]) + + sel3 = sel.union(sel2).alias() + assert sel3.corresponding_column(l1) is sel3.c.foo + assert sel3.corresponding_column(l2) is sel3.c.bar + def test_distance_on_aliases(self): a1 = table1.alias('a1') for s in (select([a1, table1], use_labels=True), |